tests: zenoh: Add bidirectional integration tests

Author: joaoantoniocardoso

tests/zenoh.rs

@@ -0,0 +1,10 @@
+mod common;
+
+#[path = "zenoh/mod.rs"]
+mod zenoh;
+
+#[path = "zenoh/json.rs"]
+mod json;
+
+#[path = "zenoh/raw.rs"]
+mod raw;

tests/zenoh/json.rs

@@ -0,0 +1,67 @@
+use std::{sync::Arc, time::Duration};
+
+use anyhow::{Context, Result};
+use clap::Parser;
+use mavlink_server::{cli, drivers::zenoh::json::Zenoh, hub, stats};
+use tracing::*;
+
+use crate::{
+    common::{pick_free_port, wait_for_stats_collection},
+    zenoh::{spawn_zenoh_router, wait_for_router, write_zenoh_client_config},
+};
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 10)]
+async fn test_zenoh_bidirectional() -> Result<()> {
+    let driver_name = "Zenoh";
+    let port = pick_free_port();
+    let _router = spawn_zenoh_router(port, "mavlink").await;
+    wait_for_router().await;
+
+    let zenoh_config = write_zenoh_client_config(port)?;
+    let zenoh_config = zenoh_config
+        .to_str()
+        .context("Zenoh config path is not valid UTF-8")?;
+
+    cli::init_with(cli::Args::parse_from(vec![
+        &std::env::args().next().unwrap_or_default(),
+        "--allow-no-endpoints",
+        "--zenoh-config-file",
+        zenoh_config,
+        "--mavlink-heartbeat-frequency",
+        "10",
+    ]));
+
+    hub::add_driver(Arc::new(Zenoh::builder(driver_name).build())).await?;
+
+    stats::set_period(Duration::from_millis(100)).await?;
+    wait_for_stats_collection().await;
+
+    let mut matching_drivers = 0;
+    for (_uuid, driver) in stats::drivers_stats().await? {
+        if *driver.name != driver_name {
+            continue;
+        }
+
+        matching_drivers += 1;
+        assert!(
+            driver.stats.input.is_some(),
+            "{driver_name} should receive mavlink over zenoh"
+        );
+        assert!(
+            driver.stats.output.is_some(),
+            "{driver_name} should publish mavlink over zenoh"
+        );
+    }
+
+    assert_eq!(
+        matching_drivers, 1,
+        "expected one {driver_name} driver in stats"
+    );
+
+    for (id, driver_info) in hub::drivers().await? {
+        debug!("Removing driver id {id:?} ({driver_info:?})");
+        hub::remove_driver(id).await?;
+    }
+
+    Ok(())
+}

tests/zenoh/mod.rs

@@ -0,0 +1,72 @@
+use std::{path::PathBuf, time::Duration};
+
+use anyhow::{Context, Result};
+use zenoh;
+
+pub async fn spawn_zenoh_router(port: u16, topic_prefix: &str) -> tokio::task::JoinHandle<()> {
+    let topic_prefix = topic_prefix.to_string();
+
+    tokio::spawn(async move {
+        let mut config = zenoh::Config::default();
+        config
+            .insert_json5("mode", r#""router""#)
+            .expect("Failed to insert router mode");
+        config
+            .insert_json5("listen/endpoints", &format!(r#"["tcp/127.0.0.1:{port}"]"#))
+            .expect("Failed to insert listen endpoints");
+
+        let session = zenoh::open(config)
+            .await
+            .expect("Failed to start zenoh router for tests");
+
+        spawn_zenoh_loopback(session, &topic_prefix).await;
+
+        std::future::pending::<()>().await
+    })
+}
+
+async fn spawn_zenoh_loopback(session: zenoh::Session, topic_prefix: &str) {
+    let out_topic = format!("{topic_prefix}/out");
+    let in_topic = format!("{topic_prefix}/in");
+
+    tokio::spawn(async move {
+        let subscriber = session
+            .declare_subscriber(&out_topic)
+            .await
+            .unwrap_or_else(|error| panic!("Failed to subscribe to {out_topic}: {error:?}"));
+        let publisher = session
+            .declare_publisher(&in_topic)
+            .await
+            .unwrap_or_else(|error| panic!("Failed to publish on {in_topic}: {error:?}"));
+
+        while let Ok(sample) = subscriber.recv_async().await {
+            if let Err(error) = publisher.put(sample.payload().to_bytes()).await {
+                panic!("Failed to loop back {out_topic} -> {in_topic}: {error:?}");
+            }
+        }
+    });
+}
+
+pub fn write_zenoh_client_config(port: u16) -> Result<PathBuf> {
+    let path = std::env::temp_dir().join(format!(
+        "mavlink-server-zenoh-test-{}-{port}.json5",
+        std::process::id()
+    ));
+
+    std::fs::write(
+        &path,
+        format!(
+            r#"{{
+  mode: "client",
+  connect: {{ endpoints: ["tcp/127.0.0.1:{port}"] }}
+}}"#
+        ),
+    )
+    .with_context(|| format!("Failed to write zenoh config to {}", path.display()))?;
+
+    Ok(path)
+}
+
+pub async fn wait_for_router() {
+    tokio::time::sleep(Duration::from_millis(500)).await;
+}

tests/zenoh/raw.rs

@@ -0,0 +1,65 @@
+use std::{sync::Arc, time::Duration};
+
+use anyhow::{Context, Result};
+use clap::Parser;
+use mavlink_server::{cli, drivers::zenoh::raw::ZenohRaw, hub, stats};
+use tracing::*;
+
+use crate::common::{pick_free_port, wait_for_stats_collection};
+use crate::zenoh::{spawn_zenoh_router, wait_for_router, write_zenoh_client_config};
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 10)]
+async fn test_zenohraw_bidirectional() -> Result<()> {
+    let driver_name = "ZenohRaw";
+    let port = pick_free_port();
+    let _router = spawn_zenoh_router(port, "mavlink_raw").await;
+    wait_for_router().await;
+
+    let zenoh_config = write_zenoh_client_config(port)?;
+    let zenoh_config = zenoh_config
+        .to_str()
+        .context("Zenoh config path is not valid UTF-8")?;
+
+    cli::init_with(cli::Args::parse_from(vec![
+        &std::env::args().next().unwrap_or_default(),
+        "--allow-no-endpoints",
+        "--zenoh-config-file",
+        zenoh_config,
+        "--mavlink-heartbeat-frequency",
+        "10",
+    ]));
+
+    hub::add_driver(Arc::new(ZenohRaw::builder(driver_name).build())).await?;
+
+    stats::set_period(Duration::from_millis(100)).await?;
+    wait_for_stats_collection().await;
+
+    let mut matching_drivers = 0;
+    for (_uuid, driver) in stats::drivers_stats().await? {
+        if *driver.name != driver_name {
+            continue;
+        }
+
+        matching_drivers += 1;
+        assert!(
+            driver.stats.input.is_some(),
+            "{driver_name} should receive mavlink over zenoh"
+        );
+        assert!(
+            driver.stats.output.is_some(),
+            "{driver_name} should publish mavlink over zenoh"
+        );
+    }
+
+    assert_eq!(
+        matching_drivers, 1,
+        "expected one {driver_name} driver in stats"
+    );
+
+    for (id, driver_info) in hub::drivers().await? {
+        debug!("Removing driver id {id:?} ({driver_info:?})");
+        hub::remove_driver(id).await?;
+    }
+
+    Ok(())
+}