mesh check and code rebase
This commit is contained in:
shashankshampi
commit
e41c08fcb2
|
|
@ -33,46 +33,6 @@ suite "GossipSub internal":
|
|||
teardown:
|
||||
checkTrackers()
|
||||
|
||||
asyncTest "subscribe/unsubscribeAll":
|
||||
let gossipSub = TestGossipSub.init(newStandardSwitch())
|
||||
|
||||
proc handler(topic: string, data: seq[byte]): Future[void] {.gcsafe.} =
|
||||
discard
|
||||
|
||||
let topic = "foobar"
|
||||
gossipSub.mesh[topic] = initHashSet[PubSubPeer]()
|
||||
gossipSub.topicParams[topic] = TopicParams.init()
|
||||
|
||||
var conns = newSeq[Connection]()
|
||||
gossipSub.gossipsub[topic] = initHashSet[PubSubPeer]()
|
||||
for i in 0 ..< 15:
|
||||
let conn = TestBufferStream.new(noop)
|
||||
conns &= conn
|
||||
let peerId = randomPeerId()
|
||||
conn.peerId = peerId
|
||||
let peer = gossipSub.getPubSubPeer(peerId)
|
||||
peer.sendConn = conn
|
||||
gossipSub.gossipsub[topic].incl(peer)
|
||||
|
||||
# test via dynamic dispatch
|
||||
gossipSub.PubSub.subscribe(topic, handler)
|
||||
|
||||
check:
|
||||
gossipSub.topics.contains(topic)
|
||||
gossipSub.gossipsub[topic].len() > 0
|
||||
gossipSub.mesh[topic].len() > 0
|
||||
|
||||
# test via dynamic dispatch
|
||||
gossipSub.PubSub.unsubscribeAll(topic)
|
||||
|
||||
check:
|
||||
topic notin gossipSub.topics # not in local topics
|
||||
topic notin gossipSub.mesh # not in mesh
|
||||
topic in gossipSub.gossipsub # but still in gossipsub table (for fanning out)
|
||||
|
||||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
||||
asyncTest "topic params":
|
||||
let params = TopicParams.init()
|
||||
params.validateParameters().tryGet()
|
||||
|
|
|
|||
|
|
@ -1,33 +1,17 @@
|
|||
# Nim-LibP2P
|
||||
# Copyright (c) 2023-2024 Status Research & Development GmbH
|
||||
# Licensed under either of
|
||||
# * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
|
||||
# * MIT license ([LICENSE-MIT](LICENSE-MIT))
|
||||
# at your option.
|
||||
# This file may not be copied, modified, or distributed except according to
|
||||
# those terms.
|
||||
|
||||
{.used.}
|
||||
|
||||
import std/[options, deques, sequtils, enumerate, algorithm, sets]
|
||||
import stew/byteutils
|
||||
import ../../libp2p/builders
|
||||
import ../../libp2p/errors
|
||||
import ../../libp2p/crypto/crypto
|
||||
import ../../libp2p/stream/bufferstream
|
||||
import ../../libp2p/protocols/pubsub/[pubsub, gossipsub, mcache, peertable]
|
||||
import ../../libp2p/protocols/pubsub/rpc/[message, messages, protobuf]
|
||||
import ../../libp2p/protocols/pubsub/[pubsub, gossipsub, mcache, mcache, peertable]
|
||||
import ../../libp2p/protocols/pubsub/rpc/[message, messages]
|
||||
import ../../libp2p/switch
|
||||
import ../../libp2p/muxers/muxer
|
||||
import ../../libp2p/protocols/pubsub/rpc/protobuf
|
||||
import utils
|
||||
import chronos
|
||||
import unittest2
|
||||
import ../helpers
|
||||
import chronos/ratelimit
|
||||
import metrics
|
||||
import chronicles
|
||||
|
||||
import ../../libp2p/protocols/pubsub/errors as pubsub_errors
|
||||
import ../helpers
|
||||
|
||||
proc noop(data: seq[byte]) {.async: (raises: [CancelledError, LPStreamError]).} =
|
||||
|
|
@ -41,54 +25,19 @@ suite "GossipSub Topic Membership Tests":
|
|||
|
||||
# Addition of Designed Test cases for 6. Topic Membership Tests: https://www.notion.so/Gossipsub-651e02d4d7894bb2ac1e4edb55f3192d
|
||||
|
||||
# Simulate the `SUBSCRIBE` event and check proper handling in the mesh and gossipsub structures
|
||||
asyncTest "handle SUBSCRIBE event":
|
||||
# Generalized setup function to initialize one or more topics
|
||||
proc setupGossipSub(
|
||||
topics: seq[string], numPeers: int
|
||||
): (TestGossipSub, seq[Connection]) =
|
||||
let gossipSub = TestGossipSub.init(newStandardSwitch())
|
||||
var conns = newSeq[Connection]()
|
||||
|
||||
# Ensure topic is correctly initialized
|
||||
let topic = "test-topic"
|
||||
for topic in topics:
|
||||
gossipSub.mesh[topic] = initHashSet[PubSubPeer]()
|
||||
gossipSub.topicParams[topic] = TopicParams.init()
|
||||
gossipSub.gossipsub[topic] = initHashSet[PubSubPeer]()
|
||||
# Initialize gossipsub for the topic
|
||||
|
||||
var conns = newSeq[Connection]()
|
||||
for i in 0 ..< 5:
|
||||
let conn = TestBufferStream.new(noop)
|
||||
conns &= conn
|
||||
let peerId = randomPeerId()
|
||||
conn.peerId = peerId
|
||||
let peer = gossipSub.getPubSubPeer(peerId)
|
||||
peer.sendConn = conn
|
||||
gossipSub.gossipsub[topic].incl(peer) # Ensure the topic is added to gossipsub
|
||||
|
||||
# Subscribe to the topic
|
||||
gossipSub.PubSub.subscribe(
|
||||
topic,
|
||||
proc(topic: string, data: seq[byte]): Future[void] {.async.} =
|
||||
discard
|
||||
,
|
||||
)
|
||||
|
||||
check gossipSub.topics.contains(topic) # Check if the topic is in topics
|
||||
check gossipSub.gossipsub[topic].len() > 0 # Check if topic added to gossipsub
|
||||
|
||||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
||||
# This test will simulate an UNSUBSCRIBE event and check if the topic is removed from the relevant data structures but remains in gossipsub
|
||||
asyncTest "handle UNSUBSCRIBE event":
|
||||
let gossipSub = TestGossipSub.init(newStandardSwitch())
|
||||
|
||||
# Ensure topic is initialized properly in all relevant data structures
|
||||
let topic = "test-topic"
|
||||
gossipSub.mesh[topic] = initHashSet[PubSubPeer]()
|
||||
gossipSub.topicParams[topic] = TopicParams.init()
|
||||
gossipSub.gossipsub[topic] = initHashSet[PubSubPeer]()
|
||||
# Initialize gossipsub for the topic
|
||||
|
||||
var conns = newSeq[Connection]()
|
||||
for i in 0 ..< 5:
|
||||
for i in 0 ..< numPeers:
|
||||
let conn = TestBufferStream.new(noop)
|
||||
conns &= conn
|
||||
let peerId = randomPeerId()
|
||||
|
|
@ -96,9 +45,16 @@ suite "GossipSub Topic Membership Tests":
|
|||
let peer = gossipSub.getPubSubPeer(peerId)
|
||||
peer.sendConn = conn
|
||||
gossipSub.gossipsub[topic].incl(peer)
|
||||
# Ensure peers are added to gossipsub for the topic
|
||||
|
||||
# Subscribe to the topic first
|
||||
return (gossipSub, conns)
|
||||
|
||||
# Wrapper function to initialize a single topic by converting it into a seq
|
||||
proc setupGossipSub(topic: string, numPeers: int): (TestGossipSub, seq[Connection]) =
|
||||
setupGossipSub(@[topic], numPeers)
|
||||
|
||||
# Helper function to subscribe to topics
|
||||
proc subscribeToTopics(gossipSub: TestGossipSub, topics: seq[string]) =
|
||||
for topic in topics:
|
||||
gossipSub.PubSub.subscribe(
|
||||
topic,
|
||||
proc(topic: string, data: seq[byte]): Future[void] {.async.} =
|
||||
|
|
@ -106,47 +62,67 @@ suite "GossipSub Topic Membership Tests":
|
|||
,
|
||||
)
|
||||
|
||||
# Now unsubscribe from the topic
|
||||
# Helper function to unsubscribe to topics
|
||||
proc unsubscribeFromTopics(gossipSub: TestGossipSub, topics: seq[string]) =
|
||||
for topic in topics:
|
||||
gossipSub.PubSub.unsubscribeAll(topic)
|
||||
|
||||
# Simulate the `SUBSCRIBE` to the topic and check proper handling in the mesh and gossipsub structures
|
||||
asyncTest "handle SUBSCRIBE to the topic":
|
||||
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() > 0
|
||||
|
||||
# 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
|
||||
asyncTest "handle UNSUBSCRIBE to the topic":
|
||||
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 # The topic should not be in topics
|
||||
check topic notin gossipSub.mesh # The topic should be removed from the mesh
|
||||
check topic notin gossipSub.topics
|
||||
check topic notin gossipSub.mesh
|
||||
check topic in gossipSub.gossipsub
|
||||
|
||||
# The topic should remain in gossipsub (for fanout)
|
||||
|
||||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
||||
# This test ensures that multiple topics can be subscribed to and unsubscribed from, with proper initialization of the topic structures.
|
||||
asyncTest "handle multiple SUBSCRIBE and UNSUBSCRIBE events":
|
||||
let gossipSub = TestGossipSub.init(newStandardSwitch())
|
||||
# Test subscribing and unsubscribing multiple topics
|
||||
asyncTest "handle SUBSCRIBE and UNSUBSCRIBE multiple topics":
|
||||
let topics = ["topic1", "topic2", "topic3"].toSeq()
|
||||
let (gossipSub, conns) = setupGossipSub(topics, 5)
|
||||
|
||||
let topics = ["topic1", "topic2", "topic3"]
|
||||
|
||||
var conns = newSeq[Connection]()
|
||||
for topic in topics:
|
||||
# Initialize all relevant structures before subscribing
|
||||
gossipSub.mesh[topic] = initHashSet[PubSubPeer]()
|
||||
gossipSub.topicParams[topic] = TopicParams.init()
|
||||
gossipSub.gossipsub[topic] = initHashSet[PubSubPeer]()
|
||||
# Initialize gossipsub for each topic
|
||||
|
||||
gossipSub.PubSub.subscribe(
|
||||
topic,
|
||||
proc(topic: string, data: seq[byte]): Future[void] {.async.} =
|
||||
discard
|
||||
,
|
||||
)
|
||||
# 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() >= 0
|
||||
|
||||
# Now unsubscribe from all topics
|
||||
for topic in topics:
|
||||
gossipSub.PubSub.unsubscribeAll(topic)
|
||||
# Unsubscribe from all topics
|
||||
unsubscribeFromTopics(gossipSub, topics)
|
||||
|
||||
# Ensure topics are removed from topics and mesh, but still present in gossipsub
|
||||
for topic in topics:
|
||||
|
|
@ -157,10 +133,10 @@ suite "GossipSub Topic Membership Tests":
|
|||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
||||
# This test ensures that the number of subscriptions does not exceed the limit set in the GossipSub parameters
|
||||
# Test ensuring that the number of subscriptions does not exceed the limit set in the GossipSub parameters
|
||||
asyncTest "subscription limit test":
|
||||
let gossipSub = TestGossipSub.init(newStandardSwitch())
|
||||
gossipSub.topicsHigh = 10 # Set a limit for the number of subscriptions
|
||||
gossipSub.topicsHigh = 10
|
||||
|
||||
var conns = newSeq[Connection]()
|
||||
for i in 0 .. gossipSub.topicsHigh + 5:
|
||||
|
|
@ -262,3 +238,40 @@ suite "GossipSub Topic Membership Tests":
|
|||
|
||||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
||||
asyncTest "handle JOIN topic with update mesh":
|
||||
let topic = "test-topic-join"
|
||||
|
||||
# Initialize the GossipSub system and simulate peer connections
|
||||
let (gossipSub, conns) = setupGossipSub(topic, 5)
|
||||
|
||||
# Subscribe to the topic
|
||||
subscribeToTopics(gossipSub, @[topic])
|
||||
|
||||
# Check that peers are added to the mesh after subscribing
|
||||
check gossipSub.mesh[topic].len() > 0
|
||||
check gossipSub.topics.contains(topic)
|
||||
|
||||
# Clean up by closing connections and stopping the gossipSub switch
|
||||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
||||
asyncTest "handle LEAVE topic with update mesh":
|
||||
let topic = "test-topic-leave"
|
||||
|
||||
# Initialize the GossipSub system and simulate peer connections
|
||||
let (gossipSub, conns) = setupGossipSub(topic, 5)
|
||||
|
||||
# Subscribe to the topic
|
||||
subscribeToTopics(gossipSub, @[topic])
|
||||
|
||||
# Now unsubscribe from the topic
|
||||
unsubscribeFromTopics(gossipSub, @[topic])
|
||||
|
||||
# Check that peers are removed from the mesh but still present in gossipsub
|
||||
check topic notin gossipSub.mesh
|
||||
check topic in gossipSub.gossipsub
|
||||
|
||||
# Clean up by closing connections and stopping the gossipSub switch
|
||||
await allFuturesThrowing(conns.mapIt(it.close()))
|
||||
await gossipSub.switch.stop()
|
||||
|
|
|
|||
Loading…
Reference in New Issue