#[cfg(test)]
mod tests {
use super::super::*;
fn build_and_inject_nodes(
peer_no: usize,
topics: Vec<String>,
to_subscribe: bool,
) -> (Gossipsub, Vec<PeerId>, Vec<TopicHash>) {
let gs_config = GossipsubConfig::default();
let mut gs: Gossipsub = Gossipsub::new(PeerId::random(), gs_config);
let mut topic_hashes = vec![];
for t in topics {
let topic = Topic::new(t);
gs.subscribe(topic.clone());
topic_hashes.push(topic.no_hash().clone());
}
let mut peers = vec![];
let dummy_connected_point = ConnectedPoint::Dialer {
address: "/ip4/0.0.0.0/tcp/0".parse().unwrap(),
};
for _ in 0..peer_no {
let peer = PeerId::random();
peers.push(peer.clone());
<Gossipsub as NetworkBehaviour>::inject_connected(
&mut gs,
peer.clone(),
dummy_connected_point.clone(),
);
if to_subscribe {
gs.handle_received_subscriptions(
&topic_hashes
.iter()
.cloned()
.map(|t| GossipsubSubscription {
action: GossipsubSubscriptionAction::Subscribe,
topic_hash: t,
})
.collect::<Vec<_>>(),
&peer,
);
};
}
return (gs, peers, topic_hashes);
}
#[test]
fn test_subscribe() {
let subscribe_topic = vec![String::from("test_subscribe")];
let (gs, _, topic_hashes) = build_and_inject_nodes(20, subscribe_topic, true);
assert!(
gs.mesh.get(&topic_hashes[0]).is_some(),
"Subscribe should add a new entry to the mesh[topic] hashmap"
);
let subscriptions =
gs.events
.iter()
.fold(vec![], |mut collected_subscriptions, e| match e {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
for s in &event.subscriptions {
match s.action {
GossipsubSubscriptionAction::Subscribe => {
collected_subscriptions.push(s.clone())
}
_ => {}
};
}
collected_subscriptions
}
_ => collected_subscriptions,
});
assert!(
subscriptions.len() == 20,
"Should send a subscription to all known peers"
);
}
#[test]
fn test_unsubscribe() {
let topic_strings = vec![String::from("topic1"), String::from("topic2")];
let topics = topic_strings
.iter()
.map(|t| Topic::new(t.clone()))
.collect::<Vec<Topic>>();
let (mut gs, _, topic_hashes) = build_and_inject_nodes(20, topic_strings, true);
for topic_hash in &topic_hashes {
assert!(
gs.topic_peers.get(&topic_hash).is_some(),
"Topic_peers contain a topic entry"
);
assert!(
gs.mesh.get(&topic_hash).is_some(),
"mesh should contain a topic entry"
);
}
assert!(
gs.unsubscribe(topics[0].clone()),
"should be able to unsubscribe successfully from each topic",
);
assert!(
gs.unsubscribe(topics[1].clone()),
"should be able to unsubscribe successfully from each topic",
);
let subscriptions =
gs.events
.iter()
.fold(vec![], |mut collected_subscriptions, e| match e {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
for s in &event.subscriptions {
match s.action {
GossipsubSubscriptionAction::Unsubscribe => {
collected_subscriptions.push(s.clone())
}
_ => {}
};
}
collected_subscriptions
}
_ => collected_subscriptions,
});
assert!(
subscriptions.len() == 40,
"Should send an unsubscribe event to all known peers"
);
for topic_hash in &topic_hashes {
assert!(
gs.mesh.get(&topic_hash).is_none(),
"All topics should have been removed from the mesh"
);
}
}
#[test]
fn test_join() {
let topic_strings = vec![String::from("topic1"), String::from("topic2")];
let topics = topic_strings
.iter()
.map(|t| Topic::new(t.clone()))
.collect::<Vec<Topic>>();
let (mut gs, _, topic_hashes) = build_and_inject_nodes(20, topic_strings, true);
assert!(
gs.unsubscribe(topics[0].clone()),
"should be able to unsubscribe successfully"
);
assert!(
gs.unsubscribe(topics[1].clone()),
"should be able to unsubscribe successfully"
);
assert!(
gs.subscribe(topics[0].clone()),
"should be able to subscribe successfully"
);
assert!(
gs.mesh.get(&topic_hashes[0]).unwrap().len() == 6,
"Should have added 6 nodes to the mesh"
);
let graft_messages =
gs.control_pool
.iter()
.fold(vec![], |mut collected_grafts, (_, controls)| {
for c in controls.iter() {
match c {
GossipsubControlAction::Graft { topic_hash: _ } => {
collected_grafts.push(c.clone())
}
_ => {}
}
}
collected_grafts
});
assert_eq!(
graft_messages.len(),
6,
"There should be 6 grafts messages sent to peers"
);
gs.fanout.insert(topic_hashes[1].clone(), vec![]);
let new_peers = vec![];
for _ in 0..3 {
let fanout_peers = gs.fanout.get_mut(&topic_hashes[1]).unwrap();
fanout_peers.push(PeerId::random());
}
gs.subscribe(topics[1].clone());
assert!(
gs.mesh.get(&topic_hashes[1]).unwrap().len() == 6,
"Should have added 6 nodes to the mesh"
);
let mesh_peers = gs.mesh.get(&topic_hashes[1]).unwrap();
for new_peer in new_peers {
assert!(
mesh_peers.contains(new_peer),
"Fanout peer should be included in the mesh"
);
}
let graft_messages =
gs.control_pool
.iter()
.fold(vec![], |mut collected_grafts, (_, controls)| {
for c in controls.iter() {
match c {
GossipsubControlAction::Graft { topic_hash: _ } => {
collected_grafts.push(c.clone())
}
_ => {}
}
}
collected_grafts
});
assert!(
graft_messages.len() == 12,
"There should be 12 grafts messages sent to peers"
);
}
#[test]
fn test_publish() {
let publish_topic = String::from("test_publish");
let (mut gs, _, topic_hashes) =
build_and_inject_nodes(20, vec![publish_topic.clone()], true);
assert!(
gs.mesh.get(&topic_hashes[0]).is_some(),
"Subscribe should add a new entry to the mesh[topic] hashmap"
);
let publish_data = vec![0; 42];
gs.publish(&Topic::new(publish_topic), publish_data);
let publishes = gs
.events
.iter()
.fold(vec![], |mut collected_publish, e| match e {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
for s in &event.messages {
collected_publish.push(s.clone());
}
collected_publish
}
_ => collected_publish,
});
let msg_id =
(gs.config.message_id_fn)(&publishes.first().expect("Should contain > 0 entries"));
assert!(
publishes.len() == 20,
"Should send a publish message to all known peers"
);
assert!(
gs.mcache.get(&msg_id).is_some(),
"Message cache should contain published message"
);
assert!(
gs.received.get(&msg_id).is_some(),
"Received cache should contain published message"
);
}
#[test]
fn test_fanout() {
let fanout_topic = String::from("test_fanout");
let (mut gs, _, topic_hashes) =
build_and_inject_nodes(20, vec![fanout_topic.clone()], true);
assert!(
gs.mesh.get(&topic_hashes[0]).is_some(),
"Subscribe should add a new entry to the mesh[topic] hashmap"
);
assert!(
gs.unsubscribe(Topic::new(fanout_topic.clone())),
"should be able to unsubscribe successfully from topic"
);
let publish_data = vec![0; 42];
gs.publish(&Topic::new(fanout_topic.clone()), publish_data);
assert_eq!(
gs.fanout
.get(&TopicHash::from_raw(fanout_topic.clone()))
.unwrap()
.len(),
gs.config.mesh_n,
"Fanout should contain `mesh_n` peers for fanout topic"
);
let publishes = gs
.events
.iter()
.fold(vec![], |mut collected_publish, e| match e {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
for s in &event.messages {
collected_publish.push(s.clone());
}
collected_publish
}
_ => collected_publish,
});
let msg_id =
(gs.config.message_id_fn)(&publishes.first().expect("Should contain > 0 entries"));
assert_eq!(
publishes.len(),
gs.config.mesh_n,
"Should send a publish message to `mesh_n` fanout peers"
);
assert!(
gs.mcache.get(&msg_id).is_some(),
"Message cache should contain published message"
);
assert!(
gs.received.get(&msg_id).is_some(),
"Received cache should contain published message"
);
}
#[test]
fn test_inject_connected() {
let (gs, peers, topic_hashes) = build_and_inject_nodes(
20,
vec![String::from("topic1"), String::from("topic2")],
true,
);
let send_events: Vec<&NetworkBehaviourAction<Arc<GossipsubRpc>, GossipsubEvent>> = gs
.events
.iter()
.filter(|e| match e {
NetworkBehaviourAction::SendEvent {
peer_id: _,
event: _,
} => true,
_ => false,
})
.collect();
for sevent in send_events.clone() {
match sevent {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
assert!(
event.subscriptions.len() == 2,
"There should be two subscriptions sent to each peer (1 for each topic)."
);
}
_ => {}
};
}
assert!(
send_events.len() == 20,
"There should be a subscription event sent to each peer."
);
for peer in peers {
let known_topics = gs.peer_topics.get(&peer).unwrap();
assert!(
known_topics == &topic_hashes,
"The topics for each node should all topics"
);
}
}
#[test]
fn test_handle_received_subscriptions() {
let topics = vec!["topic1", "topic2", "topic3", "topic4"]
.iter()
.map(|&t| String::from(t))
.collect();
let (mut gs, peers, topic_hashes) = build_and_inject_nodes(20, topics, false);
let mut subscriptions = topic_hashes[..3]
.iter()
.map(|topic_hash| GossipsubSubscription {
action: GossipsubSubscriptionAction::Subscribe,
topic_hash: topic_hash.clone(),
})
.collect::<Vec<GossipsubSubscription>>();
subscriptions.push(GossipsubSubscription {
action: GossipsubSubscriptionAction::Unsubscribe,
topic_hash: topic_hashes[topic_hashes.len() - 1].clone(),
});
let unknown_peer = PeerId::random();
gs.handle_received_subscriptions(&subscriptions, &peers[0]);
gs.handle_received_subscriptions(&subscriptions, &peers[1]);
gs.handle_received_subscriptions(&subscriptions, &unknown_peer);
let peer_topics = gs.peer_topics.get(&peers[0]).unwrap().clone();
assert!(
peer_topics == topic_hashes[..3].to_vec(),
"First peer should be subscribed to three topics"
);
let peer_topics = gs.peer_topics.get(&peers[1]).unwrap().clone();
assert!(
peer_topics == topic_hashes[..3].to_vec(),
"Second peer should be subscribed to three topics"
);
assert!(
gs.peer_topics.get(&unknown_peer).is_none(),
"Unknown peer should not have been added"
);
for topic_hash in topic_hashes[..3].iter() {
let topic_peers = gs.topic_peers.get(topic_hash).unwrap().clone();
assert!(
topic_peers == peers[..2].to_vec(),
"Two peers should be added to the first three topics"
);
}
gs.handle_received_subscriptions(
&vec![GossipsubSubscription {
action: GossipsubSubscriptionAction::Unsubscribe,
topic_hash: topic_hashes[0].clone(),
}],
&peers[0],
);
let peer_topics = gs.peer_topics.get(&peers[0]).unwrap().clone();
assert!(
peer_topics == topic_hashes[1..3].to_vec(),
"Peer should be subscribed to two topics"
);
let topic_peers = gs.topic_peers.get(&topic_hashes[0]).unwrap().clone(); assert!(
topic_peers == peers[1..2].to_vec(),
"Only the second peers should be in the first topic"
);
}
#[test]
fn test_get_random_peers() {
let gs_config = GossipsubConfig::default();
let mut gs: Gossipsub = Gossipsub::new(PeerId::random(), gs_config);
let topic_hash = Topic::new("Test".into()).no_hash().clone();
let mut peers = vec![];
for _ in 0..20 {
peers.push(PeerId::random())
}
gs.topic_peers.insert(topic_hash.clone(), peers.clone());
let random_peers =
Gossipsub::get_random_peers(&gs.topic_peers, &topic_hash, 5, { |_| true });
assert!(random_peers.len() == 5, "Expected 5 peers to be returned");
let random_peers =
Gossipsub::get_random_peers(&gs.topic_peers, &topic_hash, 30, { |_| true });
assert!(random_peers.len() == 20, "Expected 20 peers to be returned");
assert!(random_peers == peers, "Expected no shuffling");
let random_peers =
Gossipsub::get_random_peers(&gs.topic_peers, &topic_hash, 20, { |_| true });
assert!(random_peers.len() == 20, "Expected 20 peers to be returned");
assert!(random_peers == peers, "Expected no shuffling");
let random_peers =
Gossipsub::get_random_peers(&gs.topic_peers, &topic_hash, 0, { |_| true });
assert!(random_peers.len() == 0, "Expected 0 peers to be returned");
let random_peers =
Gossipsub::get_random_peers(&gs.topic_peers, &topic_hash, 5, { |_| false });
assert!(random_peers.len() == 0, "Expected 0 peers to be returned");
let random_peers =
Gossipsub::get_random_peers(&gs.topic_peers, &topic_hash, 10, {
|peer| peers.contains(peer)
});
assert!(random_peers.len() == 10, "Expected 10 peers to be returned");
}
#[test]
fn test_handle_iwant_msg_cached() {
let (mut gs, peers, _) = build_and_inject_nodes(20, Vec::new(), true);
let id = gs.config.message_id_fn;
let message = GossipsubMessage {
source: peers[11].clone(),
data: vec![1, 2, 3, 4],
sequence_number: 1u64,
topics: Vec::new(),
};
let msg_id = id(&message);
gs.mcache.put(message.clone());
gs.handle_iwant(&peers[7], vec![msg_id.clone()]);
let sent_messages = gs
.events
.iter()
.fold(vec![], |mut collected_messages, e| match e {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
for c in &event.messages {
collected_messages.push(c.clone())
}
collected_messages
}
_ => collected_messages,
});
assert!(
sent_messages.iter().any(|msg| id(msg) == msg_id),
"Expected the cached message to be sent to an IWANT peer"
);
}
#[test]
fn test_handle_iwant_msg_cached_shifted() {
let (mut gs, peers, _) = build_and_inject_nodes(20, Vec::new(), true);
let id = gs.config.message_id_fn;
for shift in 1..10 {
let message = GossipsubMessage {
source: peers[11].clone(),
data: vec![1, 2, 3, 4],
sequence_number: shift,
topics: Vec::new(),
};
let msg_id = id(&message);
gs.mcache.put(message.clone());
for _ in 0..shift {
gs.mcache.shift();
}
gs.handle_iwant(&peers[7], vec![msg_id.clone()]);
let message_exists = gs.events.iter().any(|e| match e {
NetworkBehaviourAction::SendEvent { peer_id: _, event } => {
event.messages.iter().any(|msg| id(msg) == msg_id)
}
_ => false,
});
if shift < 5 {
assert!(
message_exists,
"Expected the cached message to be sent to an IWANT peer before 5 shifts"
);
} else {
assert!(
!message_exists,
"Expected the cached message to not be sent to an IWANT peer after 5 shifts"
);
}
}
}
#[test]
fn test_handle_iwant_msg_not_cached() {
let (mut gs, peers, _) = build_and_inject_nodes(20, Vec::new(), true);
let events_before = gs.events.len();
gs.handle_iwant(&peers[7], vec![MessageId(String::from("unknown id"))]);
let events_after = gs.events.len();
assert_eq!(
events_before, events_after,
"Expected event count to stay the same"
);
}
#[test]
fn test_handle_ihave_subscribed_and_msg_not_cached() {
let (mut gs, peers, topic_hashes) =
build_and_inject_nodes(20, vec![String::from("topic1")], true);
gs.handle_ihave(
&peers[7],
vec![(
topic_hashes[0].clone(),
vec![MessageId(String::from("unknown id"))],
)],
);
let iwant_exists = match gs.control_pool.get(&peers[7]) {
Some(controls) => controls.iter().any(|c| match c {
GossipsubControlAction::IWant { message_ids } => message_ids
.iter()
.any(|m| *m.0 == String::from("unknown id")),
_ => false,
}),
_ => false,
};
assert!(
iwant_exists,
"Expected to send an IWANT control message for unkown message id"
);
}
#[test]
fn test_handle_ihave_subscribed_and_msg_cached() {
let (mut gs, peers, topic_hashes) =
build_and_inject_nodes(20, vec![String::from("topic1")], true);
let msg_id = MessageId(String::from("known id"));
gs.received.put(msg_id.clone(), ());
let events_before = gs.events.len();
gs.handle_ihave(&peers[7], vec![(topic_hashes[0].clone(), vec![msg_id])]);
let events_after = gs.events.len();
assert_eq!(
events_before, events_after,
"Expected event count to stay the same"
)
}
#[test]
fn test_handle_ihave_not_subscribed() {
let (mut gs, peers, _) = build_and_inject_nodes(20, vec![], true);
let events_before = gs.events.len();
gs.handle_ihave(
&peers[7],
vec![(
TopicHash::from_raw(String::from("unsubscribed topic")),
vec![MessageId(String::from("irrelevant id"))],
)],
);
let events_after = gs.events.len();
assert_eq!(
events_before, events_after,
"Expected event count to stay the same"
)
}
#[test]
fn test_handle_graft_is_subscribed() {
let (mut gs, peers, topic_hashes) =
build_and_inject_nodes(20, vec![String::from("topic1")], true);
gs.handle_graft(&peers[7], topic_hashes.clone());
assert!(
gs.mesh.get(&topic_hashes[0]).unwrap().contains(&peers[7]),
"Expected peer to have been added to mesh"
);
}
#[test]
fn test_handle_graft_is_not_subscribed() {
let (mut gs, peers, topic_hashes) =
build_and_inject_nodes(20, vec![String::from("topic1")], true);
gs.handle_graft(
&peers[7],
vec![TopicHash::from_raw(String::from("unsubscribed topic"))],
);
assert!(
gs.mesh.get(&topic_hashes[0]).unwrap().contains(&peers[7]),
"Expected peer to have been added to mesh"
);
}
#[test]
fn test_handle_graft_multiple_topics() {
let topics: Vec<String> = vec!["topic1", "topic2", "topic3", "topic4"]
.iter()
.map(|&t| String::from(t))
.collect();
let (mut gs, peers, topic_hashes) = build_and_inject_nodes(20, topics.clone(), true);
let mut their_topics = topic_hashes.clone();
their_topics.pop();
gs.leave(&their_topics[2]);
gs.handle_graft(&peers[7], their_topics.clone());
for i in 0..2 {
assert!(
gs.mesh.get(&topic_hashes[i]).unwrap().contains(&peers[7]),
"Expected peer to be in the mesh for the first 2 topics"
);
}
assert!(
gs.mesh.get(&topic_hashes[2]).is_none(),
"Expected the second topic to not be in the mesh"
);
}
#[test]
fn test_handle_prune_peer_in_mesh() {
let (mut gs, peers, topic_hashes) =
build_and_inject_nodes(20, vec![String::from("topic1")], true);
gs.mesh.insert(topic_hashes[0].clone(), peers.clone());
assert!(
gs.mesh.get(&topic_hashes[0]).unwrap().contains(&peers[7]),
"Expected peer to be in mesh"
);
gs.handle_prune(&peers[7], topic_hashes.clone());
assert!(
!gs.mesh.get(&topic_hashes[0]).unwrap().contains(&peers[7]),
"Expected peer to be removed from mesh"
);
}
}