test(waku-filter): Subscriber ping tests (#2023)

* Move waku filter utility functions to a different file for reuse.
* Add asyncSetup and asyncTeardown templates.
* Implement waku filter subscriber ping tests.

tests/all_tests_waku.nim

@@ -47,6 +47,7 @@ import
 
 # Waku filter test suite
 import
+  ./waku_filter_v2/test_waku_client,
   ./waku_filter_v2/test_waku_filter,
   ./waku_filter_v2/test_waku_filter_protocol
 

tests/testlib/testasync.nim

@@ -0,0 +1,16 @@
+# Sourced from: vendor/nim-libp2p/tests/testutils.nim
+# Adds the ability for asyncSetup and asyncTeardown to be used in unittest2
+
+template asyncTeardown*(body: untyped): untyped =
+  teardown:
+    waitFor((
+      proc() {.async, gcsafe.} =
+        body
+    )())
+
+template asyncSetup*(body: untyped): untyped =
+  setup:
+    waitFor((
+      proc() {.async, gcsafe.} =
+        body
+    )())

tests/waku_filter_v2/client_utils.nim

@@ -0,0 +1,33 @@
+import
+  std/[options,tables],
+  testutils/unittests,
+  chronos,
+  chronicles
+
+import
+  ../../../waku/node/peer_manager,
+  ../../../waku/waku_filter_v2,
+  ../../../waku/waku_filter_v2/client,
+  ../../../waku/waku_core,
+  ../testlib/common,
+  ../testlib/wakucore
+
+proc newTestWakuFilter*(switch: Switch): Future[WakuFilter] {.async.} =
+  let
+    peerManager = PeerManager.new(switch)
+    proto = WakuFilter.new(peerManager)
+
+  await proto.start()
+  switch.mount(proto)
+
+  return proto
+
+proc newTestWakuFilterClient*(switch: Switch, messagePushHandler: MessagePushHandler): Future[WakuFilterClient] {.async.} =
+  let
+    peerManager = PeerManager.new(switch)
+    proto = WakuFilterClient.new(rng, messagePushHandler, peerManager)
+
+  await proto.start()
+  switch.mount(proto)
+
+  return proto

tests/waku_filter_v2/test_waku_client.nim

@@ -0,0 +1,101 @@
+{.used.}
+
+import
+  std/[options,tables],
+  testutils/unittests,
+  chronos,
+  chronicles,
+  libp2p/peerstore
+
+import
+  ../../../waku/node/peer_manager,
+  ../../../waku/waku_filter_v2,
+  ../../../waku/waku_filter_v2/client,
+  ../../../waku/waku_core,
+  ../testlib/common,
+  ../testlib/wakucore,
+  ../testlib/testasync,
+  ./client_utils.nim
+
+suite "Waku Filter":
+  suite "Subscriber Ping":
+    var serverSwitch {.threadvar.}: Switch
+    var clientSwitch {.threadvar.}: Switch
+    var wakuFilter {.threadvar.}: WakuFilter
+    var wakuFilterClient {.threadvar.}: WakuFilterClient
+    var serverRemotePeerInfo {.threadvar.}: RemotePeerInfo
+    var pubsubTopic {.threadvar.}: PubsubTopic
+    var contentTopics {.threadvar.}: seq[ContentTopic]
+
+    asyncSetup:
+      let
+        voidHandler: MessagePushHandler = proc(pubsubTopic: PubsubTopic, message: WakuMessage) =
+          discard
+
+      pubsubTopic = DefaultPubsubTopic
+      contentTopics = @[DefaultContentTopic]
+      serverSwitch = newStandardSwitch()
+      clientSwitch = newStandardSwitch()
+      wakuFilter = await newTestWakuFilter(serverSwitch)
+      wakuFilterClient = await newTestWakuFilterClient(clientSwitch, voidHandler)
+      
+      await allFutures(serverSwitch.start(), clientSwitch.start())
+      serverRemotePeerInfo = serverSwitch.peerInfo.toRemotePeerInfo()
+    
+    asyncTeardown:
+      await allFutures(wakuFilter.stop(), wakuFilterClient.stop(), serverSwitch.stop(), clientSwitch.stop())
+
+    asyncTest "Active Subscription Identification":
+      # Given
+      let 
+        clientPeerId = clientSwitch.peerInfo.toRemotePeerInfo().peerId
+        subscribeResponse = await wakuFilterClient.subscribe(
+          serverRemotePeerInfo, pubsubTopic, contentTopics
+        )
+      require:
+        subscribeResponse.isOk()
+        wakuFilter.subscriptions.hasKey(clientPeerId)
+
+      # When
+      let subscribedPingResponse = await wakuFilterClient.ping(serverRemotePeerInfo)
+
+      # Then
+      check:
+        subscribedPingResponse.isOk()
+        wakuFilter.subscriptions.hasKey(clientSwitch.peerInfo.toRemotePeerInfo().peerId)
+
+    asyncTest "No Active Subscription Identification":
+      # When
+      let unsubscribedPingResponse = await wakuFilterClient.ping(serverRemotePeerInfo)
+
+      # Then
+      check:
+        unsubscribedPingResponse.isErr() # Not subscribed
+        unsubscribedPingResponse.error().kind == FilterSubscribeErrorKind.NOT_FOUND
+
+    asyncTest "After Unsubscription":
+      # Given
+      let 
+        clientPeerId = clientSwitch.peerInfo.toRemotePeerInfo().peerId
+        subscribeResponse = await wakuFilterClient.subscribe(
+          serverRemotePeerInfo, pubsubTopic, contentTopics
+        )
+      
+      require:
+        subscribeResponse.isOk()
+        wakuFilter.subscriptions.hasKey(clientPeerId)
+
+      # When
+      let unsubscribeResponse = await wakuFilterClient.unsubscribe(
+        serverRemotePeerInfo, pubsubTopic, contentTopics
+      )
+      require:
+        unsubscribeResponse.isOk()
+        not wakuFilter.subscriptions.hasKey(clientPeerId)
+
+      let unsubscribedPingResponse = await wakuFilterClient.ping(serverRemotePeerInfo)
+
+      # Then
+      check:
+        unsubscribedPingResponse.isErr() # Not subscribed
+        unsubscribedPingResponse.error().kind == FilterSubscribeErrorKind.NOT_FOUND

tests/waku_filter_v2/test_waku_filter.nim

@@ -6,70 +6,18 @@ import
   chronos,
   chronicles,
   libp2p/peerstore
+
 import
   ../../../waku/node/peer_manager,
   ../../../waku/waku_filter_v2,
   ../../../waku/waku_filter_v2/client,
   ../../../waku/waku_core,
   ../testlib/common,
-  ../testlib/wakucore
+  ../testlib/wakucore,
-
+  ./client_utils.nim
-proc newTestWakuFilter(switch: Switch): Future[WakuFilter] {.async.} =
-  let
-    peerManager = PeerManager.new(switch)
-    proto = WakuFilter.new(peerManager)
-
-  await proto.start()
-  switch.mount(proto)
-
-  return proto
-
-proc newTestWakuFilterClient(switch: Switch, messagePushHandler: MessagePushHandler): Future[WakuFilterClient] {.async.} =
-  let
-    peerManager = PeerManager.new(switch)
-    proto = WakuFilterClient.new(rng, messagePushHandler, peerManager)
-
-  await proto.start()
-  switch.mount(proto)
-
-  return proto
 
 suite "Waku Filter - end to end":
 
-  asyncTest "ping":
-    # Given
-    var
-      voidHandler: MessagePushHandler = proc(pubsubTopic: PubsubTopic, message: WakuMessage) =
-        discard
-    let
-      serverSwitch = newStandardSwitch()
-      clientSwitch = newStandardSwitch()
-      wakuFilter = await newTestWakuFilter(serverSwitch)
-      wakuFilterClient = await newTestWakuFilterClient(clientSwitch, voidHandler)
-
-    # When
-    await allFutures(serverSwitch.start(), clientSwitch.start())
-    let response = await wakuFilterClient.ping(serverSwitch.peerInfo.toRemotePeerInfo())
-
-    # Then
-    check:
-      response.isErr() # Not subscribed
-      response.error().kind == FilterSubscribeErrorKind.NOT_FOUND
-
-    # When
-    let response2 = await wakuFilterClient.subscribe(serverSwitch.peerInfo.toRemotePeerInfo(), DefaultPubsubTopic, @[DefaultContentTopic])
-
-    require response2.isOk()
-
-    let response3 = await wakuFilterClient.ping(serverSwitch.peerInfo.toRemotePeerInfo())
-
-    # Then
-    check:
-      response3.isOk() # Subscribed
-
-    # Teardown
-    await allFutures(wakuFilter.stop(), wakuFilterClient.stop(), serverSwitch.stop(), clientSwitch.stop())
-
   asyncTest "simple subscribe and unsubscribe request":
     # Given
     var