status-im/nim-libp2p
2
0
Fork 0
mirror of https://github.com/status-im/nim-libp2p.git synced 2025-01-18 17:01:36 +00:00
Code Issues Projects Releases Wiki Activity

rebase

Browse Source
This commit is contained in:
shashankshampi 2024-10-24 13:42:51 +05:30
parent d80abe09ae
commit a7e80de4d8
1 changed files with 31 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
tests/pubsub/testgossipmembership.nim
View File

@ -87,7 +87,6 @@ suite "GossipSub Topic Membership Tests":
topic: string,
handler: proc(topic: string, data: seq[byte]) {.async.},
) =
# Subscribe all nodes to the topic
for node in nodes:
node.subscribe(topic, handler)
echo "Subscribed all nodes to the topic: ", topic
@ -97,7 +96,6 @@ suite "GossipSub Topic Membership Tests":
topic: string,
handler: proc(topic: string, data: seq[byte]) {.async.},
) =
# Unsubscribe all nodes from the topic
for node in nodes:
node.unsubscribe(topic, handler)
echo "Unsubscribed all nodes from the topic: ", topic
@ -107,19 +105,13 @@ suite "GossipSub Topic Membership Tests":
let topic = "test-topic"
let (gossipSub, conns) = setupGossipSub(topic, 5)
# Subscribe to the topic
subscribeToTopics(gossipSub, @[topic])
# Check if the topic is present in the list of subscribed topics
check gossipSub.topics.contains(topic)
# Check if the topic is added to gossipsub and the peers list is not empty
check gossipSub.gossipsub[topic].len() == 5
# Close all peer connections and verify that they are properly cleaned up
await allFuturesThrowing(conns.mapIt(it.close()))
# Stop the gossipSub switch and wait for it to stop completely
await gossipSub.switch.stop()
# Simulate an UNSUBSCRIBE to the topic and check if the topic is removed from the relevant data structures but remains in gossipsub
@ -127,18 +119,14 @@ suite "GossipSub Topic Membership Tests":
let topic = "test-topic"
let (gossipSub, conns) = setupGossipSub(topic, 5)
# Subscribe to the topic first
subscribeToTopics(gossipSub, @[topic])
# Now unsubscribe from the topic
unsubscribeFromTopics(gossipSub, @[topic])
# Verify the topic is removed from relevant structures
check topic notin gossipSub.topics
check topic notin gossipSub.mesh
check topic in gossipSub.gossipsub
# The topic should remain in gossipsub
await allFuturesThrowing(conns.mapIt(it.close()))
await gossipSub.switch.stop()
@ -147,18 +135,14 @@ suite "GossipSub Topic Membership Tests":
let topics = ["topic1", "topic2", "topic3"].toSeq()
let (gossipSub, conns) = setupGossipSub(topics, 5)
# Subscribe to multiple topics
subscribeToTopics(gossipSub, topics)
# Verify that all topics are added to the topics and gossipsub
check gossipSub.topics.len == 3
for topic in topics:
check gossipSub.gossipsub[topic].len() == 5
# Unsubscribe from all topics
unsubscribeFromTopics(gossipSub, topics)
# Ensure topics are removed from topics and mesh, but still present in gossipsub
for topic in topics:
check topic notin gossipSub.topics
check topic notin gossipSub.mesh
@ -169,14 +153,13 @@ suite "GossipSub Topic Membership Tests":
# Test ensuring that the number of subscriptions does not exceed the limit set in the GossipSub parameters
asyncTest "subscription limit test":
let topicCount = 15 # Total number of topics to be tested
let gossipSubParams = 10 # Subscription limit for the topics
let topicCount = 15
let gossipSubParams = 10
let topicNames = toSeq(mapIt(0 .. topicCount - 1, "topic" & $it))
# Use setupGossipSub to initialize the GossipSub system with connections
let (gossipSub, conns) = setupGossipSub(topicNames, 0) # No peers for now
let (gossipSub, conns) = setupGossipSub(topicNames, 0)
gossipSub.topicsHigh = gossipSubParams # Set the subscription limit
gossipSub.topicsHigh = gossipSubParams
for topic in topicNames:
if gossipSub.topics.len < gossipSub.topicsHigh:
@ -187,13 +170,10 @@ suite "GossipSub Topic Membership Tests":
,
)
else:
# Assert that no subscription happens beyond the limit with custom message
doAssert gossipSub.topics.len == gossipSub.topicsHigh,
"Subscription limit exceeded for topic: " & topic
check gossipSub.topics.len == gossipSub.topicsHigh
# Ensure that the number of subscribed topics does not exceed the limit
doAssert gossipSub.topics.len <= gossipSub.topicsHigh
doAssert gossipSub.topics.len == gossipSub.topicsHigh
check gossipSub.topics.len <= gossipSub.topicsHigh
check gossipSub.topics.len == gossipSub.topicsHigh
await allFuturesThrowing(conns.mapIt(it.close()))
await gossipSub.switch.stop()
@ -202,83 +182,65 @@ suite "GossipSub Topic Membership Tests":
asyncTest "handle JOIN topic and mesh is updated":
let topic = "test-join-topic"
# Initialize the GossipSub system and simulate peer connections
let (gossipSub, conns) = setupGossipSub(topic, 5)
# Simulate peer joining the topic using commonSubscribe
commonSubscribe(@[gossipSub], topic, voidTopicHandler)
# Check that peers are added to the mesh and the topic is tracked
check gossipSub.mesh[topic].len == 5
check gossipSub.topics.contains(topic)
await allFuturesThrowing(conns.mapIt(it.close()))
await gossipSub.switch.stop()
# Test for verifying peers leaving a topic using `LEAVE(topic)`
# Test the behavior when multiple peers join and leave a topic simultaneously.
asyncTest "multiple peers join and leave topic simultaneously":
let
numberOfNodes = 6
topic = "foobar"
var nodes = newSeq[TestGossipSub]()
nodes = generateNodes(numberOfNodes, gossip = true)
nodesFut = await allFinished(nodes.mapIt(it.switch.start()))
# Initialize each node and start the switch
for i in 0 ..< numberOfNodes:
let (gossipSub, _) = setupGossipSub(topic, 5)
nodes.add(gossipSub) # Add the gossipSub instance to the sequence
await gossipSub.switch.start()
await subscribeNodes(nodes)
for node in nodes:
node.subscribe(topic, voidTopicHandler)
# Subscribe all nodes to the topic using the commonSubscribe method
commonSubscribe(nodes, topic, voidTopicHandler)
# Allow time for subscription propagation
await sleepAsync(2 * DURATION_TIMEOUT)
# Ensure that all nodes are subscribed to the topic
for i in 0 ..< numberOfNodes:
let currentGossip = nodes[i]
doAssert currentGossip.gossipsub.hasKey(topic), "Node is not subscribed to the topic"
let currentGossip = GossipSub(nodes[i])
check currentGossip.gossipsub.hasKey(topic)
for i in 0 ..< numberOfNodes:
let currentGossip = GossipSub(nodes[i])
for x in 0 ..< numberOfNodes:
for y in 0 ..< numberOfNodes:
if x != y:
await waitSub(nodes[x], nodes[y], topic)
# Allow time for mesh stabilization
await sleepAsync(2 * DURATION_TIMEOUT)
# Print the mesh size for all nodes before unsubscription
for i in 0 ..< numberOfNodes:
let currentGossip = nodes[i]
echo "Node ", i, " mesh size: ", currentGossip.mesh.getOrDefault(topic).len
# Expected number of peers in the mesh
let expectedNumberOfPeers = numberOfNodes - 1
for i in 0 ..< numberOfNodes:
let currentGossip = nodes[i]
let currentGossip = GossipSub(nodes[i])
check:
currentGossip.gossipsub[topic].len == expectedNumberOfPeers
currentGossip.mesh[topic].len == expectedNumberOfPeers
currentGossip.fanout.len == 0
# Simulate unsubscription of 3 peers
let peersToUnsubscribe = nodes[0].mesh[topic].toSeq()[0 .. 2]
# Unsubscribe these peers from all the nodes
let firstNodeGossip = GossipSub(nodes[0])
let peersToUnsubscribe = firstNodeGossip.mesh[topic].toSeq()[0 .. 2]
let peerIdsToUnsubscribe = peersToUnsubscribe.mapIt(it.peerId)
for node in nodes:
for peer in peersToUnsubscribe:
for peerId in peerIdsToUnsubscribe:
node.unsubscribe(topic, voidTopicHandler)
echo "Unsubscribing peer: ", peer.peerId
node.peers.del(peerId)
# Allow time for heartbeat to adjust the mesh
await sleepAsync(3 * DURATION_TIMEOUT)
# Check the mesh size for the first node after unsubscription
let firstNodeGossip = nodes[0]
echo "Mesh size after unsubscription: ",
firstNodeGossip.mesh.getOrDefault(topic).len
doAssert firstNodeGossip.mesh.getOrDefault(topic).len == 3,
"Expected 3 peers to remain in the mesh"
check firstNodeGossip.mesh.getOrDefault(topic).len == 3
# Assert that unsubscribed peers are no longer in the mesh
for peer in peersToUnsubscribe:
for node in nodes:
doAssert not node.mesh[topic].contains(peer),
"Unsubscribed peer should not be in the mesh"
check not firstNodeGossip.mesh[topic].contains(peer)
await allFuturesThrowing(nodes.mapIt(allFutures(it.switch.stop())))