Add tests for blob dissemination (#416)

* Make nomos-cli config public

Split nomos-cli into lib and bin and make config public
so that it can be reused in tests

* Add tests for nomos-cli disseminate

* fmt

* feature gate whole file instead of function

* Exit unsuccessfully when failing dissemination

nomos-cli/src/cmds/disseminate/mod.rs

@@ -22,22 +22,22 @@ use std::{path::PathBuf, time::Duration};
 pub struct Disseminate {
     // TODO: accept bytes
     #[clap(short, long)]
-    data: String,
+    pub data: String,
     /// Path to the network config file
     #[clap(short, long)]
-    network_config: PathBuf,
+    pub network_config: PathBuf,
     /// The data availability protocol to use. Defaults to full replication.
     #[clap(flatten)]
-    da_protocol: DaProtocolChoice,
+    pub da_protocol: DaProtocolChoice,
     /// Timeout in seconds. Defaults to 120 seconds.
     #[clap(short, long, default_value = "120")]
-    timeout: u64,
+    pub timeout: u64,
 }
 
 #[derive(Debug, Clone, Args)]
 pub struct FullReplicationSettings {
     #[clap(long, default_value = "1")]
-    num_attestations: usize,
+    pub num_attestations: usize,
 }
 
 impl Disseminate {
@@ -169,11 +169,23 @@ impl ServiceCore for DisseminateService {
                 let adapter = Libp2pAdapter::new(network_relay).await;
                 let da = FullReplication::new(AbsoluteNumber::new(da_settings.num_attestations));
 
-                if tokio::time::timeout(settings.timeout, disseminate(da, settings.bytes, adapter))
+                match tokio::time::timeout(
-                    .await
+                    settings.timeout,
-                    .is_err()
+                    disseminate(da, settings.bytes, adapter),
+                )
+                .await
                 {
-                    tracing::error!("Timeout reached, check the logs for additional details");
+                    Err(_) => {
+                        tracing::error!("Timeout reached, check the logs for additional details");
+                        std::process::exit(1);
+                    }
+                    Ok(Err(_)) => {
+                        tracing::error!(
+                            "Could not disseminate blob, check logs for additional details"
+                        );
+                        std::process::exit(1);
+                    }
+                    _ => {}
                 }
             }
         }
@@ -191,21 +203,21 @@ impl ServiceCore for DisseminateService {
 // We can enforce only sensible combinations of protocol/settings
 // are specified by using special clap directives
 #[derive(Clone, Debug, Args)]
-pub(super) struct DaProtocolChoice {
+pub struct DaProtocolChoice {
     #[clap(long, default_value = "full-replication")]
-    da_protocol: Protocol,
+    pub da_protocol: Protocol,
     #[clap(flatten)]
-    settings: ProtocolSettings,
+    pub settings: ProtocolSettings,
 }
 
 #[derive(Clone, Debug, Args)]
-struct ProtocolSettings {
+pub struct ProtocolSettings {
     #[clap(flatten)]
-    full_replication: FullReplicationSettings,
+    pub full_replication: FullReplicationSettings,
 }
 
 #[derive(Clone, Debug, ValueEnum)]
-pub(super) enum Protocol {
+pub enum Protocol {
     FullReplication,
 }
 

nomos-cli/src/cmds/mod.rs

@@ -1,6 +1,6 @@
 use clap::Subcommand;
 
-mod disseminate;
+pub mod disseminate;
 
 #[derive(Debug, Subcommand)]
 pub enum Command {

nomos-cli/src/lib.rs

@@ -0,0 +1,21 @@
+pub mod cmds;
+
+use clap::Parser;
+use cmds::Command;
+
+#[derive(Parser)]
+#[command(author, version, about, long_about = None)]
+pub struct Cli {
+    #[command(subcommand)]
+    command: Command,
+}
+
+impl Cli {
+    pub fn run(self) -> Result<(), Box<dyn std::error::Error>> {
+        self.command.run()
+    }
+
+    pub fn command(&self) -> &Command {
+        &self.command
+    }
+}

nomos-cli/src/main.rs

@@ -1,15 +1,7 @@
-mod cmds;
-
 use clap::Parser;
-use cmds::Command;
+use nomos_cli::Cli;
-#[derive(Parser)]
-#[command(author, version, about, long_about = None)]
-struct Cli {
-    #[command(subcommand)]
-    command: Command,
-}
 
 fn main() {
     let cli = Cli::parse();
-    cli.command.run().unwrap();
+    cli.run().unwrap();
 }

tests/Cargo.toml

@@ -35,6 +35,7 @@ async-trait = "0.1"
 fraction = "0.13"
 ntest = "0.9.0"
 criterion = { version = "0.5", features = ["async_tokio"] }
+nomos-cli = { path = "../nomos-cli" }
 
 [[test]]
 name = "test_consensus_happy_path"
@@ -48,6 +49,12 @@ path = "src/tests/unhappy.rs"
 name = "test_mixnet"
 path = "src/tests/mixnet.rs"
 
+[[test]]
+name = "test_cli"
+path = "src/tests/cli.rs"
+# disseminate is only implemented for libp2p
+required-features = ["libp2p"]
+
 [[bench]]
 name = "mixnet"
 path = "src/benches/mixnet.rs"

tests/src/nodes/nomos.rs

@@ -35,6 +35,7 @@ pub struct NomosNode {
     addr: SocketAddr,
     _tempdir: tempfile::TempDir,
     child: Child,
+    config: Config,
 }
 
 impl Drop for NomosNode {
@@ -77,6 +78,7 @@ impl NomosNode {
             addr: config.http.backend.address,
             child,
             _tempdir: dir,
+            config,
         };
         tokio::time::timeout(std::time::Duration::from_secs(10), async {
             node.wait_online().await
@@ -131,6 +133,10 @@ impl NomosNode {
             .map(|f| std::fs::read_to_string(f).unwrap())
             .collect::<String>()
     }
+
+    pub fn config(&self) -> &Config {
+        &self.config
+    }
 }
 
 #[async_trait::async_trait]

tests/src/tests/cli.rs

@@ -0,0 +1,48 @@
+use fraction::{Fraction, One};
+use nomos_cli::cmds::{
+    disseminate::{
+        DaProtocolChoice, Disseminate, FullReplicationSettings, Protocol, ProtocolSettings,
+    },
+    Command,
+};
+use std::time::Duration;
+use tempfile::NamedTempFile;
+use tests::{MixNode, Node, NomosNode, SpawnConfig};
+
+#[tokio::test]
+async fn disseminate_blob() {
+    let (_mixnodes, mixnet_node_configs, mixnet_topology) = MixNode::spawn_nodes(2).await;
+    let mut nodes = NomosNode::spawn_nodes(SpawnConfig::Star {
+        n_participants: 2,
+        threshold: Fraction::one(),
+        timeout: Duration::from_secs(10),
+        mixnet_node_configs,
+        mixnet_topology,
+    })
+    .await;
+
+    // kill the node so that we can reuse its network config
+    nodes[1].stop();
+
+    let network_config = nodes[1].config().network.clone();
+    let mut file = NamedTempFile::new().unwrap();
+    let config_path = file.path().to_owned();
+    serde_yaml::to_writer(&mut file, &network_config).unwrap();
+    let cmd = Command::Disseminate(Disseminate {
+        data: "Hello World".into(),
+        timeout: 20,
+        network_config: config_path,
+        da_protocol: DaProtocolChoice {
+            da_protocol: Protocol::FullReplication,
+            settings: ProtocolSettings {
+                full_replication: FullReplicationSettings {
+                    num_attestations: 1,
+                },
+            },
+        },
+    });
+
+    std::thread::spawn(move || cmd.run().unwrap())
+        .join()
+        .unwrap();
+}