Integrations tests (#221)

* make config struct fields public

* add basic integration tests

* Add libssl-dev dependency

* fix comments

---------

Co-authored-by: Gusto <bacvinka@gmail.com>

Cargo.toml

@@ -10,5 +10,6 @@ members = [
     "nomos-services/http",
     "nodes/nomos-node",
     "simulations",
-    "consensus-engine"
+    "consensus-engine",
+    "tests",
 ]

ci/Dockerfile

@@ -10,7 +10,7 @@ RUN echo 'deb http://deb.debian.org/debian bullseye-backports main' \
 
 # Dependecies for publishing documentation and building waku-bindings.
 RUN apt-get update && apt-get install -yq \
-    openssh-client git python3-pip clang \
+    libssl-dev openssh-client git python3-pip clang \
     golang-src/bullseye-backports \
     golang-doc/bullseye-backports \
     golang/bullseye-backports

nodes/nomos-node/src/lib.rs

@@ -19,11 +19,11 @@ use nomos_mempool::{
 use nomos_network::{backends::waku::Waku, NetworkService};
 use overwatch_derive::*;
 use overwatch_rs::services::{handle::ServiceHandle, ServiceData};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
 pub use tx::Tx;
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Debug, Clone, Serialize)]
 pub struct Config {
     pub log: <Logger as ServiceData>::Settings,
     pub network: <NetworkService<Waku> as ServiceData>::Settings,

nomos-services/consensus/src/lib.rs

@@ -54,9 +54,9 @@ pub type Seed = [u8; 32];
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct CarnotSettings<Fountain: FountainCode, O: Overlay> {
-    private_key: [u8; 32],
+    pub private_key: [u8; 32],
-    fountain_settings: Fountain::Settings,
+    pub fountain_settings: Fountain::Settings,
-    overlay_settings: O::Settings,
+    pub overlay_settings: O::Settings,
 }
 
 impl<Fountain: FountainCode, O: Overlay> Clone for CarnotSettings<Fountain, O> {
@@ -785,14 +785,14 @@ impl RelayMessage for ConsensusMsg {}
 #[serde_as]
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct CarnotInfo {
-    id: NodeId,
+    pub id: NodeId,
-    current_view: View,
+    pub current_view: View,
-    highest_voted_view: View,
+    pub highest_voted_view: View,
-    local_high_qc: StandardQc,
+    pub local_high_qc: StandardQc,
     #[serde_as(as = "Vec<(_, _)>")]
-    safe_blocks: HashMap<BlockId, consensus_engine::Block>,
+    pub safe_blocks: HashMap<BlockId, consensus_engine::Block>,
-    last_view_timeout_qc: Option<TimeoutQc>,
+    pub last_view_timeout_qc: Option<TimeoutQc>,
-    committed_blocks: Vec<BlockId>,
+    pub committed_blocks: Vec<BlockId>,
 }
 
 #[cfg(test)]

nomos-services/network/src/backends/waku.rs

@@ -31,8 +31,8 @@ pub struct WakuInfo {
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct WakuConfig {
     #[serde(flatten)]
-    inner: WakuNodeConfig,
+    pub inner: WakuNodeConfig,
-    initial_peers: Vec<Multiaddr>,
+    pub initial_peers: Vec<Multiaddr>,
 }
 
 /// Interaction with Waku node

tests/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "tests"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+nomos-node = { path = "../nodes/nomos-node" }
+nomos-consensus = { path = "../nomos-services/consensus" }
+nomos-network = { path = "../nomos-services/network", features = ["waku"] }
+nomos-log = { path = "../nomos-services/log" }
+nomos-http = { path = "../nomos-services/http", features = ["http"] }
+overwatch-rs = { git = "https://github.com/logos-co/Overwatch", branch = "main" }
+nomos-core = { path = "../nomos-core" }
+consensus-engine = { path = "../consensus-engine", features = ["serde"] }
+nomos-mempool = { path = "../nomos-services/mempool", features = ["waku", "mock"] }
+rand = "0.8"
+once_cell = "1"
+rand_xoshiro = "0.6"
+secp256k1 = { version = "0.26", features = ["rand"] }
+waku-bindings = "0.1.1"
+reqwest = { version = "0.11", features = ["json"] }
+tempfile = "3.6"
+serde_json = "1"
+tokio = "1"
+futures = "0.3"
+async-trait = "0.1"
+
+[[test]]
+name = "test_consensus_happy_path"
+path = "src/tests/happy.rs"
+
+
+[features]
+metrics = ["nomos-node/metrics"]

tests/src/lib.rs

@@ -0,0 +1,36 @@
+mod nodes;
+
+pub use nodes::NomosNode;
+use once_cell::sync::Lazy;
+use rand::SeedableRng;
+use rand_xoshiro::Xoshiro256PlusPlus;
+use std::fmt::Debug;
+use std::net::TcpListener;
+use std::sync::Mutex;
+
+static RNG: Lazy<Mutex<Xoshiro256PlusPlus>> =
+    Lazy::new(|| Mutex::new(Xoshiro256PlusPlus::seed_from_u64(42)));
+
+static NET_PORT: Mutex<u16> = Mutex::new(8000);
+
+pub fn get_available_port() -> u16 {
+    let mut port = NET_PORT.lock().unwrap();
+    *port += 1;
+    while TcpListener::bind(("127.0.0.1", *port)).is_err() {
+        *port += 1;
+    }
+    *port
+}
+
+#[async_trait::async_trait]
+pub trait Node: Sized {
+    type ConsensusInfo: Debug + Clone + PartialEq;
+    async fn spawn_nodes(config: SpawnConfig) -> Vec<Self>;
+    async fn consensus_info(&self) -> Self::ConsensusInfo;
+    fn stop(&mut self);
+}
+
+#[derive(Clone, Copy)]
+pub enum SpawnConfig {
+    Star { n_participants: usize },
+}

tests/src/nodes/mod.rs

@@ -0,0 +1,3 @@
+mod nomos;
+
+pub use nomos::NomosNode;

tests/src/nodes/nomos.rs

@@ -0,0 +1,180 @@
+// std
+use std::io::Read;
+use std::net::SocketAddr;
+use std::process::{Child, Command, Stdio};
+use std::time::Duration;
+// internal
+use crate::{get_available_port, Node, SpawnConfig, RNG};
+use consensus_engine::overlay::{RoundRobin, Settings};
+use nomos_consensus::{CarnotInfo, CarnotSettings};
+use nomos_http::backends::axum::AxumBackendSettings;
+use nomos_network::{
+    backends::waku::{WakuConfig, WakuInfo},
+    NetworkConfig,
+};
+use nomos_node::Config;
+use waku_bindings::{Multiaddr, PeerId};
+// crates
+use once_cell::sync::Lazy;
+use rand::Rng;
+use reqwest::Client;
+use tempfile::NamedTempFile;
+
+static CLIENT: Lazy<Client> = Lazy::new(Client::new);
+const NOMOS_BIN: &str = "../target/debug/nomos-node";
+const CARNOT_INFO_API: &str = "carnot/info";
+const NETWORK_INFO_API: &str = "network/info";
+
+pub struct NomosNode {
+    addr: SocketAddr,
+    _tempdir: tempfile::TempDir,
+    child: Child,
+}
+
+impl Drop for NomosNode {
+    fn drop(&mut self) {
+        let mut output = String::new();
+        if let Some(stdout) = &mut self.child.stdout {
+            stdout.read_to_string(&mut output).unwrap();
+        }
+        // self.child.stdout.as_mut().unwrap().read_to_string(&mut output).unwrap();
+        println!("{} stdout: {}", self.addr, output);
+        self.child.kill().unwrap();
+    }
+}
+
+impl NomosNode {
+    pub async fn spawn(config: &Config) -> Self {
+        // Waku stores the messages in a db file in the current dir, we need a different
+        // directory for each node to avoid conflicts
+        let dir = tempfile::tempdir().unwrap();
+        let mut file = NamedTempFile::new().unwrap();
+        let config_path = file.path().to_owned();
+        serde_json::to_writer(&mut file, config).unwrap();
+        let child = Command::new(std::env::current_dir().unwrap().join(NOMOS_BIN))
+            .arg(&config_path)
+            .current_dir(dir.path())
+            .stdout(Stdio::null())
+            .spawn()
+            .unwrap();
+        let node = Self {
+            addr: config.http.backend.address,
+            child,
+            _tempdir: dir,
+        };
+        node.wait_online().await;
+        node
+    }
+
+    async fn get(&self, path: &str) -> reqwest::Result<reqwest::Response> {
+        CLIENT
+            .get(format!("http://{}/{}", self.addr, path))
+            .send()
+            .await
+    }
+
+    async fn wait_online(&self) {
+        while self.get(CARNOT_INFO_API).await.is_err() {
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+    }
+
+    pub async fn peer_id(&self) -> PeerId {
+        self.get(NETWORK_INFO_API)
+            .await
+            .unwrap()
+            .json::<WakuInfo>()
+            .await
+            .unwrap()
+            .peer_id
+            .unwrap()
+    }
+
+    pub async fn get_listening_address(&self) -> Multiaddr {
+        self.get(NETWORK_INFO_API)
+            .await
+            .unwrap()
+            .json::<WakuInfo>()
+            .await
+            .unwrap()
+            .listen_addresses
+            .unwrap()
+            .swap_remove(0)
+    }
+}
+
+#[async_trait::async_trait]
+impl Node for NomosNode {
+    type ConsensusInfo = CarnotInfo;
+
+    async fn spawn_nodes(config: SpawnConfig) -> Vec<Self> {
+        match config {
+            SpawnConfig::Star { n_participants } => {
+                let mut ids = vec![[0; 32]; n_participants];
+                for id in &mut ids {
+                    RNG.lock().unwrap().fill(id);
+                }
+                let mut configs = ids
+                    .iter()
+                    .map(|id| create_node_config(ids.clone(), *id))
+                    .collect::<Vec<_>>();
+                let mut nodes = vec![Self::spawn(&configs[0]).await];
+                let listening_addr = nodes[0].get_listening_address().await;
+                configs.drain(0..1);
+                for conf in &mut configs {
+                    conf.network
+                        .backend
+                        .initial_peers
+                        .push(listening_addr.clone());
+                    nodes.push(Self::spawn(conf).await);
+                }
+                nodes
+            }
+        }
+    }
+
+    async fn consensus_info(&self) -> Self::ConsensusInfo {
+        self.get(CARNOT_INFO_API)
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap()
+    }
+
+    fn stop(&mut self) {
+        self.child.kill().unwrap();
+    }
+}
+
+fn create_node_config(nodes: Vec<[u8; 32]>, private_key: [u8; 32]) -> Config {
+    let mut config = Config {
+        network: NetworkConfig {
+            backend: WakuConfig {
+                initial_peers: vec![],
+                inner: Default::default(),
+            },
+        },
+        consensus: CarnotSettings {
+            private_key,
+            fountain_settings: (),
+            overlay_settings: Settings {
+                nodes,
+                leader: RoundRobin::new(),
+            },
+        },
+        log: Default::default(),
+        http: nomos_http::http::HttpServiceSettings {
+            backend: AxumBackendSettings {
+                address: format!("127.0.0.1:{}", get_available_port())
+                    .parse()
+                    .unwrap(),
+                cors_origins: vec![],
+            },
+        },
+        #[cfg(feature = "metrics")]
+        metrics: Default::default(),
+    };
+    config.network.backend.inner.port = Some(get_available_port() as usize);
+    config
+}

tests/src/tests/happy.rs

@@ -0,0 +1,49 @@
+use futures::stream::{self, StreamExt};
+use std::collections::HashSet;
+use tests::{Node, NomosNode, SpawnConfig};
+
+const TARGET_VIEW: i64 = 20;
+
+async fn happy_test(nodes: Vec<NomosNode>) {
+    while stream::iter(&nodes)
+        .any(|n| async move { n.consensus_info().await.current_view < TARGET_VIEW })
+        .await
+    {
+        println!(
+            "waiting... {}",
+            stream::iter(&nodes)
+                .then(|n| async move { format!("{}", n.consensus_info().await.current_view) })
+                .collect::<Vec<_>>()
+                .await
+                .join(" | ")
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    }
+    let infos = stream::iter(nodes)
+        .then(|n| async move { n.consensus_info().await })
+        .collect::<Vec<_>>()
+        .await;
+    // check that they have the same block
+    let blocks = infos
+        .iter()
+        .map(|i| {
+            i.safe_blocks
+                .values()
+                .find(|b| b.view == TARGET_VIEW)
+                .unwrap()
+        })
+        .collect::<HashSet<_>>();
+    assert_eq!(blocks.len(), 1);
+}
+
+#[tokio::test]
+async fn two_nodes_happy() {
+    let nodes = NomosNode::spawn_nodes(SpawnConfig::Star { n_participants: 2 }).await;
+    happy_test(nodes).await;
+}
+
+#[tokio::test]
+async fn three_nodes_happy() {
+    let nodes = NomosNode::spawn_nodes(SpawnConfig::Star { n_participants: 3 }).await;
+    happy_test(nodes).await;
+}