Add a modest integration test

4 files changed, 115 insertions, 2 deletions

diff --git a/.drone.yml b/.drone.yml
index 4df6631..f827928 100644
--- a/.drone.yml
+++ b/.drone.yml
@@ -39,3 +39,4 @@ steps:
       - cargo build-all-features
       - cargo build --example fullmesh
       - cargo build --example basalt --features "basalt"
+      - cargo test --all-features -- --test-threads 1
diff --git a/src/lib.rs b/src/lib.rs
index 2ce5a51..bed8821 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -25,5 +25,8 @@ mod server;
 pub mod netapp;
 pub mod peering;
 
-pub use netapp::*;
+pub use crate::netapp::*;
 pub use util::NodeID;
+
+#[cfg(test)]
+mod test;
diff --git a/src/test.rs b/src/test.rs
new file mode 100644
index 0000000..e4ec495
--- /dev/null
+++ b/src/test.rs
@@ -0,0 +1,109 @@
+use std::net::SocketAddr;
+use std::sync::{Arc};
+use std::time::{Duration};
+
+use tokio::select;
+use tokio::sync::watch;
+
+use sodiumoxide::crypto::auth;
+use sodiumoxide::crypto::sign::ed25519;
+
+use crate::peering::fullmesh::*;
+use crate::netapp::*;
+use crate::NodeID;
+
+
+#[tokio::test(flavor="current_thread")]
+async fn test_with_basic_scheduler() {
+	run_test().await
+}
+
+#[tokio::test(flavor="multi_thread", worker_threads=4)]
+async fn test_with_threaded_scheduler() {
+	run_test().await
+}
+
+async fn run_test() {
+	select! {
+		_ = run_test_inner() => (),
+		_ = tokio::time::sleep(Duration::from_secs(20)) => panic!("timeout"),
+	}
+}
+
+async fn run_test_inner() {
+	let netid = auth::gen_key();
+
+	let (pk1, sk1) = ed25519::gen_keypair();
+	let (pk2, sk2) = ed25519::gen_keypair();
+	let (pk3, sk3) = ed25519::gen_keypair();
+
+	let addr1: SocketAddr = "127.0.0.1:19991".parse().unwrap();
+	let addr2: SocketAddr = "127.0.0.1:19992".parse().unwrap();
+	let addr3: SocketAddr = "127.0.0.1:19993".parse().unwrap();
+
+	let (stop_tx, stop_rx) = watch::channel(false);
+
+	let (thread1, _netapp1, peering1) = run_netapp(netid.clone(), pk1, sk1, addr1, vec![], stop_rx.clone());
+	tokio::time::sleep(Duration::from_secs(2)).await;
+
+	// Connect second node and check it peers with everyone
+	let (thread2, _netapp2, peering2) = run_netapp(netid.clone(), pk2, sk2, addr2, vec![(pk1, addr1)], stop_rx.clone());
+	tokio::time::sleep(Duration::from_secs(5)).await;
+
+	let pl1 = peering1.get_peer_list();
+	println!("A pl1: {:?}", pl1);
+	assert_eq!(pl1.len(), 2);
+
+	let pl2 = peering2.get_peer_list();
+	println!("A pl2: {:?}", pl2);
+	assert_eq!(pl2.len(), 2);
+
+	// Connect third ndoe and check it peers with everyone
+	let (thread3, _netapp3, peering3) = run_netapp(netid, pk3, sk3, addr3, vec![(pk2, addr2)], stop_rx.clone());
+	tokio::time::sleep(Duration::from_secs(5)).await;
+
+	let pl1 = peering1.get_peer_list();
+	println!("B pl1: {:?}", pl1);
+	assert_eq!(pl1.len(), 3);
+
+	let pl2 = peering2.get_peer_list();
+	println!("B pl2: {:?}", pl2);
+	assert_eq!(pl2.len(), 3);
+
+	let pl3 = peering3.get_peer_list();
+	println!("B pl3: {:?}", pl3);
+	assert_eq!(pl3.len(), 3);
+
+	// Send stop signal and wait for everyone to finish
+	stop_tx.send(true).unwrap();
+	thread1.await.unwrap();
+	thread2.await.unwrap();
+	thread3.await.unwrap();
+}
+
+fn run_netapp(
+	netid: auth::Key,
+	_pk: NodeID,
+	sk: ed25519::SecretKey,
+	listen_addr: SocketAddr,
+	bootstrap_peers: Vec<(NodeID, SocketAddr)>,
+	must_exit: watch::Receiver<bool>)
+	->
+	(tokio::task::JoinHandle<()>, Arc<NetApp>, Arc<FullMeshPeeringStrategy>)
+{
+
+	let netapp = NetApp::new(netid, sk);
+	let peering = FullMeshPeeringStrategy::new(netapp.clone(), bootstrap_peers);
+
+	let peering2 = peering.clone();
+	let netapp2 = netapp.clone();
+	let fut = tokio::spawn(async move {
+		tokio::join!(
+			netapp2.listen(listen_addr, None, must_exit.clone()),
+			peering2.run(must_exit.clone()),
+		);
+	});
+
+	(fut, netapp, peering)
+}
+
diff --git a/src/util.rs b/src/util.rs
index e5b57ec..a87f779 100644
--- a/src/util.rs
+++ b/src/util.rs
@@ -27,7 +27,7 @@ where
 /// from a watcher that tells us when to exit.
 /// Usefull in a select statement to interrupt another
 /// future:
-/// ```
+/// ```ignore
 /// select!(
 ///     _ = a_long_task() => Success,
 ///     _ = await_exit(must_exit) => Interrupted,