logos-messaging/logos-messaging-go-bindings
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-go-bindings.git synced 2026-01-04 23:13:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix review points

Browse Source
This commit is contained in:
aya 2025-02-19 09:21:59 +02:00
parent 02781be64f
commit e53eb9d583
6 changed files with 363 additions and 378 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
waku/nodes_basic_test.go
View File

@ -19,7 +19,6 @@ func TestBasicWakuNodes(t *testing.T) {
// Use defer to ensure proper cleanup
defer func() {
Debug("Stopping and destroying Node")
node.StopAndDestroy()
}()

119
waku/nwaku.go
View File

@ -334,7 +334,6 @@ import (
"github.com/waku-org/go-waku/waku/v2/utils"
"github.com/waku-org/waku-go-bindings/waku/common"
"go.uber.org/zap"
"google.golang.org/protobuf/proto"
)
const requestTimeout = 30 * time.Second
@ -683,6 +682,7 @@ func (n *WakuNode) GetConnectedPeers() (peer.IDSlice, error) {
}
func (n *WakuNode) RelaySubscribe(pubsubTopic string) error {
wg := sync.WaitGroup{}
if pubsubTopic == "" {
err := errors.New("pubsub topic is empty")
Error("Failed to subscribe to relay: %v", err)
@ -696,10 +696,6 @@ func (n *WakuNode) RelaySubscribe(pubsubTopic string) error {
}
Debug("Attempting to subscribe to relay on node %s, pubsubTopic: %s", n.nodeName, pubsubTopic)
wg := sync.WaitGroup{}
wg.Add(1)
var resp = C.allocResp(unsafe.Pointer(&wg))
var cPubsubTopic = C.CString(pubsubTopic)
@ -707,10 +703,9 @@ func (n *WakuNode) RelaySubscribe(pubsubTopic string) error {
defer C.free(unsafe.Pointer(cPubsubTopic))
Debug("Calling cGoWakuRelaySubscribe on node %s, pubsubTopic: %s", n.nodeName, pubsubTopic)
wg.Add(1)
C.cGoWakuRelaySubscribe(n.wakuCtx, cPubsubTopic, resp)
Debug("Waiting for response from cGoWakuRelaySubscribe on node %s", n.nodeName)
wg.Wait() // Ensures the function completes before proceeding
wg.Wait()
if C.getRet(resp) == C.RET_OK {
Debug("Successfully subscribed to relay on node %s, pubsubTopic: %s", n.nodeName, pubsubTopic)
@ -953,7 +948,7 @@ func (n *WakuNode) RelayPublishNoCTX(pubsubTopic string, message *pb.WakuMessage
}
// Handling context internally with a timeout
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
defer cancel()
Debug("Attempting to publish message via relay on node %s", n.nodeName)
@ -1416,19 +1411,24 @@ func StartWakuNode(nodeName string, customCfg *WakuConfig) (*WakuNode, error) {
var nodeCfg WakuConfig
if customCfg == nil {
nodeCfg = DefaultWakuConfig
} else {
nodeCfg = *customCfg
}
tcpPort, udpPort, err := GetFreePortIfNeeded(0, 0)
if err != nil {
Error("Failed to allocate unique ports: %v", err)
tcpPort, udpPort = 0, 0 // Fallback to OS-assigned ports
}
if nodeCfg.TcpPort == 0 || nodeCfg.Discv5UdpPort == 0 {
tcpPort, udpPort, err := GetFreePortIfNeeded(0, 0)
if err != nil {
Error("Failed to allocate unique ports: %v", err)
tcpPort, udpPort = 0, 0
}
nodeCfg.TcpPort = tcpPort
nodeCfg.Discv5UdpPort = udpPort
if nodeCfg.TcpPort == 0 {
nodeCfg.TcpPort = tcpPort
}
if nodeCfg.Discv5UdpPort == 0 {
nodeCfg.Discv5UdpPort = udpPort
}
}
Debug("Creating %s", nodeName)
node, err := NewWakuNode(&nodeCfg, nodeName)
@ -1448,6 +1448,7 @@ func StartWakuNode(nodeName string, customCfg *WakuConfig) (*WakuNode, error) {
}
func (n *WakuNode) StopAndDestroy() error {
Debug("Stopping and destroying Node")
if n == nil {
err := errors.New("waku node is nil")
Error("Failed to stop and destroy: %v", err)
@ -1550,91 +1551,11 @@ func ConnectAllPeers(nodes []*WakuNode) error {
return nil
}
func (n *WakuNode) VerifyMessageReceived(expectedMessage *pb.WakuMessage, expectedHash common.MessageHash) error {
timeout := 3 * time.Second
Debug("Verifying if the message was received on node %s, timeout: %v", n.nodeName, timeout)
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
select {
case envelope := <-n.MsgChan:
if envelope == nil {
Error("Received envelope is nil on node %s", n.nodeName)
return errors.New("received envelope is nil")
}
if string(expectedMessage.Payload) != string(envelope.Message().Payload) {
Error("Payload does not match on node %s", n.nodeName)
return errors.New("payload does not match")
}
if expectedMessage.ContentTopic != envelope.Message().ContentTopic {
Error("Content topic does not match on node %s", n.nodeName)
return errors.New("content topic does not match")
}
if expectedHash != envelope.Hash() {
Error("Message hash does not match on node %s", n.nodeName)
return errors.New("message hash does not match")
}
Debug("Message received and verified successfully on node %s, Message: %s", n.nodeName, string(envelope.Message().Payload))
return nil
case <-ctx.Done():
Error("Timeout: message not received within %v on node %s", timeout, n.nodeName)
return errors.New("timeout: message not received within the given duration")
}
}
func (n *WakuNode) CreateMessage(customMessage ...*pb.WakuMessage) *pb.WakuMessage {
Debug("Creating a WakuMessage on node %s", n.nodeName)
if len(customMessage) > 0 && customMessage[0] != nil {
Debug("Using provided custom message on node %s", n.nodeName)
return customMessage[0]
}
Debug("Using default message format on node %s", n.nodeName)
defaultMessage := &pb.WakuMessage{
Payload: []byte("This is a default Waku message payload"),
ContentTopic: "test-content-topic",
Version: proto.Uint32(0),
Timestamp: proto.Int64(time.Now().UnixNano()),
}
Debug("Successfully created a default WakuMessage on node %s", n.nodeName)
return defaultMessage
}
func WaitForAutoConnection(nodeList []*WakuNode) error {
Debug("Waiting for auto-connection of nodes...")
var hardWait = 30 * time.Second
Debug("Applying hard wait of %v seconds before checking connections", hardWait.Seconds())
time.Sleep(hardWait)
for _, node := range nodeList {
peers, err := node.GetConnectedPeers()
if err != nil {
Error("Failed to get connected peers for node %s: %v", node.nodeName, err)
return err
}
if len(peers) < 1 {
Error("Node %s has no connected peers, expected at least 1", node.nodeName)
return errors.New("expected at least one connected peer")
}
Debug("Node %s has %d connected peers", node.nodeName, len(peers))
}
Debug("Auto-connection check completed successfully")
return nil
}
func SubscribeNodesToTopic(nodes []*WakuNode, topic string) error {
for _, node := range nodes {
Debug("Subscribing node %s to topic %s", node.nodeName, topic)
err := RetryWithBackOff(func() error {
return node.RelaySubscribe(topic)
})
err := node.RelaySubscribe(topic)
if err != nil {
Error("Failed to subscribe node %s to topic %s: %v", node.nodeName, topic, err)
return err

98
waku/nwaku_test_utils.go
View File

@ -1,7 +1,9 @@
package waku
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
@ -10,6 +12,9 @@ import (
"time"
"github.com/cenkalti/backoff/v3"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/waku-go-bindings/waku/common"
"google.golang.org/protobuf/proto"
)
type NwakuInfo struct {
@ -77,3 +82,96 @@ func RetryWithBackOff(o func() error, options ...BackOffOption) error {
b.Reset()
return backoff.Retry(o, &b)
}
func (n *WakuNode) CreateMessage(customMessage ...*pb.WakuMessage) *pb.WakuMessage {
Debug("Creating a WakuMessage on node %s", n.nodeName)
if len(customMessage) > 0 && customMessage[0] != nil {
Debug("Using provided custom message on node %s", n.nodeName)
return customMessage[0]
}
Debug("Using default message format on node %s", n.nodeName)
defaultMessage := &pb.WakuMessage{
Payload: []byte("This is a default Waku message payload"),
ContentTopic: "test-content-topic",
Version: proto.Uint32(0),
Timestamp: proto.Int64(time.Now().UnixNano()),
}
Debug("Successfully created a default WakuMessage on node %s", n.nodeName)
return defaultMessage
}
func WaitForAutoConnection(nodeList []*WakuNode) error {
Debug("Waiting for auto-connection of nodes...")
options := func(b *backoff.ExponentialBackOff) {
b.MaxElapsedTime = 30 * time.Second
}
err := RetryWithBackOff(func() error {
for _, node := range nodeList {
peers, err := node.GetConnectedPeers()
if err != nil {
return err
}
if len(peers) < 1 {
return errors.New("expected at least one connected peer") // Retry
}
Debug("Node %s has %d connected peers", node.nodeName, len(peers))
}
return nil
}, options)
if err != nil {
Error("Auto-connection failed after retries: %v", err)
return err
}
Debug("Auto-connection check completed successfully")
return nil
}
func (n *WakuNode) VerifyMessageReceived(expectedMessage *pb.WakuMessage, expectedHash common.MessageHash, timeout ...time.Duration) error {
var verifyTimeout time.Duration
if len(timeout) > 0 {
verifyTimeout = timeout[0]
} else {
verifyTimeout = DefaultTimeOut
}
Debug("Verifying if the message was received on node %s, timeout: %v", n.nodeName, verifyTimeout)
ctx, cancel := context.WithTimeout(context.Background(), verifyTimeout)
defer cancel()
select {
case envelope := <-n.MsgChan:
if envelope == nil {
Error("Received envelope is nil on node %s", n.nodeName)
return errors.New("received envelope is nil")
}
if string(expectedMessage.Payload) != string(envelope.Message().Payload) {
Error("Payload does not match on node %s", n.nodeName)
return errors.New("payload does not match")
}
if expectedMessage.ContentTopic != envelope.Message().ContentTopic {
Error("Content topic does not match on node %s", n.nodeName)
return errors.New("content topic does not match")
}
if expectedHash != envelope.Hash() {
Error("Message hash does not match on node %s", n.nodeName)
return errors.New("message hash does not match")
}
Debug("Message received and verified successfully on node %s, Message: %s", n.nodeName, string(envelope.Message().Payload))
return nil
case <-ctx.Done():
Error("Timeout: message not received within %v on node %s", verifyTimeout, n.nodeName)
return errors.New("timeout: message not received within the given duration")
}
}

150
waku/peer_connections_test.go
View File

@ -31,8 +31,6 @@ func TestDisconnectPeerNodes(t *testing.T) {
require.NoError(t, err, "Failed to get PeerID for Node B")
require.True(t, slices.Contains(connectedPeers, nodeBPeerID), "Node B should be a peer of Node A before disconnection")
time.Sleep(3 * time.Second)
Debug("Disconnecting Node A from Node B")
err = nodeA.DisconnectPeer(nodeB)
require.NoError(t, err, "Failed to disconnect nodes")
@ -78,16 +76,152 @@ func TestConnectMultipleNodesToSingleNode(t *testing.T) {
err = node3.ConnectPeer(node1)
require.NoError(t, err, "Failed to connect Node 3 to Node 1")
Debug("Verifying connected peers for Node 3")
connectedPeers, err := node3.GetConnectedPeers()
require.NoError(t, err, "Failed to get connected peers for Node 3")
node1PeerID, err := node1.PeerID()
Debug("Verifying connected peers for Node 1")
connectedPeers, err := node1.GetConnectedPeers()
require.NoError(t, err, "Failed to get connected peers for Node 1")
node3PeerID, err := node3.PeerID()
require.NoError(t, err, "Failed to get PeerID for Node 1")
node2PeerID, err := node2.PeerID()
require.NoError(t, err, "Failed to get PeerID for Node 2")
require.True(t, slices.Contains(connectedPeers, node1PeerID), "Node 1 should be a peer of Node 3")
require.True(t, slices.Contains(connectedPeers, node2PeerID), "Node 2 should be a peer of Node 3")
require.True(t, slices.Contains(connectedPeers, node3PeerID), "Node 3 should be a peer of Node 1")
require.True(t, slices.Contains(connectedPeers, node2PeerID), "Node 2 should be a peer of Node 1")
Debug("Test completed successfully: multiple nodes connected to a single node and verified peers")
}
func TestDiscv5PeerMeshCount(t *testing.T) {
Debug("Starting test to verify peer count in mesh using Discv5 after topic subscription")
node1Config := DefaultWakuConfig
node1Config.Relay = true
Debug("Creating Node1")
node1, err := StartWakuNode("Node1", &node1Config)
require.NoError(t, err, "Failed to start Node1")
enrNode1, err := node1.ENR()
require.NoError(t, err, "Failed to get ENR for Node1")
node2Config := DefaultWakuConfig
node2Config.Discv5BootstrapNodes = []string{enrNode1.String()}
node2Config.Relay = true
Debug("Creating Node2 with Node1 as Discv5 bootstrap")
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err, "Failed to start Node2")
require.NoError(t, err, "Failed to get ENR for Node2")
node3Config := DefaultWakuConfig
node3Config.Discv5BootstrapNodes = []string{enrNode1.String()}
node3Config.Relay = true
Debug("Creating Node3 with Node2 as Discv5 bootstrap")
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err, "Failed to start Node3")
defer func() {
Debug("Stopping and destroying all Waku nodes")
node1.StopAndDestroy()
node2.StopAndDestroy()
}()
defaultPubsubTopic := DefaultPubsubTopic
Debug("Default pubsub topic retrieved: %s", defaultPubsubTopic)
err = SubscribeNodesToTopic([]*WakuNode{node1, node2, node3}, defaultPubsubTopic)
require.NoError(t, err, "Failed to subscribe all nodes to the topic")
Debug("Waiting for nodes to auto-connect via Discv5")
err = WaitForAutoConnection([]*WakuNode{node1, node2, node3})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
Debug("Fetching number of peers in mesh for Node1 before stopping Node3")
peerCountBefore, err := node1.GetNumPeersInMesh(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of peers in mesh for Node1 before stopping Node3")
Debug("Total number of peers in mesh for Node1 before stopping Node3: %d", peerCountBefore)
require.Equal(t, 2, peerCountBefore, "Expected Node1 to have exactly 2 peers in the mesh before stopping Node3")
Debug("Stopping Node3")
node3.StopAndDestroy()
Debug("Waiting for network update after Node3 stops")
time.Sleep(10 * time.Second)
Debug("Fetching number of peers in mesh for Node1 after stopping Node3")
peerCountAfter, err := node1.GetNumPeersInMesh(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of peers in mesh for Node1 after stopping Node3")
Debug("Total number of peers in mesh for Node1 after stopping Node3: %d", peerCountAfter)
require.Equal(t, 1, peerCountAfter, "Expected Node1 to have exactly 1 peer in the mesh after stopping Node3")
Debug("Test successfully verified peer count change after stopping Node3")
}
// this test commented as it will fail will be changed to have external ip in future task
/*
func TestDiscv5GetPeersConnected(t *testing.T) {
Debug("Starting test to verify peer count in mesh with 4 nodes using Discv5 (Chained Connection)")
node1Config := DefaultWakuConfig
node1Config.Relay = true
Debug("Creating Node1")
node1, err := StartWakuNode("Node1", &node1Config)
require.NoError(t, err, "Failed to start Node1")
enrNode1, err := node1.ENR()
require.NoError(t, err, "Failed to get ENR for Node1")
node2Config := DefaultWakuConfig
node2Config.Relay = true
node2Config.Discv5BootstrapNodes = []string{enrNode1.String()}
Debug("Creating Node2 with Node1 as Discv5 bootstrap")
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err, "Failed to start Node2")
enrNode2, err := node2.ENR()
require.NoError(t, err, "Failed to get ENR for Node2")
node3Config := DefaultWakuConfig
node3Config.Relay = true
node3Config.Discv5BootstrapNodes = []string{enrNode2.String()}
Debug("Creating Node3 with Node2 as Discv5 bootstrap")
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err, "Failed to start Node3")
enrNode3, err := node3.ENR()
require.NoError(t, err, "Failed to get ENR for Node3")
node4Config := DefaultWakuConfig
node4Config.Relay = true
node4Config.Discv5BootstrapNodes = []string{enrNode3.String()}
Debug("Creating Node4 with Node3 as Discv5 bootstrap")
node4, err := StartWakuNode("Node4", &node4Config)
require.NoError(t, err, "Failed to start Node4")
defer func() {
Debug("Stopping and destroying all Waku nodes")
node1.StopAndDestroy()
node2.StopAndDestroy()
node3.StopAndDestroy()
node4.StopAndDestroy()
}()
Debug("Waiting for nodes to auto-connect via Discv5")
err = WaitForAutoConnection([]*WakuNode{node1, node2, node3, node4})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
Debug("Fetching number of peers in connected to Node1")
peerCount, err := node1.GetNumConnectedPeers()
require.NoError(t, err, "Failed to get number of peers in mesh for Node1")
Debug("Total number of peers connected to Node1: %d", peerCount)
require.Equal(t, 3, peerCount, "Expected Node1 to have exactly 3 peers in the mesh")
Debug("Test successfully verified peer count in mesh with 4 nodes using Discv5 (Chained Connection)")
}
*/

361
waku/relay_test.go
View File

@ -1,95 +1,69 @@
package waku
import (
"fmt"
"testing"
"time"
"github.com/cenkalti/backoff/v3"
"github.com/stretchr/testify/require"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/waku-go-bindings/waku/common"
"google.golang.org/protobuf/proto"
)
func TestNumRelayPeers(t *testing.T) {
Debug("Starting test to verify relay subscription with multiple nodes via Discv5")
func TestVerifyNumConnectedRelayPeers(t *testing.T) {
// Configure Node1
node1Config := DefaultWakuConfig
node1Config.Relay = true
node1Cfg := DefaultWakuConfig
node1Cfg.Relay = true
node1, err := StartWakuNode("node1", &node1Cfg)
if err != nil {
t.Fatalf("Failed to start node1: %v", err)
}
node2Cfg := DefaultWakuConfig
node2Cfg.Relay = true
node2, err := StartWakuNode("node2", &node2Cfg)
if err != nil {
t.Fatalf("Failed to start node2: %v", err)
}
Debug("Creating Node1 with Relay enabled")
node1, err := StartWakuNode("Node1", &node1Config)
require.NoError(t, err, "Failed to start Node1")
// Retrieve Node1's ENR for Discv5 bootstrapping
enrNode1, err := node1.ENR()
require.NoError(t, err, "Failed to get ENR for Node1")
// Configure Node2 with Node1 as its Discv5 bootstrap node
node2Config := DefaultWakuConfig
node2Config.Relay = true
node2Config.Discv5BootstrapNodes = []string{enrNode1.String()}
Debug("Creating Node2 with Node1 as Discv5 bootstrap")
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err, "Failed to start Node2")
// Configure Node3 with Node1 as its Discv5 bootstrap node
node3Config := DefaultWakuConfig
node3Config.Relay = true
node3Config.Discv5BootstrapNodes = []string{enrNode1.String()}
Debug("Creating Node3 with Node1 as Discv5 bootstrap")
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err, "Failed to start Node3")
node3, err := StartWakuNode("node3", nil)
if err != nil {
t.Fatalf("Failed to start node3: %v", err)
}
defer func() {
Debug("Stopping and destroying all Waku nodes")
node1.StopAndDestroy()
node2.StopAndDestroy()
node3.StopAndDestroy()
}()
defaultPubsubTopic := DefaultPubsubTopic
Debug("Default pubsub topic retrieved: %s", defaultPubsubTopic)
Debug("Waiting for nodes to auto-connect via Discv5")
err = WaitForAutoConnection([]*WakuNode{node1, node2, node3})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
Debug("Subscribing Node1 to the default pubsub topic: %s", defaultPubsubTopic)
err = RetryWithBackOff(func() error {
return node1.RelaySubscribe(defaultPubsubTopic)
})
require.NoError(t, err)
Debug("Subscribing Node2 to the default pubsub topic: %s", defaultPubsubTopic)
err = RetryWithBackOff(func() error {
return node2.RelaySubscribe(defaultPubsubTopic)
})
require.NoError(t, err)
Debug("Subscribing Node3 to the default pubsub topic: %s", defaultPubsubTopic)
err = RetryWithBackOff(func() error {
return node3.RelaySubscribe(defaultPubsubTopic)
})
require.NoError(t, err)
Debug("Waiting for peer connections to stabilize")
time.Sleep(10 * time.Second)
Debug("Fetching number of connected peers for Node1")
peers, err := node1.GetNumConnectedRelayPeers(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of connected relay peers for Node1")
Debug("Total number of connected relay peers for Node1: %d", peers)
if peers != 2 {
Error("Expected Node1 to have exactly 2 relay peers, but got %d", peers)
t.FailNow()
err = node2.ConnectPeer(node1)
if err != nil {
t.Fatalf("Failed to connect node2 to node1: %v", err)
}
Debug("Test successfully verified that Node1 has 2 relay peers after Node2 and Node3 subscribed")
err = node3.ConnectPeer(node1)
if err != nil {
t.Fatalf("Failed to connect node3 to node1: %v", err)
}
connectedPeersNode1, err := node1.GetConnectedPeers()
if err != nil {
t.Fatalf("Failed to get connected peers for node1: %v", err)
}
if len(connectedPeersNode1) != 2 {
t.Fatalf("Expected 2 connected peers on node1, but got %d", len(connectedPeersNode1))
}
numRelayPeers, err := node1.GetNumConnectedRelayPeers()
if err != nil {
t.Fatalf("Failed to get connected relay peers for node1: %v", err)
}
if numRelayPeers != 1 {
t.Fatalf("Expected 1 relay peer on node1, but got %d", numRelayPeers)
}
t.Logf("Successfully connected node2 and node3 to node1. Relay Peers: %d, Total Peers: %d", numRelayPeers, len(connectedPeersNode1))
}
func TestRelayMessageTransmission(t *testing.T) {
@ -122,9 +96,7 @@ func TestRelayMessageTransmission(t *testing.T) {
Debug("Subscribing ReceiverNode to the default pubsub topic")
defaultPubsubTopic := DefaultPubsubTopic
err = RetryWithBackOff(func() error {
return receiverNode.RelaySubscribe(defaultPubsubTopic)
})
err = receiverNode.RelaySubscribe(defaultPubsubTopic)
require.NoError(t, err)
Debug("Creating and publishing message")
@ -142,9 +114,6 @@ func TestRelayMessageTransmission(t *testing.T) {
require.NoError(t, err)
require.NotEmpty(t, msgHash)
Debug("Waiting to ensure message delivery")
time.Sleep(2 * time.Second)
Debug("Verifying message reception")
err = RetryWithBackOff(func() error {
msgHashObj, _ := common.ToMessageHash(msgHash)
@ -261,7 +230,7 @@ func TestSendmsgInvalidPayload(t *testing.T) {
}
func TestRelayNodesNotConnectedDirectly(t *testing.T) {
Debug("Starting TestMessageNotReceivedWithoutRelay")
Debug("Starting TestRelayNodesNotConnectedDirectly")
Debug("Creating Sender Node with Relay enabled")
senderConfig := DefaultWakuConfig
@ -273,23 +242,25 @@ func TestRelayNodesNotConnectedDirectly(t *testing.T) {
Debug("Creating Relay-Enabled Receiver Node (Node2)")
node2Config := DefaultWakuConfig
node2Config.Relay = true
enrNode, err := senderNode.ENR()
if err != nil {
require.Error(t, err, "Can't find node ENR")
}
node2Config.Discv5BootstrapNodes = []string{enrNode.String()}
// Use static nodes instead of ENR
node1Address, err := senderNode.ListenAddresses()
require.NoError(t, err, "Failed to get sender node address")
node2Config.Staticnodes = []string{node1Address[0].String()}
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err)
defer node2.StopAndDestroy()
Debug("Creating Non-Relay Receiver Node (Node3)")
Debug("Creating Relay-Enabled Receiver Node (Node3)")
node3Config := DefaultWakuConfig
node3Config.Relay = true
enrNode2, err := node2.ENR()
if err != nil {
require.Error(t, err, "Can't find node ENR")
}
node3Config.Discv5BootstrapNodes = []string{enrNode2.String()}
// Use static nodes instead of Discv5
node2Address, err := node2.ListenAddresses()
require.NoError(t, err, "Failed to get node2 address")
node3Config.Staticnodes = []string{node2Address[0].String()}
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err)
defer node3.StopAndDestroy()
@ -299,13 +270,13 @@ func TestRelayNodesNotConnectedDirectly(t *testing.T) {
err = node2.RelaySubscribe(defaultPubsubTopic)
require.NoError(t, err)
Debug("Subscribing Node2 to the default pubsub topic")
Debug("Subscribing Node3 to the default pubsub topic")
err = node3.RelaySubscribe(defaultPubsubTopic)
require.NoError(t, err)
Debug("Waiting for nodes to auto-connect before proceeding")
Debug("Waiting for nodes to connect before proceeding")
err = WaitForAutoConnection([]*WakuNode{senderNode, node2, node3})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
require.NoError(t, err, "Nodes did not connect within timeout")
Debug("SenderNode is publishing a message")
message := senderNode.CreateMessage()
@ -317,11 +288,11 @@ func TestRelayNodesNotConnectedDirectly(t *testing.T) {
err = node2.VerifyMessageReceived(message, msgHash)
require.NoError(t, err, "Node2 should have received the message")
Debug("Verifying that Node3 receive the message")
Debug("Verifying that Node3 received the message")
err = node3.VerifyMessageReceived(message, msgHash)
require.NoError(t, err, "Node3 should have received the message")
require.NoError(t, err, "Node3 should have received the message")
Debug("TestMessageNotReceivedWithoutRelay completed successfully")
Debug("TestRelayNodesNotConnectedDirectly completed successfully")
}
func TestRelaySubscribeAndPeerCountChange(t *testing.T) {
@ -334,26 +305,25 @@ func TestRelaySubscribeAndPeerCountChange(t *testing.T) {
node1, err := StartWakuNode("Node1", &node1Config)
require.NoError(t, err, "Failed to start Node1")
enrNode1, err := node1.ENR()
require.NoError(t, err, "Failed to get ENR for Node1")
node1Address, err := node1.ListenAddresses()
require.NoError(t, err, "Failed to get listening address for Node1")
node2Config := DefaultWakuConfig
node2Config.Relay = true
node2Config.Discv5BootstrapNodes = []string{enrNode1.String()}
node2Config.Staticnodes = []string{node1Address[0].String()}
Debug("Creating Node2 with Node1 as Discv5 bootstrap")
Debug("Creating Node2 with Node1 as a static node")
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err, "Failed to start Node2")
//enrNode2, err := node2.ENR() // commented till confirm from Gabriel
//if num peers shall be affected by it or not
//require.NoError(t, err, "Failed to get ENR for Node2")
node2Address, err := node2.ListenAddresses()
require.NoError(t, err, "Failed to get listening address for Node2")
node3Config := DefaultWakuConfig
node3Config.Relay = true
node3Config.Discv5BootstrapNodes = []string{enrNode1.String()}
node3Config.Staticnodes = []string{node2Address[0].String()}
Debug("Creating Node3 with Node2 as Discv5 bootstrap")
Debug("Creating Node3 with Node2 as a static node")
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err, "Failed to start Node3")
@ -366,186 +336,55 @@ func TestRelaySubscribeAndPeerCountChange(t *testing.T) {
defaultPubsubTopic := DefaultPubsubTopic
Debug("Default pubsub topic retrieved: %s", defaultPubsubTopic)
Debug("Waiting for nodes to auto-connect via Discv5")
Debug("Waiting for nodes to connect via static node configuration")
err = WaitForAutoConnection([]*WakuNode{node1, node2, node3})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
require.NoError(t, err, "Nodes did not connect within timeout")
Debug("Subscribing Node1 to the default pubsub topic: %s", defaultPubsubTopic)
err = RetryWithBackOff(func() error {
return node1.RelaySubscribe(defaultPubsubTopic)
})
err = node1.RelaySubscribe(defaultPubsubTopic)
require.NoError(t, err)
Debug("Subscribing Node2 to the default pubsub topic: %s", defaultPubsubTopic)
err = RetryWithBackOff(func() error {
return node2.RelaySubscribe(defaultPubsubTopic)
})
err = node2.RelaySubscribe(defaultPubsubTopic)
require.NoError(t, err)
Debug("Subscribing Node3 to the default pubsub topic: %s", defaultPubsubTopic)
err = RetryWithBackOff(func() error {
return node3.RelaySubscribe(defaultPubsubTopic)
})
err = node3.RelaySubscribe(defaultPubsubTopic)
require.NoError(t, err)
Debug("Waiting for peer connections to stabilize")
time.Sleep(10 * time.Second)
options := func(b *backoff.ExponentialBackOff) {
b.MaxElapsedTime = 10 * time.Second // Only set the max wait time
}
require.NoError(t, RetryWithBackOff(func() error {
numPeers, err := node1.GetNumConnectedRelayPeers(defaultPubsubTopic)
if err != nil {
return err
}
if numPeers != 2 {
return fmt.Errorf("expected 2 relay peers, got %d", numPeers)
}
return nil
}, options), "Peers did not stabilize in time")
Debug("Fetching number of connected peers for Node1")
peersBeforeStopping, err := node1.GetNumConnectedRelayPeers(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of connected relay peers for Node1")
Debug("Total number of connected relay peers for Node1 before stopping Node3: %d", peersBeforeStopping)
require.Equal(t, 2, peersBeforeStopping, "Expected Node1 to have exactly 2 relay peers before stopping Node3")
Debug("Stopping Node3")
node3.StopAndDestroy()
Debug("Waiting for network to update after Node3 stops")
time.Sleep(10 * time.Second)
require.NoError(t, RetryWithBackOff(func() error {
numPeers, err := node1.GetNumConnectedRelayPeers(defaultPubsubTopic)
if err != nil {
return err
}
if numPeers != 1 {
return fmt.Errorf("expected 1 relay peer after stopping Node3, got %d", numPeers)
}
return nil
}, options), "Peer count did not update after stopping Node3")
Debug("Fetching number of connected peers for Node1 after stopping Node3")
peersAfterStopping, err := node1.GetNumConnectedRelayPeers(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of connected relay peers for Node1 after stopping Node3")
Debug("Total number of connected relay peers for Node1 after stopping Node3: %d", peersAfterStopping)
require.Equal(t, 1, peersAfterStopping, "Expected Node1 to have exactly 1 relay peer after stopping Node3")
Debug("Test successfully verified peer count changes as expected after stopping Node3")
}
func TestDiscv5PeerMeshCount(t *testing.T) {
Debug("Starting test to verify peer count in mesh using Discv5 after topic subscription")
node1Config := DefaultWakuConfig
node1Config.Relay = true
Debug("Creating Node1")
node1, err := StartWakuNode("Node1", &node1Config)
require.NoError(t, err, "Failed to start Node1")
enrNode1, err := node1.ENR()
require.NoError(t, err, "Failed to get ENR for Node1")
node2Config := DefaultWakuConfig
node2Config.Discv5BootstrapNodes = []string{enrNode1.String()}
node2Config.Relay = true
Debug("Creating Node2 with Node1 as Discv5 bootstrap")
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err, "Failed to start Node2")
require.NoError(t, err, "Failed to get ENR for Node2")
node3Config := DefaultWakuConfig
node3Config.Discv5BootstrapNodes = []string{enrNode1.String()}
node3Config.Relay = true
Debug("Creating Node3 with Node2 as Discv5 bootstrap")
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err, "Failed to start Node3")
defer func() {
Debug("Stopping and destroying all Waku nodes")
node1.StopAndDestroy()
node2.StopAndDestroy()
}()
defaultPubsubTopic := DefaultPubsubTopic
Debug("Default pubsub topic retrieved: %s", defaultPubsubTopic)
err = SubscribeNodesToTopic([]*WakuNode{node1, node2, node3}, defaultPubsubTopic)
require.NoError(t, err, "Failed to subscribe all nodes to the topic")
Debug("Waiting for nodes to auto-connect via Discv5")
err = WaitForAutoConnection([]*WakuNode{node1, node2, node3})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
Debug("Fetching number of peers in mesh for Node1 before stopping Node3")
peerCountBefore, err := node1.GetNumPeersInMesh(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of peers in mesh for Node1 before stopping Node3")
Debug("Total number of peers in mesh for Node1 before stopping Node3: %d", peerCountBefore)
require.Equal(t, 2, peerCountBefore, "Expected Node1 to have exactly 2 peers in the mesh before stopping Node3")
Debug("Stopping Node3")
node3.StopAndDestroy()
Debug("Waiting for network update after Node3 stops")
time.Sleep(10 * time.Second)
Debug("Fetching number of peers in mesh for Node1 after stopping Node3")
peerCountAfter, err := node1.GetNumPeersInMesh(defaultPubsubTopic)
require.NoError(t, err, "Failed to get number of peers in mesh for Node1 after stopping Node3")
Debug("Total number of peers in mesh for Node1 after stopping Node3: %d", peerCountAfter)
require.Equal(t, 1, peerCountAfter, "Expected Node1 to have exactly 1 peer in the mesh after stopping Node3")
Debug("Test successfully verified peer count change after stopping Node3")
}
func TestDiscv5GetPeersConnected(t *testing.T) {
Debug("Starting test to verify peer count in mesh with 4 nodes using Discv5 (Chained Connection)")
node1Config := DefaultWakuConfig
node1Config.Relay = true
Debug("Creating Node1")
node1, err := StartWakuNode("Node1", &node1Config)
require.NoError(t, err, "Failed to start Node1")
enrNode1, err := node1.ENR()
require.NoError(t, err, "Failed to get ENR for Node1")
node2Config := DefaultWakuConfig
node2Config.Relay = true
node2Config.Discv5BootstrapNodes = []string{enrNode1.String()}
Debug("Creating Node2 with Node1 as Discv5 bootstrap")
node2, err := StartWakuNode("Node2", &node2Config)
require.NoError(t, err, "Failed to start Node2")
enrNode2, err := node2.ENR()
require.NoError(t, err, "Failed to get ENR for Node2")
node3Config := DefaultWakuConfig
node3Config.Relay = true
node3Config.Discv5BootstrapNodes = []string{enrNode2.String()}
Debug("Creating Node3 with Node2 as Discv5 bootstrap")
node3, err := StartWakuNode("Node3", &node3Config)
require.NoError(t, err, "Failed to start Node3")
enrNode3, err := node3.ENR()
require.NoError(t, err, "Failed to get ENR for Node3")
node4Config := DefaultWakuConfig
node4Config.Relay = true
node4Config.Discv5BootstrapNodes = []string{enrNode3.String()}
Debug("Creating Node4 with Node3 as Discv5 bootstrap")
node4, err := StartWakuNode("Node4", &node4Config)
require.NoError(t, err, "Failed to start Node4")
defer func() {
Debug("Stopping and destroying all Waku nodes")
node1.StopAndDestroy()
node2.StopAndDestroy()
node3.StopAndDestroy()
node4.StopAndDestroy()
}()
Debug("Waiting for nodes to auto-connect via Discv5")
err = WaitForAutoConnection([]*WakuNode{node1, node2, node3, node4})
require.NoError(t, err, "Nodes did not auto-connect within timeout")
Debug("Fetching number of peers in connected to Node1")
peerCount, err := node1.GetNumConnectedPeers()
require.NoError(t, err, "Failed to get number of peers in mesh for Node1")
Debug("Total number of peers connected to Node1: %d", peerCount)
require.Equal(t, 3, peerCount, "Expected Node1 to have exactly 3 peers in the mesh")
Debug("Test successfully verified peer count in mesh with 4 nodes using Discv5 (Chained Connection)")
}
func TestRelaySubscribeFailsWhenRelayDisabled(t *testing.T) {
Debug("Starting test to verify that subscribing to a topic fails when Relay is disabled")

12
waku/test_data.go
View File

@ -8,13 +8,6 @@ var DefaultWakuConfig WakuConfig
func init() {
udpPort, _, err1 := GetFreePortIfNeeded(0, 0)
tcpPort, _, err2 := GetFreePortIfNeeded(0, 0)
if err1 != nil || err2 != nil {
Error("Failed to get free ports %v %v", err1, err2)
}
DefaultWakuConfig = WakuConfig{
Relay: false,
LogLevel: "DEBUG",
@ -25,12 +18,13 @@ func init() {
Store: false,
Filter: false,
Lightpush: false,
Discv5UdpPort: udpPort,
TcpPort: tcpPort,
Discv5UdpPort: 0,
TcpPort: 0,
}
}
const ConnectPeerTimeout = 10 * time.Second //default timeout for node to connect to another node
const DefaultTimeOut = 3 * time.Second
var DefaultPubsubTopic = "/waku/2/rs/16/64"
var (