chore: Da dispersal unit tests update

nomos-cli/Cargo.toml

@@ -9,7 +9,7 @@ description = "Cli app to interact with Nomos nodes and perform various tasks"
 [dependencies]
 fraction = "0.13"
 tracing = "0.1"
-tracing-subscriber = "0.3"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 async-trait = "0.1"
 clap = { version = "4", features = ["derive"] }
 serde_yaml = "0.9"

nomos-cli/src/da/network/swarm.rs

@@ -129,3 +129,78 @@ where
         }
     }
 }
+
+#[cfg(test)]
+pub mod test {
+    use crate::da::network::swarm::ExecutorSwarm;
+    use crate::test_utils::AllNeighbours;
+    use kzgrs_backend::common::blob::DaBlob;
+    use kzgrs_backend::common::Column;
+    use libp2p::identity::Keypair;
+    use libp2p::PeerId;
+    use nomos_da_network_core::address_book::AddressBook;
+    use nomos_da_network_core::swarm::validator::ValidatorSwarm;
+    use nomos_libp2p::Multiaddr;
+    use std::time::Duration;
+    use tokio::sync::mpsc::unbounded_channel;
+
+    #[tokio::test]
+    async fn test_dispersal_with_swarms() {
+        let k1 = Keypair::generate_ed25519();
+        let k2 = Keypair::generate_ed25519();
+        let executor_peer = PeerId::from_public_key(&k1.public());
+        let validator_peer = PeerId::from_public_key(&k2.public());
+        let neighbours = AllNeighbours {
+            neighbours: [
+                PeerId::from_public_key(&k1.public()),
+                PeerId::from_public_key(&k2.public()),
+            ]
+            .into_iter()
+            .collect(),
+        };
+
+        let addr: Multiaddr = "/ip4/127.0.0.1/udp/5063/quic-v1".parse().unwrap();
+        let addr2 = addr.clone().with_p2p(validator_peer).unwrap();
+        let addr2_book = AddressBook::from_iter(vec![(executor_peer, addr2.clone())]);
+
+        let (dispersal_broadcast_sender, _) = unbounded_channel();
+
+        let mut executor =
+            ExecutorSwarm::new(k1, neighbours.clone(), dispersal_broadcast_sender.clone());
+        let (mut validator, _) = ValidatorSwarm::new(k2, neighbours.clone(), addr2_book);
+
+        tokio::spawn(async move {
+            let validator_swarm = validator.protocol_swarm_mut();
+            validator_swarm.listen_on(addr).unwrap();
+
+            validator.run().await;
+        });
+
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        executor.dial(addr2).unwrap();
+        tokio::time::sleep(Duration::from_secs(1)).await;
+
+        let executor_disperse_blob_sender = executor.swarm.behaviour().blobs_sender();
+
+        println!("Sending blob...");
+        executor_disperse_blob_sender
+            .send((
+                0,
+                DaBlob {
+                    column_idx: 0,
+                    column: Column(vec![]),
+                    column_commitment: Default::default(),
+                    aggregated_column_commitment: Default::default(),
+                    aggregated_column_proof: Default::default(),
+                    rows_commitments: vec![],
+                    rows_proofs: vec![],
+                },
+            ))
+            .unwrap();
+
+        let executor_task = tokio::spawn(async move {
+            executor.run().await;
+        });
+        assert!(executor_task.await.is_ok());
+    }
+}

nomos-cli/src/lib.rs

@@ -2,6 +2,9 @@ pub mod api;
 pub mod cmds;
 pub mod da;
 
+#[cfg(test)]
+pub mod test_utils;
+
 use clap::Parser;
 use cmds::Command;
 

nomos-cli/src/test_utils.rs

@@ -0,0 +1,25 @@
+use libp2p::PeerId;
+use std::collections::HashSet;
+use subnetworks_assignations::MembershipHandler;
+
+#[derive(Clone)]
+pub struct AllNeighbours {
+    pub neighbours: HashSet<PeerId>,
+}
+
+impl MembershipHandler for AllNeighbours {
+    type NetworkId = u32;
+    type Id = PeerId;
+
+    fn membership(&self, _self_id: &Self::Id) -> HashSet<Self::NetworkId> {
+        [0].into_iter().collect()
+    }
+
+    fn is_allowed(&self, _id: &Self::Id) -> bool {
+        true
+    }
+
+    fn members_of(&self, _network_id: &Self::NetworkId) -> HashSet<Self::Id> {
+        self.neighbours.clone()
+    }
+}