node: Test push/pull flow

Signed-off-by: Slack Coder <slackcoder@server.ky>

node/src/protocol.rs

@@ -51,7 +51,13 @@ pub type Timestamp = u64;
 
 /// A protocol event.
 #[derive(Debug, Clone)]
-pub enum Event {}
+pub enum Event {
+    RefsFetched {
+        from: Url,
+        project: Id,
+        updated: Vec<RefUpdate>,
+    },
+}
 
 /// Error returned by [`Command::Fetch`].
 #[derive(thiserror::Error, Debug)]
@@ -221,6 +227,11 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Protoco
         &self.context.storage
     }
 
+    /// Get the mutable storage instance.
+    pub fn storage_mut(&mut self) -> &mut T {
+        &mut self.context.storage
+    }
+
     /// Get the local protocol time.
     pub fn local_time(&self) -> LocalTime {
         self.context.clock.local_time()

node/src/protocol/peer.rs

@@ -201,8 +201,16 @@ impl Peer {
                     // TODO: Check that we're tracking this user as well.
                     if ctx.config.is_tracking(&id) {
                         // TODO: Check refs to see if we should try to fetch or not.
-                        let updated = ctx.fetch(&id, git);
-                        if !updated.is_empty() {
+                        let updated_refs = ctx.fetch(&id, git);
+                        let is_updated = !updated_refs.is_empty();
+
+                        ctx.io.push_back(Io::Event(Event::RefsFetched {
+                            from: git.clone(),
+                            project: id.clone(),
+                            updated: updated_refs,
+                        }));
+
+                        if is_updated {
                             return Ok(Some(Message::RefsUpdate {
                                 id,
                                 signer,

node/src/test/peer.rs

@@ -204,7 +204,13 @@ where
         msgs.into_iter()
     }
 
-    /// Get a draining iterator over the peers's I/O outbox.
+    /// Get a draining iterator over the peer's emitted events.
+    pub fn events(&mut self) -> impl Iterator<Item = Event> + '_ {
+        self.outbox()
+            .filter_map(|io| if let Io::Event(e) = io { Some(e) } else { None })
+    }
+
+    /// Get a draining iterator over the peer's I/O outbox.
     pub fn outbox(&mut self) -> impl Iterator<Item = Io<Event, DisconnectReason>> + '_ {
         self.protocol.outbox().drain(..)
     }

node/src/test/tests.rs

@@ -368,6 +368,80 @@ fn test_persistent_peer_reconnect() {
     assert_matches!(alice.outbox().next(), None);
 }
 
+#[test]
+fn test_push_and_pull() {
+    logger::init(log::Level::Debug);
+
+    let tempdir = tempfile::tempdir().unwrap();
+
+    let storage_alice = Storage::open(
+        tempdir.path().join("alice").join("storage"),
+        MockSigner::default(),
+    )
+    .unwrap();
+    let repo = fixtures::repository(tempdir.path().join("working"));
+    let mut alice = Peer::new("alice", [7, 7, 7, 7], storage_alice);
+
+    let storage_bob = Storage::open(
+        tempdir.path().join("bob").join("storage"),
+        MockSigner::default(),
+    )
+    .unwrap();
+    let mut bob = Peer::new("bob", [8, 8, 8, 8], storage_bob);
+
+    let storage_eve = Storage::open(
+        tempdir.path().join("eve").join("storage"),
+        MockSigner::default(),
+    )
+    .unwrap();
+    let mut eve = Peer::new("eve", [9, 9, 9, 9], storage_eve);
+
+    // Alice and Bob connect to Eve.
+    alice.command(protocol::Command::Connect(eve.addr()));
+    bob.command(protocol::Command::Connect(eve.addr()));
+
+    let mut sim = Simulation::new(
+        LocalTime::now(),
+        alice.rng.clone(),
+        simulator::Options::default(),
+    )
+    .initialize([&mut alice, &mut bob, &mut eve]);
+
+    // Here we expect Alice to connect to Eve.
+    sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled());
+
+    // Alice creates a new project.
+    let (proj_id, _) = rad::init(
+        &repo,
+        "alice",
+        "alice's repo",
+        storage::BranchName::from("master"),
+        alice.storage_mut(),
+    )
+    .unwrap();
+
+    // Bob tracks Alice's project.
+    let (sender, _) = chan::bounded(1);
+    bob.command(protocol::Command::Track(proj_id.clone(), sender));
+
+    // Eve tracks Alice's project.
+    let (sender, _) = chan::bounded(1);
+    eve.command(protocol::Command::Track(proj_id.clone(), sender));
+
+    // Neither of them have it in the beginning.
+    assert!(eve.storage().get(&proj_id).unwrap().is_none());
+    assert!(bob.storage().get(&proj_id).unwrap().is_none());
+
+    // Alice announces her refs.
+    // We now expect Eve to fetch Alice's project from Alice.
+    // Then we expect Bob to fetch Alice's project from Eve.
+    // TODO: Check that Bob is fetching from Eve and not Alice, via an event.
+    alice.command(protocol::Command::AnnounceRefsUpdate(proj_id.clone()));
+    sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled());
+    assert!(eve.storage().get(&proj_id).unwrap().is_some());
+    assert!(bob.storage().get(&proj_id).unwrap().is_some());
+}
+
 #[test]
 fn prop_inventory_exchange_dense() {
     fn property(alice_inv: MockStorage, bob_inv: MockStorage, eve_inv: MockStorage) {