test: sharding tests refactor (#1883)

* sharding tests refactor

* small fixes

* adjust clusterID based on version

* fix typo

* fix dispatchEvent test

* sharding unit tests

* port adjustment

* update unit tests

* fix 1902

* adjust content topic tests

* adjust metdata tests

* small adjustments

* fix

* update resolveAutoshardingCluster version

* skip autosharding tests for nwaku < 0.27.0

* skip autosharding tests for nwaku < 0.27.0

packages/tests/src/lib/service_node.ts

@@ -290,7 +290,7 @@ export class ServiceNode {
     }
 
     return this.restCall<boolean>(
-      `/relay/v1/auto/message`,
+      `/relay/v1/auto/messages`,
       "POST",
       message,
       async (response) => response.status === 200

packages/tests/src/types.ts

@@ -15,6 +15,7 @@ export interface Args {
   discv5Discovery?: boolean;
   storeMessageDbUrl?: string;
   pubsubTopic?: Array<string>;
+  contentTopic?: Array<string>;
   websocketSupport?: boolean;
   tcpPort?: number;
   restPort?: number;

packages/tests/src/utils/generate_test_data.ts

@@ -1,7 +1,5 @@
 import { createDecoder, createEncoder, Decoder, Encoder } from "@waku/core";
 
-import { DOCKER_IMAGE_NAME } from "../lib/service_node";
-
 // Utility to generate test data for multiple topics tests.
 export function generateTestData(topicCount: number): {
   contentTopics: string[];
@@ -22,19 +20,3 @@ export function generateTestData(topicCount: number): {
     decoders
   };
 }
-
-// Utility to add test conditions based on nwaku/go-waku versions
-export function isNwakuAtLeast(requiredVersion: string): boolean {
-  const versionRegex = /(?:v)?(\d+\.\d+(?:\.\d+)?)/;
-  const match = DOCKER_IMAGE_NAME.match(versionRegex);
-
-  if (match) {
-    const version = match[0].substring(1); // Remove the 'v' prefix
-    return (
-      version.localeCompare(requiredVersion, undefined, { numeric: true }) >= 0
-    );
-  } else {
-    // If there is no match we assume that it's a version close to master so we return True
-    return true;
-  }
-}

packages/tests/src/utils/index.ts

@@ -6,3 +6,4 @@ export * from "./delay.js";
 export * from "./base64_utf8.js";
 export * from "./waitForConnections.js";
 export * from "./custom_mocha_hooks.js";
+export * from "./waku_versions_utils.js";

packages/tests/src/utils/waku_versions_utils.ts

@@ -0,0 +1,34 @@
+import { Logger } from "@waku/utils";
+
+import { DOCKER_IMAGE_NAME } from "../lib/service_node";
+
+const log = new Logger("test:utils");
+
+// Utility to add test conditions based on nwaku/go-waku versions
+export function isNwakuAtLeast(requiredVersion: string): boolean {
+  const versionRegex = /(?:v)?(\d+\.\d+(?:\.\d+)?)/;
+  const match = DOCKER_IMAGE_NAME.match(versionRegex);
+
+  if (match) {
+    const version = match[0].substring(1); // Remove the 'v' prefix
+    return (
+      version.localeCompare(requiredVersion, undefined, { numeric: true }) >= 0
+    );
+  } else {
+    // If there is no match we assume that it's a version close to master so we return True
+    return true;
+  }
+}
+
+// Utility to resolve autosharding cluster ID
+export function resolveAutoshardingCluster(clusterId: number): number {
+  if (isNwakuAtLeast("0.27.0")) {
+    log.info(`Using clusterID ${clusterId} for autosharding`);
+    return clusterId;
+  } else {
+    // for versions older than 0.27.0 the autosharding cluster was hardcoded to 1
+    // https://github.com/waku-org/nwaku/pull/2505
+    log.warn("Falling back to clusterID 1 for autosharding");
+    return 1;
+  }
+}

packages/tests/tests/filter/single_node/multiple_pubsub.node.spec.ts

@@ -19,8 +19,10 @@ import { expect } from "chai";
 import {
   afterEachCustom,
   beforeEachCustom,
+  isNwakuAtLeast,
   makeLogFileName,
   MessageCollector,
+  resolveAutoshardingCluster,
   ServiceNode,
   tearDownNodes
 } from "../../../src/index.js";
@@ -186,6 +188,7 @@ describe("Waku Filter V2: Multiple PubsubTopics", function () {
 describe("Waku Filter V2 (Autosharding): Multiple PubsubTopics", function () {
   // Set the timeout for all tests in this suite. Can be overwritten at test level
   this.timeout(30000);
+  const clusterId = resolveAutoshardingCluster(3);
   let waku: LightNode;
   let nwaku: ServiceNode;
   let nwaku2: ServiceNode;
@@ -196,39 +199,45 @@ describe("Waku Filter V2 (Autosharding): Multiple PubsubTopics", function () {
   const customContentTopic2 = "/myapp/1/latest/proto";
   const autoshardingPubsubTopic1 = contentTopicToPubsubTopic(
     customContentTopic1,
-    3
+    clusterId
   );
   const autoshardingPubsubTopic2 = contentTopicToPubsubTopic(
     customContentTopic2,
-    3
+    clusterId
   );
   const contentTopicInfo: ContentTopicInfo = {
-    clusterId: 3,
+    clusterId: clusterId,
     contentTopics: [customContentTopic1, customContentTopic2]
   };
   const customEncoder1 = createEncoder({
     contentTopic: customContentTopic1,
     pubsubTopicShardInfo: {
-      clusterId: 3,
+      clusterId: clusterId,
       shard: contentTopicToShardIndex(customContentTopic1)
     }
   });
   const customDecoder1 = createDecoder(customContentTopic1, {
-    clusterId: 3,
+    clusterId: clusterId,
     shard: contentTopicToShardIndex(customContentTopic1)
   });
   const customEncoder2 = createEncoder({
     contentTopic: customContentTopic2,
     pubsubTopicShardInfo: {
-      clusterId: 3,
+      clusterId: clusterId,
       shard: contentTopicToShardIndex(customContentTopic2)
     }
   });
   const customDecoder2 = createDecoder(customContentTopic2, {
-    clusterId: 3,
+    clusterId: clusterId,
     shard: contentTopicToShardIndex(customContentTopic2)
   });
 
+  before(async () => {
+    if (!isNwakuAtLeast("0.27.0")) {
+      this.ctx.skip();
+    }
+  });
+
   beforeEachCustom(this, async () => {
     [nwaku, waku] = await runNodes(
       this.ctx,
@@ -309,7 +318,8 @@ describe("Waku Filter V2 (Autosharding): Multiple PubsubTopics", function () {
       lightpush: true,
       relay: true,
       pubsubTopic: [autoshardingPubsubTopic2],
-      clusterId: 3
+      clusterId: clusterId,
+      contentTopic: [customContentTopic2]
     });
     await waku.dial(await nwaku2.getMultiaddrWithId());
     await waitForRemotePeer(waku, [Protocols.Filter, Protocols.LightPush]);

packages/tests/tests/filter/single_node/utils.ts

@@ -1,5 +1,6 @@
 import { waitForRemotePeer } from "@waku/core";
 import {
+  ContentTopicInfo,
   DefaultPubsubTopic,
   LightNode,
   ProtocolCreateOptions,
@@ -26,17 +27,25 @@ export async function runNodes(
 ): Promise<[ServiceNode, LightNode]> {
   const nwaku = new ServiceNode(makeLogFileName(context));
 
+  function isContentTopicInfo(info: ShardingParams): info is ContentTopicInfo {
+    return (info as ContentTopicInfo).contentTopics !== undefined;
+  }
+
   await nwaku.start(
     {
       filter: true,
       lightpush: true,
       relay: true,
       pubsubTopic: pubsubTopics,
-      ...(shardInfo && { clusterId: shardInfo.clusterId })
+      // Conditionally include clusterId if shardInfo exists
+      ...(shardInfo && { clusterId: shardInfo.clusterId }),
+      // Conditionally include contentTopic if shardInfo exists and clusterId is 1
+      ...(shardInfo &&
+        isContentTopicInfo(shardInfo) &&
+        shardInfo.clusterId === 1 && { contentTopic: shardInfo.contentTopics })
     },
     { retries: 3 }
   );
-
   const waku_options: ProtocolCreateOptions = {
     staticNoiseKey: NOISE_KEY_1,
     libp2p: { addresses: { listen: ["/ip4/0.0.0.0/tcp/0/ws"] } },

packages/tests/tests/getPeers.spec.ts

@@ -20,7 +20,9 @@ import Sinon from "sinon";
 import {
   afterEachCustom,
   beforeEachCustom,
+  isNwakuAtLeast,
   makeLogFileName,
+  resolveAutoshardingCluster,
   ServiceNode,
   tearDownNodes
 } from "../src/index.js";
@@ -30,6 +32,13 @@ describe("getConnectedPeersForProtocolAndShard", function () {
   let serviceNode1: ServiceNode;
   let serviceNode2: ServiceNode;
   const contentTopic = "/test/2/waku-light-push/utf8";
+  const autoshardingClusterId = resolveAutoshardingCluster(6);
+
+  before(async () => {
+    if (!isNwakuAtLeast("0.27.0")) {
+      this.ctx.skip();
+    }
+  });
 
   beforeEachCustom(this, async () => {
     serviceNode1 = new ServiceNode(makeLogFileName(this.ctx) + "1");
@@ -231,7 +240,7 @@ describe("getConnectedPeersForProtocolAndShard", function () {
     this.timeout(15000);
 
     const shardInfo: ContentTopicInfo = {
-      clusterId: 2,
+      clusterId: autoshardingClusterId,
       contentTopics: [contentTopic]
     };
 
@@ -240,6 +249,7 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo),
+      contentTopic: [contentTopic],
       lightpush: true,
       relay: true
     });
@@ -263,12 +273,12 @@ describe("getConnectedPeersForProtocolAndShard", function () {
     this.timeout(15000);
 
     const shardInfo1: ContentTopicInfo = {
-      clusterId: 2,
+      clusterId: autoshardingClusterId,
       contentTopics: [contentTopic]
     };
 
     const shardInfo2: ContentTopicInfo = {
-      clusterId: 2,
+      clusterId: autoshardingClusterId,
       contentTopics: ["/test/5/waku-light-push/utf8"]
     };
 
@@ -278,6 +288,7 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo1.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo1),
+      contentTopic: [contentTopic],
       lightpush: true,
       relay: true
     });
@@ -288,6 +299,7 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo2.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo2),
+      contentTopic: [contentTopic],
       lightpush: true,
       relay: true
     });
@@ -315,12 +327,12 @@ describe("getConnectedPeersForProtocolAndShard", function () {
     this.timeout(15000);
 
     const shardInfo1: ContentTopicInfo = {
-      clusterId: 2,
+      clusterId: autoshardingClusterId,
       contentTopics: [contentTopic]
     };
 
     const shardInfo2: ContentTopicInfo = {
-      clusterId: 3,
+      clusterId: 2,
       contentTopics: [contentTopic]
     };
 
@@ -330,6 +342,7 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo1.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo1),
+      contentTopic: [contentTopic],
       lightpush: true,
       relay: true
     });
@@ -367,12 +380,12 @@ describe("getConnectedPeersForProtocolAndShard", function () {
     this.timeout(15000);
 
     const shardInfo1: ContentTopicInfo = {
-      clusterId: 2,
+      clusterId: autoshardingClusterId,
       contentTopics: [contentTopic]
     };
 
     const shardInfo2: ContentTopicInfo = {
-      clusterId: 3,
+      clusterId: 2,
       contentTopics: ["/test/5/waku-light-push/utf8"]
     };
 
@@ -382,6 +395,7 @@ describe("getConnectedPeersForProtocolAndShard", function () {
       peerExchange: true,
       clusterId: shardInfo1.clusterId,
       pubsubTopic: shardInfoToPubsubTopics(shardInfo1),
+      contentTopic: [contentTopic],
       lightpush: true,
       relay: true
     });

packages/tests/tests/light-push/single_node/multiple_pubsub.node.spec.ts

@@ -19,8 +19,10 @@ import { expect } from "chai";
 import {
   afterEachCustom,
   beforeEachCustom,
+  isNwakuAtLeast,
   makeLogFileName,
   MessageCollector,
+  resolveAutoshardingCluster,
   ServiceNode,
   tearDownNodes
 } from "../../../src/index.js";
@@ -177,7 +179,7 @@ describe("Waku Light Push (Autosharding): Multiple PubsubTopics", function () {
   let nwaku2: ServiceNode;
   let messageCollector: MessageCollector;
 
-  const clusterId = 2;
+  const clusterId = resolveAutoshardingCluster(4);
   const customContentTopic1 = "/waku/2/content/test.js";
   const customContentTopic2 = "/myapp/1/latest/proto";
   const autoshardingPubsubTopic1 = contentTopicToPubsubTopic(
@@ -203,6 +205,12 @@ describe("Waku Light Push (Autosharding): Multiple PubsubTopics", function () {
 
   let nimPeerId: PeerId;
 
+  before(async () => {
+    if (!isNwakuAtLeast("0.27.0")) {
+      this.ctx.skip();
+    }
+  });
+
   beforeEachCustom(this, async () => {
     [nwaku, waku] = await runNodes(
       this.ctx,

packages/tests/tests/light-push/utils.ts

@@ -1,5 +1,6 @@
 import { createEncoder, waitForRemotePeer } from "@waku/core";
 import {
+  ContentTopicInfo,
   DefaultPubsubTopic,
   LightNode,
   Protocols,
@@ -23,13 +24,23 @@ export async function runNodes(
   shardInfo?: ShardingParams
 ): Promise<[ServiceNode, LightNode]> {
   const nwaku = new ServiceNode(makeLogFileName(context));
+
+  function isContentTopicInfo(info: ShardingParams): info is ContentTopicInfo {
+    return (info as ContentTopicInfo).contentTopics !== undefined;
+  }
+
   await nwaku.start(
     {
       lightpush: true,
       filter: true,
       relay: true,
       pubsubTopic: pubsubTopics,
-      ...(shardInfo && { clusterId: shardInfo.clusterId })
+      // Conditionally include clusterId if shardInfo exists
+      ...(shardInfo && { clusterId: shardInfo.clusterId }),
+      // Conditionally include contentTopic if shardInfo exists and clusterId is 1
+      ...(shardInfo &&
+        isContentTopicInfo(shardInfo) &&
+        shardInfo.clusterId === 1 && { contentTopic: shardInfo.contentTopics })
     },
     { retries: 3 }
   );