diff --git a/tests/api/test_all.nim b/tests/api/test_all.nim index 108551bc2..22af4748d 100644 --- a/tests/api/test_all.nim +++ b/tests/api/test_all.nim @@ -7,4 +7,5 @@ import ./test_api_send, ./test_api_subscription, ./test_api_receive, - ./test_api_health + ./test_api_health, + ./test_messaging_rest diff --git a/tests/api/test_messaging_rest.nim b/tests/api/test_messaging_rest.nim new file mode 100644 index 000000000..8e86d912e --- /dev/null +++ b/tests/api/test_messaging_rest.nim @@ -0,0 +1,172 @@ +{.used.} + +import + std/[options, net, sequtils], + chronos, + testutils/unittests, + presto, + presto/client as presto_client, + libp2p/crypto/crypto +import brokers/broker_context +import logos_delivery +import + logos_delivery/messaging/rest_api/client as messaging_rest_client, + logos_delivery/waku/common/base64 +import tools/confutils/cli_args +import ../testlib/[wakucore, testasync] + +## Integration test for the messaging REST endpoints and their event cache. +## +## A full `LogosDelivery` node is started with REST enabled (so `start` mounts +## the messaging handlers + event listeners), then driven through the generated +## `client.nim` stubs. Event observability is exercised deterministically by +## emitting the MessagingClient events directly on the node's broker context — +## the same context the send/recv services emit on — so we do not depend on real +## network delivery. + +proc restNodeConf(): WakuNodeConf = + var conf = defaultWakuNodeConf().valueOr: + raiseAssert error + conf.mode = cli_args.WakuMode.Core + conf.listenAddress = parseIpAddress("0.0.0.0") + conf.tcpPort = Port(0) + conf.discv5UdpPort = Port(0) + conf.clusterId = some(3'u16) + conf.numShardsInNetwork = 1 + conf.reliabilityEnabled = some(true) + conf.rest = true + conf.restAddress = parseIpAddress("127.0.0.1") + conf.restPort = 0'u16 # bind to an ephemeral port + return conf + +proc restClientFor(node: LogosDelivery): RestClientRef = + let boundPort = node.waku.restServer.httpServer.address.port + newRestHttpClient(initTAddress(parseIpAddress("127.0.0.1"), boundPort)) + +const settleDelay = 200.milliseconds + ## Event emit + listener run are asyncSpawned; give them a turn before polling. + +suite "Messaging REST API": + asyncTest "subscribe / unsubscribe / send endpoints respond": + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = (await LogosDelivery.new(restNodeConf())).valueOr: + raiseAssert error + (await node.start()).isOkOr: + raiseAssert "Failed to start node: " & error + let client = restClientFor(node) + + let contentTopic = "/test/1/messaging-rest/proto" + + let subResp = await client.messagingPostSubscriptionsV1(@[contentTopic]) + check subResp.status == 200 + + let msg = MessagingMessage( + payload: base64.encode("hello rest"), + contentTopic: contentTopic, + ephemeral: none(bool), + meta: none(Base64String), + ) + let sendResp = await client.messagingPostMessagesV1(msg) + check: + sendResp.status == 200 + sendResp.data.requestId.len > 0 + + let unsubResp = await client.messagingDeleteSubscriptionsV1(@[contentTopic]) + check unsubResp.status == 200 + + (await node.stop()).isOkOr: + raiseAssert "Failed to stop node: " & error + + asyncTest "send events are grouped by requestId and evict after poll": + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = (await LogosDelivery.new(restNodeConf())).valueOr: + raiseAssert error + (await node.start()).isOkOr: + raiseAssert "Failed to start node: " & error + let client = restClientFor(node) + let brokerCtx = node.waku.brokerCtx + + let reqA = RequestId("req-A") + let reqB = RequestId("req-B") + + # reqA sees the full lifecycle; reqB only an error. + MessagePropagatedEvent.emit( + brokerCtx, MessagePropagatedEvent(requestId: reqA, messageHash: "0xaa") + ) + MessageSentEvent.emit( + brokerCtx, MessageSentEvent(requestId: reqA, messageHash: "0xaa") + ) + MessageErrorEvent.emit( + brokerCtx, + MessageErrorEvent(requestId: reqB, messageHash: "0xbb", error: "boom"), + ) + await sleepAsync(settleDelay) + + # GET by id returns only reqA and removes it. + let byIdResp = await client.messagingGetSendEventsByIdV1($reqA) + check: + byIdResp.status == 200 + byIdResp.data.requestId == $reqA + byIdResp.data.events.len == 2 + byIdResp.data.events.anyIt(it.kind == SendEventKind.Sent) + byIdResp.data.events.anyIt(it.kind == SendEventKind.Propagated) + + # Unknown / already-polled id → 404. + let missingResp = await client.messagingGetSendEventsByIdV1($reqA) + check missingResp.status == 404 + + # GET all now returns only reqB (with its error), then clears. + let allResp = await client.messagingGetSendEventsV1() + check: + allResp.status == 200 + allResp.data.len == 1 + allResp.data[0].requestId == $reqB + allResp.data[0].events.len == 1 + allResp.data[0].events[0].kind == SendEventKind.Error + allResp.data[0].events[0].error == "boom" + + let emptyResp = await client.messagingGetSendEventsV1() + check: + emptyResp.status == 200 + emptyResp.data.len == 0 + + (await node.stop()).isOkOr: + raiseAssert "Failed to stop node: " & error + + asyncTest "received messages are observable, capped, and evict after poll": + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = (await LogosDelivery.new(restNodeConf())).valueOr: + raiseAssert error + (await node.start()).isOkOr: + raiseAssert "Failed to start node: " & error + let client = restClientFor(node) + let brokerCtx = node.waku.brokerCtx + + # Emit more than the cache capacity (50); oldest must be dropped. + const total = 55 + for i in 0 ..< total: + let wm = + fakeWakuMessage(payload = "msg-" & $i, contentTopic = "/test/1/recv/proto") + MessageReceivedEvent.emit( + brokerCtx, MessageReceivedEvent(messageHash: "0x" & $i, message: wm) + ) + await sleepAsync(settleDelay) + + let resp = await client.messagingGetReceivedMessagesV1() + check: + resp.status == 200 + resp.data.len == 50 # capped at DefaultMaxReceived + # oldest (0..4) evicted, newest retained, oldest-first ordering + resp.data[0].messageHash == "0x5" + resp.data[^1].messageHash == "0x" & $(total - 1) + + let emptyResp = await client.messagingGetReceivedMessagesV1() + check: + emptyResp.status == 200 + emptyResp.data.len == 0 + + (await node.stop()).isOkOr: + raiseAssert "Failed to stop node: " & error