Messaging rest tests

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

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