node, cli: Refactor test environment

crates/radicle-cli/tests/util/environment.rs

@@ -0,0 +1,316 @@
+use std::path::PathBuf;
+use std::str::FromStr;
+
+use localtime::LocalTime;
+use radicle::cob::cache::COBS_DB_FILE;
+use radicle::crypto::ssh::{keystore::MemorySigner, Keystore};
+use radicle::crypto::{KeyPair, Seed};
+use radicle::node::policy::store as policy;
+use radicle::node::{self, UserAgent};
+use radicle::node::{Alias, Config, POLICIES_DB_FILE};
+use radicle::profile::Home;
+use radicle::profile::{self};
+use radicle::storage::git::transport;
+use radicle::{Profile, Storage};
+
+use radicle_node::test::node::{Node, NodeHandle};
+
+use crate::util::formula::formula;
+
+pub(crate) mod config {
+    use super::*;
+    use radicle::node::config::{Config, Limits, Network, RateLimit, RateLimits};
+
+    /// Configuration for a test seed node.
+    ///
+    /// It sets the `RateLimit::capacity` to `usize::MAX` ensuring
+    /// that there are no rate limits for test nodes, since they all
+    /// operate on the same IP address. This prevents any announcement
+    /// messages from being dropped.
+    pub fn seed(alias: &'static str) -> Config {
+        Config {
+            network: Network::Test,
+            relay: node::config::Relay::Always,
+            limits: Limits {
+                rate: RateLimits {
+                    inbound: RateLimit {
+                        fill_rate: 1.0,
+                        capacity: usize::MAX,
+                    },
+                    outbound: RateLimit {
+                        fill_rate: 1.0,
+                        capacity: usize::MAX,
+                    },
+                },
+                ..Limits::default()
+            },
+            external_addresses: vec![node::Address::from_str(&format!(
+                "{alias}.radicle.example:8776"
+            ))
+            .unwrap()],
+            ..node(alias)
+        }
+    }
+
+    /// Relay node config.
+    pub fn relay(alias: &'static str) -> Config {
+        Config {
+            relay: node::config::Relay::Always,
+            ..node(alias)
+        }
+    }
+
+    /// Test node config.
+    pub fn node(alias: &'static str) -> Config {
+        Config::test(Alias::new(alias))
+    }
+}
+
+/// Test environment.
+pub struct Environment {
+    tempdir: tempfile::TempDir,
+    users: usize,
+}
+
+impl Default for Environment {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Environment {
+    /// Create a new test environment.
+    fn named(name: &'static str) -> Self {
+        Self {
+            tempdir: tempfile::TempDir::with_prefix("radicle-".to_owned() + name).unwrap(),
+            users: 0,
+        }
+    }
+
+    /// Create a new test environment.
+    pub fn new() -> Self {
+        Self::named("")
+    }
+
+    /// Return the temp directory path.
+    pub fn tempdir(&self) -> PathBuf {
+        self.tempdir.path().into()
+    }
+
+    /// Path to the working directory designated for given alias.
+    pub fn work(&self, has_alias: &impl HasAlias) -> PathBuf {
+        self.tempdir().join("work").join(has_alias.alias().as_ref())
+    }
+
+    /// We don't have `RAD_HOME` or `HOME` to rely on to compute a home as usual.
+    pub fn home(&self, alias: &Alias) -> Home {
+        Home::new(
+            self.tempdir()
+                .join("home")
+                .join(alias.to_string())
+                .join(".radicle"),
+        )
+        .unwrap()
+    }
+
+    /// Create a new default configuration.
+    pub fn config(&self, alias: &str) -> profile::Config {
+        let alias = Alias::new(alias);
+        profile::Config {
+            node: node::Config::test(alias),
+            cli: radicle::cli::Config { hints: false },
+            public_explorer: radicle::explorer::Explorer::default(),
+            preferred_seeds: vec![],
+            web: radicle::web::Config::default(),
+        }
+    }
+
+    /// Create a new profile in this environment.
+    /// This should be used when a running node is not required.
+    /// Using this function is only necessary if the desired configuration
+    /// differs from the default provided by [`Environment::config`] as
+    /// for this default the convenience function [`Environment::profile`]
+    /// is provided.
+    pub fn profile_with(&mut self, config: profile::Config) -> Profile {
+        let alias = config.alias().clone();
+        let home = self.home(&alias);
+        let keypair = KeyPair::from_seed(Seed::from([!(self.users as u8); 32]));
+        let policies_db = home.node().join(POLICIES_DB_FILE);
+        let cobs_db = home.cobs().join(COBS_DB_FILE);
+
+        config.write(&home.config()).unwrap();
+
+        let storage = Storage::open(
+            home.storage(),
+            radicle::git::UserInfo {
+                alias: alias.clone(),
+                key: keypair.pk.into(),
+            },
+        )
+        .unwrap();
+
+        let mut db = home.cobs_db_mut().unwrap();
+        db.migrate(radicle::cob::migrate::ignore).unwrap();
+
+        policy::Store::open(policies_db).unwrap();
+        home.database_mut()
+            .unwrap()
+            .init(
+                &keypair.pk.into(),
+                config.node.features(),
+                &alias,
+                &UserAgent::default(),
+                LocalTime::now().into(),
+                config.node.external_addresses.iter(),
+            )
+            .unwrap();
+
+        radicle::cob::cache::Store::open(cobs_db).unwrap();
+
+        transport::local::register(storage.clone());
+        let keystore = Keystore::new(&home.keys());
+        keystore.store(keypair.clone(), "radicle", None).unwrap();
+
+        // Ensures that each user has a unique but deterministic public key.
+        self.users += 1;
+
+        Profile {
+            home,
+            storage,
+            keystore,
+            public_key: keypair.pk.into(),
+            config,
+        }
+    }
+
+    /// Create a new profile using a the default configuration from [`Environment::config`].
+    pub fn profile(&mut self, alias: &'static str) -> Profile {
+        self.profile_with(self.config(alias))
+    }
+
+    /// Create a new node in this environment. This should be used when a running node
+    /// is required. Use [`Environment::profile`] otherwise.
+    /// Using this function is only necessary when the node configuration differs
+    /// from the standard ones ([`config::node`], [`config::relay`], [`config::seed`]),
+    /// as for each of them a convenience function
+    /// (resp. [`Environment::node`], [`Environment::relay`], [`Environment::seed`]).
+    /// is provided to reduce boilerplate.
+    pub fn node_with(&mut self, node: Config) -> Node<MemorySigner> {
+        let alias = node.alias.clone();
+        let profile = self.profile_with(profile::Config {
+            node,
+            ..self.config(alias.as_ref())
+        });
+        Node::new(profile)
+    }
+
+    /// Convenience method for creating a [`Node<MemorySigner>`]
+    /// using configuration [`config::node`] within this [`Environment`].
+    pub fn node(&mut self, alias: &'static str) -> Node<MemorySigner> {
+        self.node_with(config::node(alias))
+    }
+
+    /// Convenience method for creating a [`Node<MemorySigner>`]
+    /// using configuration [`config::relay`] within this [`Environment`].
+    pub fn relay(&mut self, alias: &'static str) -> Node<MemorySigner> {
+        self.node_with(config::relay(alias))
+    }
+
+    /// Convenience method for creating a [`Node<MemorySigner>`]
+    /// using configuration [`config::seed`] within this [`Environment`].
+    pub fn seed(&mut self, alias: &'static str) -> Node<MemorySigner> {
+        self.node_with(config::seed(alias))
+    }
+
+    /// Convenience method for placing repository fixture.
+    pub fn repository(
+        &self,
+        has_alias: &impl HasAlias,
+    ) -> (radicle_cli::git::Repository, radicle_cli::git::Oid) {
+        radicle::test::fixtures::repository(self.work(has_alias).as_path())
+    }
+
+    // Convenience method for exectuing a test formula with standard configuration.
+    pub fn test(
+        &self,
+        test_file: &'static str,
+        subject: &(impl HasAlias + HasHome),
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        formula(
+            self.work(subject).as_ref(),
+            PathBuf::from("examples").join(test_file.to_owned() + ".md"),
+        )?
+        .env(
+            "RAD_HOME",
+            subject.home().path().to_path_buf().to_string_lossy(),
+        )
+        .run()?;
+
+        Ok(())
+    }
+
+    pub fn tests(
+        &self,
+        test_files: impl IntoIterator<Item = &'static str>,
+        subject: &(impl HasAlias + HasHome),
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        for test_file in test_files {
+            self.test(test_file, subject)?;
+        }
+
+        Ok(())
+    }
+
+    /// Convenience method for creating exactly one profile with alias "alice"
+    /// and running tests within it.
+    pub fn alice(test_files: impl IntoIterator<Item = &'static str>) {
+        let mut env = Environment::new();
+        let alice = env.profile("alice");
+        env.repository(&alice);
+        env.tests(test_files, &alice).unwrap();
+    }
+}
+
+pub trait HasAlias {
+    fn alias(&self) -> &Alias;
+}
+
+impl HasAlias for Node<MemorySigner> {
+    fn alias(&self) -> &Alias {
+        &self.config.alias
+    }
+}
+
+impl HasAlias for Profile {
+    fn alias(&self) -> &Alias {
+        self.config.alias()
+    }
+}
+
+impl<G> HasAlias for NodeHandle<G> {
+    fn alias(&self) -> &Alias {
+        &self.alias
+    }
+}
+
+pub trait HasHome {
+    fn home(&self) -> &Home;
+}
+
+impl HasHome for Profile {
+    fn home(&self) -> &Home {
+        &self.home
+    }
+}
+
+impl HasHome for Node<MemorySigner> {
+    fn home(&self) -> &Home {
+        &self.home
+    }
+}
+
+impl HasHome for NodeHandle<MemorySigner> {
+    fn home(&self) -> &Home {
+        &self.home
+    }
+}

crates/radicle-cli/tests/util/formula.rs

@@ -0,0 +1,39 @@
+use std::path::Path;
+
+use radicle::profile::env;
+
+use radicle_cli_test::TestFormula;
+
+pub(crate) fn formula(
+    root: &Path,
+    test: impl AsRef<Path>,
+) -> Result<TestFormula, Box<dyn std::error::Error>> {
+    const RAD_SEED: &str = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff";
+
+    let mut formula = TestFormula::new(root.to_path_buf());
+    let base = Path::new(env!("CARGO_MANIFEST_DIR"));
+
+    formula
+        .env("GIT_AUTHOR_DATE", "1671125284")
+        .env("GIT_AUTHOR_EMAIL", "radicle@localhost")
+        .env("GIT_AUTHOR_NAME", "radicle")
+        .env("GIT_COMMITTER_DATE", "1671125284")
+        .env("GIT_COMMITTER_EMAIL", "radicle@localhost")
+        .env("GIT_COMMITTER_NAME", "radicle")
+        .env("EDITOR", "true")
+        .env("TZ", "UTC")
+        .env("LANG", "C")
+        .env("USER", "alice")
+        .env(env::RAD_PASSPHRASE, "radicle")
+        .env(env::RAD_KEYGEN_SEED, RAD_SEED)
+        .env(env::RAD_RNG_SEED, "0")
+        .env(env::RAD_LOCAL_TIME, "1671125284")
+        .envs(radicle::git::env::GIT_DEFAULT_CONFIG)
+        .build(&[
+            ("radicle-remote-helper", "git-remote-rad"),
+            ("radicle-cli", "rad"),
+        ])
+        .file(base.join(test))?;
+
+    Ok(formula)
+}

crates/radicle-cli/tests/util/mod.rs

@@ -0,0 +1,2 @@
+pub(crate) mod environment;
+pub(crate) mod formula;

crates/radicle-node/src/test.rs

@@ -1,7 +1,7 @@
 pub mod arbitrary;
-pub mod environment;
 pub mod gossip;
 pub mod handle;
+pub mod node;
 pub mod peer;
 pub mod simulator;
 

crates/radicle-node/src/test/node.rs

@@ -0,0 +1,532 @@
+use std::io::BufRead as _;
+use std::mem::ManuallyDrop;
+use std::path::Path;
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    fs, io, iter, net, process, thread, time,
+    time::Duration,
+};
+
+use crossbeam_channel as chan;
+
+use radicle::cob;
+use radicle::cob::issue;
+use radicle::crypto::signature::Signer;
+use radicle::crypto::ssh::keystore::MemorySigner;
+use radicle::crypto::test::signer::MockSigner;
+use radicle::crypto::Signature;
+use radicle::git;
+use radicle::git::refname;
+use radicle::identity::{RepoId, Visibility};
+use radicle::node::config::ConnectAddress;
+use radicle::node::policy::store as policy;
+use radicle::node::seed::Store as _;
+use radicle::node::Config;
+use radicle::node::{self, Alias};
+use radicle::node::{ConnectOptions, Handle as _};
+use radicle::node::{Database, POLICIES_DB_FILE};
+use radicle::profile::{env, Home, Profile};
+use radicle::rad;
+use radicle::storage::{ReadStorage as _, RemoteRepository as _, SignRepository as _};
+use radicle::test::fixtures;
+use radicle::Storage;
+
+use crate::node::device::Device;
+use crate::node::NodeId;
+use crate::service::Event;
+use crate::storage::git::transport;
+use crate::{runtime, runtime::Handle, service, Runtime};
+
+/// A node that can be run.
+pub struct Node<G> {
+    pub id: NodeId,
+    pub home: Home,
+    pub signer: Device<G>,
+    pub storage: Storage,
+    pub config: Config,
+    pub db: service::Stores<Database>,
+    pub policies: policy::Store<policy::Write>,
+}
+
+impl Node<MemorySigner> {
+    pub fn new(profile: Profile) -> Self {
+        let signer = Device::from(MemorySigner::load(&profile.keystore, None).unwrap());
+        let id = *profile.id();
+        let policies_db = profile.home.node().join(POLICIES_DB_FILE);
+        let policies = policy::Store::open(policies_db).unwrap();
+        let db = profile.database_mut().unwrap();
+        let db = service::Stores::from(db);
+
+        Node {
+            id,
+            home: profile.home,
+            config: profile.config.node,
+            signer,
+            db,
+            policies,
+            storage: profile.storage,
+        }
+    }
+}
+
+/// Handle to a running node.
+pub struct NodeHandle<G: 'static> {
+    pub id: NodeId,
+    pub alias: Alias,
+    pub storage: Storage,
+    pub signer: Device<G>,
+    pub home: Home,
+    pub addr: net::SocketAddr,
+    pub thread: ManuallyDrop<thread::JoinHandle<Result<(), runtime::Error>>>,
+    pub handle: ManuallyDrop<Handle>,
+}
+
+impl<G: 'static> Drop for NodeHandle<G> {
+    fn drop(&mut self) {
+        log::debug!(target: "test", "Node {} shutting down..", self.id);
+
+        unsafe { ManuallyDrop::take(&mut self.handle) }
+            .shutdown()
+            .unwrap();
+        unsafe { ManuallyDrop::take(&mut self.thread) }
+            .join()
+            .unwrap()
+            .unwrap();
+    }
+}
+
+impl<G: Signer<Signature> + cyphernet::Ecdh> NodeHandle<G> {
+    /// Connect this node to another node, and wait for the connection to be established both ways.
+    pub fn connect(&mut self, remote: &NodeHandle<G>) -> &mut Self {
+        let local_events = self.handle.events();
+        let remote_events = remote.handle.events();
+
+        self.handle
+            .connect(remote.id, remote.addr.into(), ConnectOptions::default())
+            .ok();
+
+        local_events
+            .iter()
+            .find(|e| {
+                matches!(
+                    e, Event::PeerConnected { nid } if nid == &remote.id
+                )
+            })
+            .unwrap();
+        remote_events
+            .iter()
+            .find(|e| {
+                matches!(
+                    e, Event::PeerConnected { nid } if nid == &self.id
+                )
+            })
+            .unwrap();
+
+        self
+    }
+
+    pub fn disconnect(&mut self, remote: &NodeHandle<G>) {
+        self.handle.disconnect(remote.id).unwrap();
+    }
+
+    /// Shutdown node.
+    pub fn shutdown(self) {
+        drop(self)
+    }
+
+    /// Get the full address of this node.
+    pub fn address(&self) -> ConnectAddress {
+        (self.id, node::Address::from(self.addr)).into()
+    }
+
+    /// Get routing table entries.
+    pub fn routing(&self) -> impl Iterator<Item = (RepoId, NodeId)> {
+        use node::routing::Store as _;
+
+        self.home.routing_mut().unwrap().entries().unwrap()
+    }
+
+    pub fn inventory(&self) -> impl Iterator<Item = RepoId> + '_ {
+        self.routing()
+            .filter(|(_, n)| *n == self.id)
+            .map(|(r, _)| r)
+    }
+
+    /// 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,
+        remotes: impl IntoIterator<Item = &'a NodeHandle<G>>,
+    ) -> BTreeSet<(RepoId, NodeId)> {
+        converge(iter::once(self).chain(remotes))
+    }
+
+    /// Wait until this node's routing table contains the given routes.
+    #[track_caller]
+    pub fn routes_to(&self, routes: &[(RepoId, NodeId)]) {
+        log::debug!(target: "test", "Waiting for {} to route to {:?}", self.id, routes);
+        let events = self.handle.events();
+
+        loop {
+            let mut remaining: BTreeSet<_> = routes.iter().collect();
+
+            for (rid, nid) in self.routing() {
+                if !remaining.remove(&(rid, nid)) {
+                    log::debug!(target: "test", "Found unexpected route for {}: ({rid}, {nid})", self.id);
+                }
+            }
+            if remaining.is_empty() {
+                break;
+            }
+            events
+                .wait(
+                    |e| matches!(e, Event::SeedDiscovered { .. }).then_some(()),
+                    time::Duration::from_secs(6),
+                )
+                .unwrap();
+        }
+    }
+
+    /// 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 a repository.
+    #[track_caller]
+    pub fn has_repository(&self, rid: &RepoId) {
+        log::debug!(target: "test", "Waiting for {} to have {rid}", self.id);
+        let events = self.handle.events();
+
+        loop {
+            if self.storage.repository(*rid).is_ok() {
+                log::debug!(target: "test", "Node {} has {rid}", self.id);
+                break;
+            }
+            events
+                .wait(
+                    |e| matches!(e, Event::RefsFetched { .. }).then_some(()),
+                    time::Duration::from_secs(6),
+                )
+                .unwrap();
+        }
+    }
+
+    /// Wait until this node has the inventory of another node.
+    #[track_caller]
+    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();
+
+        loop {
+            if let Ok(repo) = self.storage.repository(*rid) {
+                if repo.remote(nid).is_ok() {
+                    log::debug!(target: "test", "Node {} has {rid}/{nid}", self.id);
+                    break;
+                }
+            }
+            events
+                .wait(
+                    |e| matches!(e, Event::RefsFetched { .. }).then_some(()),
+                    time::Duration::from_secs(6),
+                )
+                .unwrap();
+        }
+    }
+
+    /// Clone a repo into a directory.
+    pub fn clone<P: AsRef<Path>>(&self, rid: RepoId, cwd: P) -> io::Result<()> {
+        self.rad("clone", &[rid.to_string().as_str()], cwd)
+    }
+
+    /// Fork a repo.
+    pub fn fork<P: AsRef<Path>>(&self, rid: RepoId, cwd: P) -> io::Result<()> {
+        self.clone(rid, &cwd)?;
+        self.rad("fork", &[rid.to_string().as_str()], &cwd)?;
+        self.announce(rid, 1, &cwd)?;
+
+        Ok(())
+    }
+
+    /// Announce a repo.
+    pub fn announce<P: AsRef<Path>>(&self, rid: RepoId, replicas: usize, cwd: P) -> io::Result<()> {
+        self.rad(
+            "sync",
+            &[
+                rid.to_string().as_str(),
+                "--announce",
+                "--replicas",
+                replicas.to_string().as_str(),
+            ],
+            cwd,
+        )
+    }
+
+    /// Init a repo.
+    pub fn init<P: AsRef<Path>>(&self, name: &str, desc: &str, cwd: P) -> io::Result<()> {
+        self.rad(
+            "init",
+            &[
+                "--name",
+                name,
+                "--description",
+                desc,
+                "--default-branch",
+                "master",
+                "--public",
+            ],
+            cwd,
+        )
+    }
+
+    /// Run a `rad` CLI command.
+    pub fn rad<P: AsRef<Path>>(&self, cmd: &str, args: &[&str], cwd: P) -> io::Result<()> {
+        let cwd = cwd.as_ref();
+        log::debug!(target: "test", "Running `rad {cmd} {args:?}` in {}..", cwd.display());
+
+        fs::create_dir_all(cwd)?;
+
+        let result = process::Command::new(snapbox::cmd::cargo_bin("rad"))
+            .env_clear()
+            .envs(env::vars().filter(|(k, _)| k == "PATH"))
+            .env("GIT_AUTHOR_DATE", "1671125284")
+            .env("GIT_AUTHOR_EMAIL", "radicle@localhost")
+            .env("GIT_AUTHOR_NAME", "radicle")
+            .env("GIT_COMMITTER_DATE", "1671125284")
+            .env("GIT_COMMITTER_EMAIL", "radicle@localhost")
+            .env("GIT_COMMITTER_NAME", "radicle")
+            .env(
+                env::RAD_HOME,
+                self.home.path().to_string_lossy().to_string(),
+            )
+            .env(env::RAD_PASSPHRASE, "radicle")
+            .env(env::RAD_LOCAL_TIME, "1671125284")
+            .env("TZ", "UTC")
+            .env("LANG", "C")
+            .envs(git::env::GIT_DEFAULT_CONFIG)
+            .current_dir(cwd)
+            .arg(cmd)
+            .args(args)
+            .output()?;
+
+        for line in io::BufReader::new(io::Cursor::new(&result.stdout))
+            .lines()
+            .map_while(Result::ok)
+        {
+            log::debug!(target: "test", "rad {cmd}: {line}");
+        }
+
+        log::debug!(
+            target: "test",
+            "Ran command `rad {cmd}` (status={})", result.status.code().unwrap()
+        );
+
+        if !result.status.success() {
+            return Err(io::ErrorKind::Other.into());
+        }
+        Ok(())
+    }
+
+    /// Create an [`issue::Issue`] in the `NodeHandle`'s storage.
+    pub fn issue(&self, rid: RepoId, title: &str, desc: &str) -> cob::ObjectId {
+        let repo = self.storage.repository(rid).unwrap();
+        let mut issues = issue::Cache::no_cache(&repo).unwrap();
+        *issues
+            .create(title, desc, &[], &[], [], &self.signer)
+            .unwrap()
+            .id()
+    }
+}
+
+impl Node<MockSigner> {
+    /// Create a new node.
+    pub fn init(base: &Path, config: Config) -> Self {
+        let home = base.join(
+            iter::repeat_with(fastrand::alphanumeric)
+                .take(8)
+                .collect::<String>(),
+        );
+        let home = Home::new(home).unwrap();
+        let signer = Device::mock();
+        let storage = Storage::open(
+            home.storage(),
+            git::UserInfo {
+                alias: config.alias.clone(),
+                key: *signer.public_key(),
+            },
+        )
+        .unwrap();
+        let policies = home.policies_mut().unwrap();
+        let db = home.database_mut().unwrap();
+        let db = service::Stores::from(db);
+
+        log::debug!(target: "test", "Node::init {}: {}", config.alias, signer.public_key());
+        Self {
+            id: *signer.public_key(),
+            home,
+            signer,
+            storage,
+            config,
+            db,
+            policies,
+        }
+    }
+}
+
+impl<G: cyphernet::Ecdh<Pk = NodeId> + Signer<Signature> + Clone> Node<G> {
+    /// Spawn a node in its own thread.
+    pub fn spawn(self) -> NodeHandle<G> {
+        let alias = self.config.alias.clone();
+        let listen = vec![([0, 0, 0, 0], 0).into()];
+        let (_, signals) = chan::bounded(1);
+        let rt = Runtime::init(
+            self.home.clone(),
+            self.config,
+            listen,
+            signals,
+            self.signer.clone(),
+        )
+        .unwrap();
+        let addr = *rt.local_addrs.first().unwrap();
+        let id = *self.signer.public_key();
+        let handle = ManuallyDrop::new(rt.handle.clone());
+        let thread = ManuallyDrop::new(runtime::thread::spawn(&id, "runtime", move || rt.run()));
+
+        NodeHandle {
+            id,
+            alias,
+            storage: self.storage,
+            signer: self.signer,
+            home: self.home,
+            addr,
+            handle,
+            thread,
+        }
+    }
+
+    /// Populate a storage instance with a project from the given repository.
+    pub fn project_from(
+        &mut self,
+        name: &str,
+        description: &str,
+        repo: &git::raw::Repository,
+    ) -> RepoId {
+        transport::local::register(self.storage.clone());
+
+        let branch = refname!("master");
+        let id = rad::init(
+            repo,
+            name.try_into().unwrap(),
+            description,
+            branch.clone(),
+            Visibility::default(),
+            &self.signer,
+            &self.storage,
+        )
+        .map(|(id, _, _)| id)
+        .unwrap();
+
+        assert!(self.policies.seed(&id, node::policy::Scope::All).unwrap());
+
+        log::debug!(
+            target: "test",
+            "Initialized project {id} for node {}", self.signer.public_key()
+        );
+
+        // Push local branches to storage.
+        let mut refs = Vec::<(git::Qualified, git::Qualified)>::new();
+        for branch in repo.branches(Some(git::raw::BranchType::Local)).unwrap() {
+            let (branch, _) = branch.unwrap();
+            let name = git::RefString::try_from(branch.name().unwrap().unwrap()).unwrap();
+
+            refs.push((
+                git::lit::refs_heads(&name).into(),
+                git::lit::refs_heads(&name).into(),
+            ));
+        }
+        git::push(repo, "rad", refs.iter().map(|(a, b)| (a, b))).unwrap();
+
+        radicle::git::set_upstream(
+            repo,
+            &*radicle::rad::REMOTE_NAME,
+            branch.clone(),
+            radicle::git::refs::workdir::branch(&branch),
+        )
+        .unwrap();
+
+        self.storage
+            .repository(id)
+            .unwrap()
+            .sign_refs(&self.signer)
+            .unwrap();
+
+        id
+    }
+
+    /// Populate a storage instance with a project.
+    pub fn project(&mut self, name: &str, description: &str) -> RepoId {
+        let tmp = tempfile::tempdir().unwrap();
+        let (repo, _) = fixtures::repository(tmp.path());
+
+        self.project_from(name, description, &repo)
+    }
+}
+
+/// Checks whether the nodes have converged in their routing tables.
+#[track_caller]
+pub fn converge<'a, G: Signer<Signature> + cyphernet::Ecdh + 'static>(
+    nodes: impl IntoIterator<Item = &'a NodeHandle<G>>,
+) -> BTreeSet<(RepoId, NodeId)> {
+    let nodes = nodes.into_iter().collect::<Vec<_>>();
+
+    let mut all_routes = BTreeSet::<(RepoId, NodeId)>::new();
+    let mut remaining = BTreeMap::from_iter(nodes.iter().map(|node| (node.id, node)));
+
+    // First build the set of all routes.
+    for node in &nodes {
+        // Routes from the routing table.
+        for (rid, seed_id) in node.routing() {
+            all_routes.insert((rid, seed_id));
+        }
+        // Routes from the local inventory.
+        for rid in node.inventory() {
+            all_routes.insert((rid, node.id));
+        }
+    }
+
+    // Then, while there are nodes remaining to converge, check each node to see if
+    // its routing table has all routes. If so, remove it from the remaining nodes.
+    while !remaining.is_empty() {
+        remaining.retain(|_, node| {
+            let routing = node.routing();
+            let routes = BTreeSet::from_iter(routing);
+
+            if routes.is_superset(&all_routes) {
+                log::debug!(target: "test", "Node {} has converged", node.id);
+                return false;
+            } else {
+                let diff = all_routes.symmetric_difference(&routes).collect::<Vec<_>>();
+                log::debug!(target: "test", "Node has missing routes: {diff:?}");
+            }
+            true
+        });
+        thread::sleep(Duration::from_millis(100));
+    }
+    all_routes
+}

crates/radicle-node/src/tests/e2e.rs

@@ -15,8 +15,8 @@ use crate::node::{Config, ConnectOptions};
 use crate::service;
 use crate::service::policy::Scope;
 use crate::storage::git::transport;
-use crate::test::environment::{converge, Environment, Node};
 use crate::test::logger;
+use crate::test::node::{converge, Node};
 
 mod config {
     use super::*;
@@ -29,6 +29,17 @@ mod config {
             ..Config::test(Alias::new(alias))
         }
     }
+
+    /// Get the scale or "test size". This is used to scale tests with more
+    /// data. Defaults to `1`.
+    pub fn scale() -> usize {
+        std::env::var("RAD_TEST_SCALE")
+            .map(|s| {
+                s.parse()
+                    .expect("repository: invalid value for `RAD_TEST_SCALE`")
+            })
+            .unwrap_or(1)
+    }
 }
 
 #[test]
@@ -657,12 +668,11 @@ fn test_fetch_unseeded() {
 fn test_large_fetch() {
     logger::init(log::Level::Debug);
 
-    let env = Environment::new();
-    let scale = env.scale();
-    let mut alice = Node::init(&env.tmp(), config::relay("alice"));
-    let bob = Node::init(&env.tmp(), config::relay("bob"));
-
     let tmp = tempfile::tempdir().unwrap();
+    let scale = config::scale();
+    let mut alice = Node::init(tmp.path(), config::relay("alice"));
+    let bob = Node::init(tmp.path(), config::relay("bob"));
+
     let (repo, _) = fixtures::repository(tmp.path());
     fixtures::populate(&repo, scale.max(3));
 
@@ -695,8 +705,8 @@ fn test_large_fetch() {
 fn test_concurrent_fetches() {
     logger::init(log::Level::Debug);
 
-    let env = Environment::new();
-    let scale = env.scale();
+    let tmp = tempfile::tempdir().unwrap();
+    let scale = config::scale();
     let repos = scale.max(4);
     let limits = Limits {
         // Have one fetch be queued.
@@ -706,7 +716,7 @@ fn test_concurrent_fetches() {
     let mut bob_repos = HashSet::new();
     let mut alice_repos = HashSet::new();
     let mut alice = Node::init(
-        &env.tmp(),
+        tmp.path(),
         service::Config {
             limits: limits.clone(),
             relay: radicle::node::config::Relay::Always,
@@ -714,7 +724,7 @@ fn test_concurrent_fetches() {
         },
     );
     let mut bob = Node::init(
-        &env.tmp(),
+        tmp.path(),
         service::Config {
             limits,
             relay: radicle::node::config::Relay::Always,