add unit tests

Author: sky-mart

Cargo.lock

@@ -34,6 +34,7 @@ orb-build-info = { workspace = true, features = ["build-script"] }
 
 [dev-dependencies]
 dbus-launch = "0.2.0"
+portpicker = "0.1.1"
 tokio = { workspace = true, features = ["sync", "test-util"] }
 
 [package.metadata.deb]

supervisor/Cargo.toml

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{path::PathBuf, time::Duration};
 
 use color_eyre::eyre::WrapErr as _;
 use futures::{future::TryFutureExt as _, FutureExt as _};
@@ -13,6 +13,7 @@ use crate::{
         SIGNUP_PROXY_DEFAULT_OBJECT_PATH, SIGNUP_PROXY_DEFAULT_WELL_KNOWN_NAME,
     },
     tasks,
+    tasks::zoci::GONDOR_BIN,
 };
 
 pub const DBUS_WELL_KNOWN_NAME: &str = "org.worldcoin.OrbSupervisor1";
@@ -41,6 +42,7 @@ pub struct Settings {
     pub well_known_name: String,
     pub download_throttle: Duration,
     pub stop_core_after_signup: Duration,
+    pub gondor_bin: PathBuf,
 }
 
 impl Settings {
@@ -55,6 +57,7 @@ impl Settings {
             well_known_name: DBUS_WELL_KNOWN_NAME.to_string(),
             download_throttle: DEFAULT_DURATION_TO_ALLOW_DOWNLOADS,
             stop_core_after_signup: DURATION_TO_STOP_CORE_AFTER_LAST_SIGNUP,
+            gondor_bin: PathBuf::from(GONDOR_BIN),
         }
     }
 }
@@ -151,9 +154,10 @@ impl Application {
             tasks::spawn_signup_started_task(&self.settings, &self.session_connection)
                 .await?;
 
-        let zoci_handles = tasks::spawn_zoci_receiver(&self.zenorb)
-            .await
-            .wrap_err("failed to spawn zoci receiver")?;
+        let zoci_handles =
+            tasks::spawn_zoci_receiver(&self.zenorb, self.settings.gondor_bin.clone())
+                .await
+                .wrap_err("failed to spawn zoci receiver")?;
 
         let ((),) = tokio::try_join!(
             // All tasks are joined here

supervisor/src/startup.rs

@@ -1,3 +1,5 @@
+use std::path::PathBuf;
+
 use color_eyre::{
     eyre::{eyre, WrapErr},
     Result,
@@ -7,20 +9,21 @@ use tokio::task::JoinHandle;
 use tracing::{info, instrument};
 use zenorb::{zenoh::query::Query, zoci::ZociQueryExt, Zenorb};
 
-const GONDOR_BIN: &str = "/usr/local/bin/gondor-calls-for-ota";
+pub const GONDOR_BIN: &str = "/usr/local/bin/gondor-calls-for-ota";
 
-pub async fn spawn_zoci_receiver(zenorb: &Zenorb) -> Result<Vec<JoinHandle<()>>> {
+pub async fn spawn_zoci_receiver(
+    zenorb: &Zenorb,
+    gondor_bin: PathBuf,
+) -> Result<Vec<JoinHandle<()>>> {
     zenorb
-        .receiver(())
-        .queryable("job/gondor-calls-for-ota", |(), query| {
-            gondor_calls_for_ota(query)
-        })
+        .receiver(gondor_bin)
+        .queryable("job/gondor-calls-for-ota", gondor_calls_for_ota)
         .run()
         .await
 }
 
 #[instrument(skip(query))]
-async fn gondor_calls_for_ota(query: Query) -> Result<()> {
+async fn gondor_calls_for_ota(gondor_bin: PathBuf, query: Query) -> Result<()> {
     let response = async {
         let version = query.payload_str()?;
         let version = version.trim();
@@ -29,9 +32,9 @@ async fn gondor_calls_for_ota(query: Query) -> Result<()> {
             return Err(eyre!("missing target version"));
         }
 
-        info!("running {GONDOR_BIN} {version}");
+        info!("running {} {version}", gondor_bin.display());
 
-        let output = Command::new(GONDOR_BIN)
+        let output = Command::new(&gondor_bin)
             .arg(version)
             .output()
             .await

supervisor/src/tasks/zoci.rs

@@ -57,15 +57,22 @@ pub fn make_settings(dbus_instances: &DbusInstances) -> Settings {
 
 pub async fn spawn_supervisor_service(
     settings: Settings,
+    zenorb: Zenorb,
 ) -> color_eyre::Result<Application> {
     Lazy::force(&TRACING);
-    let zenorb = Zenorb::from_cfg(isolated_zenoh_cfg())
+    let application = Application::build(settings.clone(), zenorb).await?;
+    Ok(application)
+}
+
+/// A self-contained `Zenorb` that doesn't connect to anything. Suitable for tests
+/// that only need to satisfy the `Application::build` signature without exercising
+/// any zenoh-backed behavior.
+pub async fn isolated_supervisor_zenorb() -> color_eyre::Result<Zenorb> {
+    Zenorb::from_cfg(isolated_zenoh_cfg())
         .liveliness(false)
         .orb_id(OrbId::from_str("ea2ea744").unwrap())
         .with_name("supervisor")
-        .await?;
-    let application = Application::build(settings.clone(), zenorb).await?;
-    Ok(application)
+        .await
 }
 
 fn isolated_zenoh_cfg() -> zenoh::Config {
@@ -76,6 +83,34 @@ fn isolated_zenoh_cfg() -> zenoh::Config {
     cfg
 }
 
+/// Spins up an in-process zenoh router on a free loopback port and returns a pair of
+/// `Zenorb` clients (named via the supplied arguments) that share the same orb id and
+/// communicate through that router. The returned `zenoh::Session` is the router and
+/// must be kept alive for the duration of the test.
+pub async fn spawn_zenoh_router_and_clients(
+    name_a: &str,
+    name_b: &str,
+) -> color_eyre::Result<(zenoh::Session, Zenorb, Zenorb)> {
+    let port = portpicker::pick_unused_port().expect("no free ports");
+    let router = zenoh::open(zenorb::router_cfg(port))
+        .await
+        .map_err(|e| color_eyre::eyre::eyre!("{e}"))?;
+
+    let orb_id = OrbId::from_str("ea2ea744").unwrap();
+    let zenorb_a = Zenorb::from_cfg(zenorb::client_cfg(port))
+        .liveliness(false)
+        .orb_id(orb_id.clone())
+        .with_name(name_a)
+        .await?;
+    let zenorb_b = Zenorb::from_cfg(zenorb::client_cfg(port))
+        .liveliness(false)
+        .orb_id(orb_id)
+        .with_name(name_b)
+        .await?;
+
+    Ok((router, zenorb_a, zenorb_b))
+}
+
 #[proxy(
     interface = "org.worldcoin.OrbSupervisor1.Manager",
     gen_async = true,

supervisor/tests/it/helpers.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{path::PathBuf, time::Duration};
 
 use tap::TapFallible;
 use tracing::error;
@@ -11,8 +11,10 @@ async fn supervisor_disallows_downloads_if_signup_started_received(
     let dbus_instances = helpers::launch_dbuses().await??;
 
     let settings = helpers::make_settings(&dbus_instances);
+    let zenorb = helpers::isolated_supervisor_zenorb().await?;
 
-    let application = helpers::spawn_supervisor_service(settings.clone()).await?;
+    let application =
+        helpers::spawn_supervisor_service(settings.clone(), zenorb).await?;
     let _application_handle = tokio::spawn(application.run());
 
     let update_agent_proxy =
@@ -47,7 +49,9 @@ async fn supervisor_stops_orb_core_when_update_permission_is_requested(
     let dbus_instances = helpers::launch_dbuses().await??;
 
     let settings = helpers::make_settings(&dbus_instances);
-    let application = helpers::spawn_supervisor_service(settings.clone()).await?;
+    let zenorb = helpers::isolated_supervisor_zenorb().await?;
+    let application =
+        helpers::spawn_supervisor_service(settings.clone(), zenorb).await?;
 
     let _application_handle = tokio::spawn(application.run());
 
@@ -91,3 +95,33 @@ async fn supervisor_stops_orb_core_when_update_permission_is_requested(
 
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn application_serves_gondor_zoci_handler_end_to_end() -> color_eyre::Result<()> {
+    let dbus_instances = helpers::launch_dbuses().await??;
+
+    let (_router, supervisor_zenorb, client_zenorb) =
+        helpers::spawn_zenoh_router_and_clients("supervisor", "test-client").await?;
+
+    let mut settings = helpers::make_settings(&dbus_instances);
+    settings.gondor_bin = PathBuf::from("/bin/true");
+
+    let application =
+        helpers::spawn_supervisor_service(settings, supervisor_zenorb).await?;
+    let _application_handle = tokio::spawn(application.run());
+
+    // Give Application::run a beat to register its zoci queryable on the router.
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    let reply = client_zenorb
+        .command_raw("supervisor/job/gondor-calls-for-ota", "v1.0.0")
+        .await?;
+
+    if let Err(reply_err) = reply {
+        let payload =
+            String::from_utf8_lossy(&reply_err.payload().to_bytes()).into_owned();
+        panic!("expected success reply, got error: {payload}");
+    }
+
+    Ok(())
+}