node: Improvements to test reliability

radicle-cli/tests/commands.rs

@@ -12,7 +12,7 @@ use radicle::node::{Address, Alias, DEFAULT_TIMEOUT};
 use radicle::prelude::RepoId;
 use radicle::profile;
 use radicle::profile::Home;
-use radicle::storage::{ReadStorage, RemoteRepository};
+use radicle::storage::{ReadStorage, RefUpdate, RemoteRepository};
 use radicle::test::fixtures;
 
 use radicle_cli_test::TestFormula;
@@ -333,7 +333,7 @@ fn rad_id() {
     let events = alice.handle.events();
     bob.fork(acme, bob.home.path()).unwrap();
     bob.announce(acme, 2, bob.home.path()).unwrap();
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
 
     // Alice must have Bob to try add them as a delegate
     events
@@ -383,12 +383,12 @@ fn rad_id_multi_delegate() {
     eve.connect(&alice).converge([&alice]);
 
     bob.fork(acme, working.join("bob")).unwrap();
-    bob.has_inventory_of(&acme, &alice.id);
+    bob.has_remote_of(&acme, &alice.id);
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
 
     eve.fork(acme, working.join("eve")).unwrap();
-    eve.has_inventory_of(&acme, &bob.id);
+    eve.has_remote_of(&acme, &bob.id);
-    alice.has_inventory_of(&acme, &eve.id);
+    alice.has_remote_of(&acme, &eve.id);
 
     // TODO: Have formula with two connected nodes and a tracked project.
 
@@ -541,7 +541,7 @@ fn rad_id_conflict() {
 
     bob.fork(acme, working.join("bob")).unwrap();
     bob.announce(acme, 2, bob.home.path()).unwrap();
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
 
     formula(&environment.tmp(), "examples/rad-id-conflict.md")
         .unwrap()
@@ -983,9 +983,9 @@ fn rad_clean() {
 
     bob.fork(acme, bob.home.path()).unwrap();
     bob.announce(acme, 1, bob.home.path()).unwrap();
-    bob.has_inventory_of(&acme, &alice.id);
+    bob.has_remote_of(&acme, &alice.id);
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
-    eve.has_inventory_of(&acme, &alice.id);
+    eve.has_remote_of(&acme, &alice.id);
 
     formula(&environment.tmp(), "examples/rad-clean.md")
         .unwrap()
@@ -1117,8 +1117,8 @@ fn rad_clone_all() {
     // Fork and sync repo.
     bob.fork(acme, bob.home.path()).unwrap();
     bob.announce(acme, 2, bob.home.path()).unwrap();
-    bob.has_inventory_of(&acme, &alice.id);
+    bob.has_remote_of(&acme, &alice.id);
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
 
     test(
         "examples/rad-clone-all.md",
@@ -1127,7 +1127,7 @@ fn rad_clone_all() {
         [],
     )
     .unwrap();
-    eve.has_inventory_of(&acme, &bob.id);
+    eve.has_remote_of(&acme, &bob.id);
 }
 
 #[test]
@@ -1509,7 +1509,14 @@ fn test_cob_replication() {
     // Wait for Alice to fetch the clone refs.
     events
         .wait(
-            |e| matches!(e, Event::RefsFetched { .. }).then_some(()),
+            |e| {
+                matches!(
+                    e,
+                    Event::RefsFetched { updated, .. }
+                    if updated.iter().any(|u| matches!(u, RefUpdate::Created { .. }))
+                )
+                .then_some(())
+            },
             time::Duration::from_secs(6),
         )
         .unwrap();
@@ -1634,6 +1641,8 @@ fn rad_sync() {
     bob.routes_to(&[(acme, alice.id)]);
     eve.routes_to(&[(acme, alice.id)]);
     alice.routes_to(&[(acme, alice.id), (acme, eve.id), (acme, bob.id)]);
+    alice.is_synced_with(&acme, &eve.id);
+    alice.is_synced_with(&acme, &bob.id);
 
     test(
         "examples/rad-sync.md",
@@ -1709,12 +1718,18 @@ fn test_replication_via_seed() {
     seed.routes_to(&[(rid, alice.id), (rid, seed.id), (rid, bob.id)]);
     bob.routes_to(&[(rid, alice.id), (rid, seed.id), (rid, bob.id)]);
 
-    seed_events
+    seed_events.iter().any(|e| {
-        .iter()
+        matches!(
-        .any(|e| matches!(e, Event::RefsFetched { remote, .. } if remote == bob.id));
+            e, Event::RefsFetched { updated, remote, .. }
-    alice_events
+            if remote == bob.id && updated.iter().any(|u| u.is_created())
-        .iter()
+        )
-        .any(|e| matches!(e, Event::RefsFetched { remote, .. } if remote == seed.id));
+    });
+    alice_events.iter().any(|e| {
+        matches!(
+            e, Event::RefsFetched { updated, remote, .. }
+            if remote == seed.id && updated.iter().any(|u| u.is_created())
+        )
+    });
 
     seed.storage
         .repository(rid)
@@ -1763,13 +1778,13 @@ fn rad_remote() {
     bob.routes_to(&[(rid, alice.id)]);
     bob.fork(rid, bob.home.path()).unwrap();
     bob.announce(rid, 2, bob.home.path()).unwrap();
-    alice.has_inventory_of(&rid, &bob.id);
+    alice.has_remote_of(&rid, &bob.id);
 
     eve.connect(&bob);
     eve.routes_to(&[(rid, alice.id)]);
     eve.fork(rid, eve.home.path()).unwrap();
     eve.announce(rid, 2, eve.home.path()).unwrap();
-    alice.has_inventory_of(&rid, &eve.id);
+    alice.has_remote_of(&rid, &eve.id);
 
     test(
         "examples/rad-remote.md",
@@ -2090,7 +2105,7 @@ fn git_push_diverge() {
 
     bob.connect(&alice).converge([&alice]);
     bob.fork(acme, working.join("bob")).unwrap();
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
 
     formula(&environment.tmp(), "examples/git/git-push-diverge.md")
         .unwrap()
@@ -2131,7 +2146,7 @@ fn rad_push_and_pull_patches() {
 
     bob.connect(&alice).converge([&alice]);
     bob.fork(acme, working.join("bob")).unwrap();
-    alice.has_inventory_of(&acme, &bob.id);
+    alice.has_remote_of(&acme, &bob.id);
 
     formula(&environment.tmp(), "examples/rad-push-and-pull-patches.md")
         .unwrap()

radicle-node/src/test/environment.rs

@@ -19,6 +19,7 @@ use radicle::identity::{RepoId, Visibility};
 use radicle::node::config::ConnectAddress;
 use radicle::node::policy::store as policy;
 use radicle::node::routing::Store;
+use radicle::node::seed::Store as _;
 use radicle::node::Database;
 use radicle::node::{Alias, POLICIES_DB_FILE};
 use radicle::node::{ConnectOptions, Handle as _};
@@ -244,6 +245,14 @@ impl<G: Signer + cyphernet::Ecdh> NodeHandle<G> {
             .unwrap()
     }
 
+    /// Get sync status of a repo.
+    pub fn synced_seeds(&self, rid: &RepoId) -> Vec<node::seed::SyncedSeed> {
+        let db = Database::reader(self.home.node().join(node::NODE_DB_FILE)).unwrap();
+        let seeds = db.seeds_for(rid).unwrap();
+
+        seeds.into_iter().collect::<Result<Vec<_>, _>>().unwrap()
+    }
+
     /// Wait until this node's routing table matches the remotes.
     pub fn converge<'a>(
         &'a self,
@@ -278,9 +287,23 @@ impl<G: Signer + cyphernet::Ecdh> NodeHandle<G> {
         }
     }
 
+    /// Wait until this node is synced with another node, for the given repository.
+    #[track_caller]
+    pub fn is_synced_with(&mut self, rid: &RepoId, nid: &NodeId) {
+        log::debug!(target: "test", "Waiting for {} to be in sync with {nid} for {rid}", self.id);
+
+        loop {
+            let seeds = self.handle.seeds(*rid).unwrap();
+            if seeds.iter().any(|s| s.nid == *nid && s.is_synced()) {
+                break;
+            }
+            thread::sleep(Duration::from_millis(100));
+        }
+    }
+
     /// Wait until this node has the inventory of another node.
     #[track_caller]
-    pub fn has_inventory_of(&self, rid: &RepoId, nid: &NodeId) {
+    pub fn has_remote_of(&self, rid: &RepoId, nid: &NodeId) {
         log::debug!(target: "test", "Waiting for {} to have {rid}/{nid}", self.id);
         let events = self.handle.events();
 

radicle-node/src/tests.rs

@@ -1559,7 +1559,7 @@ fn test_refs_synced_event() {
 }
 
 #[test]
-fn test_push_and_pull() {
+fn test_init_and_seed() {
     let tempdir = tempfile::tempdir().unwrap();
 
     let storage_alice = Storage::open(
@@ -1626,19 +1626,22 @@ fn test_push_and_pull() {
     assert!(bob.get(proj_id).unwrap().is_none());
 
     // Bob seeds Alice's project.
-    let (sender, _) = chan::bounded(1);
+    let (sender, receiver) = chan::bounded(1);
     bob.command(service::Command::Seed(
         proj_id,
         policy::Scope::default(),
         sender,
     ));
+    assert!(receiver.recv().unwrap());
+
     // Eve seeds Alice's project.
-    let (sender, _) = chan::bounded(1);
+    let (sender, receiver) = chan::bounded(1);
     eve.command(service::Command::Seed(
         proj_id,
         policy::Scope::default(),
         sender,
     ));
+    assert!(receiver.recv().unwrap());
 
     let (send, _) = chan::bounded(1);
     // Alice announces her inventory.

radicle/src/storage.rs

@@ -229,6 +229,16 @@ impl RefUpdate {
             RefUpdate::Skipped { name, .. } => name.as_refstr(),
         }
     }
+
+    /// Is it an update.
+    pub fn is_updated(&self) -> bool {
+        matches!(self, RefUpdate::Updated { .. })
+    }
+
+    /// Is it a create.
+    pub fn is_created(&self) -> bool {
+        matches!(self, RefUpdate::Created { .. })
+    }
 }
 
 impl fmt::Display for RefUpdate {

radicle/src/test.rs

@@ -59,11 +59,11 @@ pub fn fetch<W: WriteRepository>(
 
     repo.set_identity_head()?;
     repo.set_head()?;
+
     let validations = repo.validate()?;
     if !validations.is_empty() {
         return Err(crate::storage::FetchError::Validation { validations });
     }
-
     Ok(updates)
 }