fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-04 23:22:16 +00:00
Code Issues Projects Releases Wiki Activity

protocols/kad: Fix tests + minor fix in mdns and noise (#1320)

Browse Source 
* misc/mdns: Fix missleading error message

* protocols/noise: Remove unneeded tokio-io import

* protocols/kad: Update tests to use stable futures
This commit is contained in:
Max Inden 2019-11-28 16:12:02 +01:00 committed by GitHub
parent 8d22e98abc
commit 26f58d20a8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 150 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
misc/mdns/src/behaviour.rs
View File

@ -236,7 +236,7 @@ where
} }
}, },
Poll::Pending => (), Poll::Pending => (),
Poll::Ready(Err(err)) => warn!("tokio timer has errored: {:?}", err), Poll::Ready(Err(err)) => warn!("timer has errored: {:?}", err),
} }
} }

216
protocols/kad/src/behaviour/test.rs
View File

@ -25,7 +25,11 @@ use super::*;
use crate::K_VALUE; use crate::K_VALUE;
use crate::kbucket::Distance; use crate::kbucket::Distance;
use crate::record::store::MemoryStore; use crate::record::store::MemoryStore;
use futures::future; use futures::{
prelude::*,
executor::block_on,
future::poll_fn,
};
use libp2p_core::{ use libp2p_core::{
PeerId, PeerId,
Transport, Transport,
@ -42,7 +46,6 @@ use libp2p_yamux as yamux;
use quickcheck::*; use quickcheck::*;
use rand::{Rng, random, thread_rng}; use rand::{Rng, random, thread_rng};
use std::{collections::{HashSet, HashMap}, io, num::NonZeroUsize, u64}; use std::{collections::{HashSet, HashMap}, io, num::NonZeroUsize, u64};
use tokio::runtime::current_thread;
use multihash::{Multihash, Hash::SHA2256}; use multihash::{Multihash, Hash::SHA2256};
type TestSwarm = Swarm< type TestSwarm = Swarm<
@ -120,27 +123,30 @@ fn bootstrap() {
let expected_known = swarm_ids.iter().skip(1).cloned().collect::<HashSet<_>>(); let expected_known = swarm_ids.iter().skip(1).cloned().collect::<HashSet<_>>();
// Run test // Run test
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for (i, swarm) in swarms.iter_mut().enumerate() { for (i, swarm) in swarms.iter_mut().enumerate() {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::BootstrapResult(Ok(ok)))) => { Poll::Ready(Some(Ok(KademliaEvent::BootstrapResult(Ok(ok))))) => {
assert_eq!(i, 0); assert_eq!(i, 0);
assert_eq!(ok.peer, swarm_ids[0]); assert_eq!(ok.peer, swarm_ids[0]);
let known = swarm.kbuckets.iter() let known = swarm.kbuckets.iter()
.map(|e| e.node.key.preimage().clone()) .map(|e| e.node.key.preimage().clone())
.collect::<HashSet<_>>(); .collect::<HashSet<_>>();
assert_eq!(expected_known, known); assert_eq!(expected_known, known);
return Ok(Async::Ready(())); return Poll::Ready(())
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
let mut rng = thread_rng(); let mut rng = thread_rng();
@ -175,27 +181,30 @@ fn query_iter() {
expected_distances.sort(); expected_distances.sort();
// Run test // Run test
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for (i, swarm) in swarms.iter_mut().enumerate() { for (i, swarm) in swarms.iter_mut().enumerate() {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::GetClosestPeersResult(Ok(ok)))) => { Poll::Ready(Some(Ok(KademliaEvent::GetClosestPeersResult(Ok(ok))))) => {
assert_eq!(&ok.key[..], search_target.as_bytes()); assert_eq!(&ok.key[..], search_target.as_bytes());
assert_eq!(swarm_ids[i], expected_swarm_id); assert_eq!(swarm_ids[i], expected_swarm_id);
assert_eq!(swarm.queries.size(), 0); assert_eq!(swarm.queries.size(), 0);
assert!(expected_peer_ids.iter().all(|p| ok.peers.contains(p))); assert!(expected_peer_ids.iter().all(|p| ok.peers.contains(p)));
let key = kbucket::Key::new(ok.key); let key = kbucket::Key::new(ok.key);
assert_eq!(expected_distances, distances(&key, ok.peers)); assert_eq!(expected_distances, distances(&key, ok.peers));
return Ok(Async::Ready(())); return Poll::Ready(());
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
let mut rng = thread_rng(); let mut rng = thread_rng();
@ -220,24 +229,27 @@ fn unresponsive_not_returned_direct() {
let search_target = PeerId::random(); let search_target = PeerId::random();
swarms[0].get_closest_peers(search_target.clone()); swarms[0].get_closest_peers(search_target.clone());
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::GetClosestPeersResult(Ok(ok)))) => { Poll::Ready(Some(Ok(KademliaEvent::GetClosestPeersResult(Ok(ok))))) => {
assert_eq!(&ok.key[..], search_target.as_bytes()); assert_eq!(&ok.key[..], search_target.as_bytes());
assert_eq!(ok.peers.len(), 0); assert_eq!(ok.peers.len(), 0);
return Ok(Async::Ready(())); return Poll::Ready(());
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
#[test] #[test]
@ -261,25 +273,28 @@ fn unresponsive_not_returned_indirect() {
let search_target = PeerId::random(); let search_target = PeerId::random();
swarms[1].get_closest_peers(search_target.clone()); swarms[1].get_closest_peers(search_target.clone());
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::GetClosestPeersResult(Ok(ok)))) => { Poll::Ready(Some(Ok(KademliaEvent::GetClosestPeersResult(Ok(ok))))) => {
assert_eq!(&ok.key[..], search_target.as_bytes()); assert_eq!(&ok.key[..], search_target.as_bytes());
assert_eq!(ok.peers.len(), 1); assert_eq!(ok.peers.len(), 1);
assert_eq!(ok.peers[0], first_peer_id); assert_eq!(ok.peers[0], first_peer_id);
return Ok(Async::Ready(())); return Poll::Ready(());
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
#[test] #[test]
@ -294,30 +309,33 @@ fn get_record_not_found() {
let target_key = record::Key::from(Multihash::random(SHA2256)); let target_key = record::Key::from(Multihash::random(SHA2256));
swarms[0].get_record(&target_key, Quorum::One); swarms[0].get_record(&target_key, Quorum::One);
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::GetRecordResult(Err(e)))) => { Poll::Ready(Some(Ok(KademliaEvent::GetRecordResult(Err(e))))) => {
if let GetRecordError::NotFound { key, closest_peers, } = e { if let GetRecordError::NotFound { key, closest_peers, } = e {
assert_eq!(key, target_key); assert_eq!(key, target_key);
assert_eq!(closest_peers.len(), 2); assert_eq!(closest_peers.len(), 2);
assert!(closest_peers.contains(&swarm_ids[1])); assert!(closest_peers.contains(&swarm_ids[1]));
assert!(closest_peers.contains(&swarm_ids[2])); assert!(closest_peers.contains(&swarm_ids[2]));
return Ok(Async::Ready(())); return Poll::Ready(());
} else { } else {
panic!("Unexpected error result: {:?}", e); panic!("Unexpected error result: {:?}", e);
} }
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
#[test] #[test]
@ -351,14 +369,14 @@ fn put_record() {
// The accumulated results for one round of publishing. // The accumulated results for one round of publishing.
let mut results = Vec::new(); let mut results = Vec::new();
current_thread::run( block_on(
future::poll_fn(move || loop { poll_fn(move |ctx| loop {
// Poll all swarms until they are "NotReady". // Poll all swarms until they are "Pending".
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::PutRecordResult(res))) | Poll::Ready(Some(Ok(KademliaEvent::PutRecordResult(res)))) |
Async::Ready(Some(KademliaEvent::RepublishRecordResult(res))) => { Poll::Ready(Some(Ok(KademliaEvent::RepublishRecordResult(res)))) => {
match res { match res {
Err(e) => panic!(e), Err(e) => panic!(e),
Ok(ok) => { Ok(ok) => {
@ -368,16 +386,18 @@ fn put_record() {
} }
} }
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
// All swarms are NotReady and not enough results have been collected // All swarms are Pending and not enough results have been collected
// so far, thus wait to be polled again for further progress. // so far, thus wait to be polled again for further progress.
if results.len() != records.len() { if results.len() != records.len() {
return Ok(Async::NotReady) return Poll::Pending
} }
// Consume the results, checking that each record was replicated // Consume the results, checking that each record was replicated
@ -422,7 +442,7 @@ fn put_record() {
} }
assert_eq!(swarms[0].store.records().count(), 0); assert_eq!(swarms[0].store.records().count(), 0);
// All records have been republished, thus the test is complete. // All records have been republished, thus the test is complete.
return Ok(Async::Ready(())); return Poll::Ready(());
} }
// Tell the replication job to republish asap. // Tell the replication job to republish asap.
@ -449,24 +469,27 @@ fn get_value() {
swarms[1].store.put(record.clone()).unwrap(); swarms[1].store.put(record.clone()).unwrap();
swarms[0].get_record(&record.key, Quorum::One); swarms[0].get_record(&record.key, Quorum::One);
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::GetRecordResult(Ok(ok)))) => { Poll::Ready(Some(Ok(KademliaEvent::GetRecordResult(Ok(ok))))) => {
assert_eq!(ok.records.len(), 1); assert_eq!(ok.records.len(), 1);
assert_eq!(ok.records.first(), Some(&record)); assert_eq!(ok.records.first(), Some(&record));
return Ok(Async::Ready(())); return Poll::Ready(());
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
#[test] #[test]
@ -485,23 +508,26 @@ fn get_value_many() {
let quorum = Quorum::N(NonZeroUsize::new(num_results).unwrap()); let quorum = Quorum::N(NonZeroUsize::new(num_results).unwrap());
swarms[0].get_record(&record.key, quorum); swarms[0].get_record(&record.key, quorum);
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::GetRecordResult(Ok(ok)))) => { Poll::Ready(Some(Ok(KademliaEvent::GetRecordResult(Ok(ok))))) => {
assert_eq!(ok.records.len(), num_results); assert_eq!(ok.records.len(), num_results);
assert_eq!(ok.records.first(), Some(&record)); assert_eq!(ok.records.first(), Some(&record));
return Ok(Async::Ready(())); return Poll::Ready(());
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
Ok(Async::NotReady) Poll::Pending
})) })
)
} }
#[test] #[test]
@ -529,14 +555,14 @@ fn add_provider() {
swarms[0].start_providing(k.clone()); swarms[0].start_providing(k.clone());
} }
current_thread::run( block_on(
future::poll_fn(move || loop { poll_fn(move |ctx| loop {
// Poll all swarms until they are "NotReady". // Poll all swarms until they are "Pending".
for swarm in &mut swarms { for swarm in &mut swarms {
loop { loop {
match swarm.poll().unwrap() { match swarm.poll_next_unpin(ctx) {
Async::Ready(Some(KademliaEvent::StartProvidingResult(res))) | Poll::Ready(Some(Ok(KademliaEvent::StartProvidingResult(res)))) |
Async::Ready(Some(KademliaEvent::RepublishProviderResult(res))) => { Poll::Ready(Some(Ok(KademliaEvent::RepublishProviderResult(res)))) => {
match res { match res {
Err(e) => panic!(e), Err(e) => panic!(e),
Ok(ok) => { Ok(ok) => {
@ -545,8 +571,10 @@ fn add_provider() {
} }
} }
} }
Async::Ready(_) => (), // Ignore any other event.
Async::NotReady => break, Poll::Ready(Some(Ok(_))) => (),
e @ Poll::Ready(_) => panic!("Unexpected return value: {:?}", e),
Poll::Pending => break,
} }
} }
} }
@ -559,7 +587,7 @@ fn add_provider() {
if !published { if !published {
// Still waiting for all requests to be sent for one round // Still waiting for all requests to be sent for one round
// of publishing. // of publishing.
return Ok(Async::NotReady) return Poll::Pending
} }
// A round of publishing is complete. Consume the results, checking that // A round of publishing is complete. Consume the results, checking that
@ -578,7 +606,7 @@ fn add_provider() {
if actual.len() != replication_factor.get() { if actual.len() != replication_factor.get() {
// Still waiting for some nodes to process the request. // Still waiting for some nodes to process the request.
results.push(key); results.push(key);
return Ok(Async::NotReady) return Poll::Pending
} }
let mut expected = swarm_ids.clone().split_off(1); let mut expected = swarm_ids.clone().split_off(1);
@ -608,7 +636,7 @@ fn add_provider() {
} }
assert_eq!(swarms[0].store.provided().count(), 0); assert_eq!(swarms[0].store.provided().count(), 0);
// All records have been republished, thus the test is complete. // All records have been republished, thus the test is complete.
return Ok(Async::Ready(())); return Poll::Ready(());
} }
// Initiate the second round of publishing by telling the // Initiate the second round of publishing by telling the
@ -636,12 +664,12 @@ fn exceed_jobs_max_queries() {
assert_eq!(swarms[0].queries.size(), num); assert_eq!(swarms[0].queries.size(), num);
current_thread::run( block_on(
future::poll_fn(move || { poll_fn(move |ctx| {
for _ in 0 .. num { for _ in 0 .. num {
// There are no other nodes, so the queries finish instantly. // There are no other nodes, so the queries finish instantly.
if let Ok(Async::Ready(Some(e))) = swarms[0].poll() { if let Poll::Ready(Some(e)) = swarms[0].poll_next_unpin(ctx) {
if let KademliaEvent::BootstrapResult(r) = e { if let Ok(KademliaEvent::BootstrapResult(r)) = e {
assert!(r.is_ok(), "Unexpected error") assert!(r.is_ok(), "Unexpected error")
} else { } else {
panic!("Unexpected event: {:?}", e) panic!("Unexpected event: {:?}", e)
@ -650,7 +678,7 @@ fn exceed_jobs_max_queries() {
panic!("Expected event") panic!("Expected event")
} }
} }
Ok(Async::Ready(())) Poll::Ready(())
})) })
)
} }

54
protocols/kad/src/jobs.rs
View File

@ -326,6 +326,7 @@ impl AddProviderJob {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use crate::record::store::MemoryStore; use crate::record::store::MemoryStore;
use futures::{executor::block_on, future::poll_fn};
use quickcheck::*; use quickcheck::*;
use rand::Rng; use rand::Rng;
use super::*; use super::*;
@ -362,20 +363,20 @@ mod tests {
for r in records { for r in records {
let _ = store.put(r); let _ = store.put(r);
} }
// Polling with an instant beyond the deadline for the next run
// is guaranteed to run the job, without the job needing to poll the `Delay` block_on(poll_fn(|ctx| {
// and thus without needing to run `poll` in the context of a task let now = Instant::now() + job.inner.interval;
// for testing purposes. // All (non-expired) records in the store must be yielded by the job.
let now = Instant::now() + job.inner.interval; for r in store.records().map(|r| r.into_owned()).collect::<Vec<_>>() {
// All (non-expired) records in the store must be yielded by the job. if !r.is_expired(now) {
for r in store.records().map(|r| r.into_owned()).collect::<Vec<_>>() { assert_eq!(job.poll(ctx, &mut store, now), Poll::Ready(r));
if !r.is_expired(now) { assert!(job.is_running());
assert_eq!(job.poll(&mut store, now), Poll::Ready(r)); }
assert!(job.is_running());
} }
} assert_eq!(job.poll(ctx, &mut store, now), Poll::Pending);
assert_eq!(job.poll(&mut store, now), Poll::Pending); assert!(!job.is_running());
assert!(!job.is_running()); Poll::Ready(())
}));
} }
quickcheck(prop as fn(_)) quickcheck(prop as fn(_))
@ -392,23 +393,22 @@ mod tests {
r.provider = id.clone(); r.provider = id.clone();
let _ = store.add_provider(r); let _ = store.add_provider(r);
} }
// Polling with an instant beyond the deadline for the next run
// is guaranteed to run the job, without the job needing to poll the `Delay` block_on(poll_fn(|ctx| {
// and thus without needing to run `poll` in the context of a task let now = Instant::now() + job.inner.interval;
// for testing purposes. // All (non-expired) records in the store must be yielded by the job.
let now = Instant::now() + job.inner.interval; for r in store.provided().map(|r| r.into_owned()).collect::<Vec<_>>() {
// All (non-expired) records in the store must be yielded by the job. if !r.is_expired(now) {
for r in store.provided().map(|r| r.into_owned()).collect::<Vec<_>>() { assert_eq!(job.poll(ctx, &mut store, now), Poll::Ready(r));
if !r.is_expired(now) { assert!(job.is_running());
assert_eq!(job.poll(&mut store, now), Poll::Ready(r)); }
assert!(job.is_running());
} }
} assert_eq!(job.poll(ctx, &mut store, now), Poll::Pending);
assert_eq!(job.poll(&mut store, now), Poll::Pending); assert!(!job.is_running());
assert!(!job.is_running()); Poll::Ready(())
}));
} }
quickcheck(prop as fn(_)) quickcheck(prop as fn(_))
} }
} }

1
protocols/noise/Cargo.toml
View File

@ -18,7 +18,6 @@ protobuf = "2.8"
rand = "^0.7.2" rand = "^0.7.2"
ring = { version = "0.16.9", features = ["alloc"], default-features = false } ring = { version = "0.16.9", features = ["alloc"], default-features = false }
snow = { version = "0.6.1", features = ["ring-resolver"], default-features = false } snow = { version = "0.6.1", features = ["ring-resolver"], default-features = false }
tokio-io = "0.1"
x25519-dalek = "0.5" x25519-dalek = "0.5"
zeroize = "1" zeroize = "1"