Allow for per-test dbus sessions

src/daemon.rs

@@ -92,7 +92,8 @@ pub async fn run(cfg: Config, backends: Vec<Box<dyn Backend>>) -> Result<(), Dae
     tracing::info!(target: "controller", added, "initial discovery");
 
     let (dbus_tx, mut dbus_rx) = tokio::sync::mpsc::channel(1);
-    let _dbus_connection = dbus::serve(dbus::Service::new(dbus_tx)).await?;
+    let service = dbus::Service::new(dbus_tx);
+    let _dbus_connection = dbus::serve(service).await?;
     let mut inotify_events = setup_inotify(&backends)?;
     let mut poll = interval(Duration::from_secs_f64(cfg.scale_poll_secs));
     poll.set_missed_tick_behavior(MissedTickBehavior::Skip);

src/dbus.rs

@@ -327,8 +327,22 @@ impl Service {
 
 /// Spin the D-Bus service on the well-known name / object path
 /// backed by `service` and drive replies from the controller mpsc.
+///
+/// Component tests set `IO_THREAD_CONTROLLER_DBUS_ADDRESS` to use their
+/// private bus; production leaves it unset and uses the system bus.
 pub async fn serve(service: Service) -> Result<zbus::Connection, zbus::Error> {
-    let conn = zbus::connection::Builder::system()?
+    let builder = match std::env::var("IO_THREAD_CONTROLLER_DBUS_ADDRESS") {
+        Ok(address) => zbus::connection::Builder::address(address.as_str())?,
+        Err(_) => zbus::connection::Builder::system()?,
+    };
+    serve_with(builder, service).await
+}
+
+async fn serve_with(
+    builder: zbus::connection::Builder<'_>,
+    service: Service,
+) -> Result<zbus::Connection, zbus::Error> {
+    let conn = builder
         .name(DBUS_BUS_NAME)?
         .serve_at(DBUS_OBJECT_PATH, service)?
         .build()

tests/component/harness/dbus.py

@@ -0,0 +1,68 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Nutanix, Inc. All rights reserved.
+#
+# Author: Leonardo Forchini <leonardo.forchini@nutanix.com>
+
+import signal
+import subprocess
+
+from harness.common import wait_for
+
+
+class DBusServer:
+    """Test only DBus instance."""
+    def __init__(self, root):
+        """
+        Create a new DBus environment, enabling each test to have its own
+        bus and ensuring isolation.
+        """
+        self._socket = root / "dbus.sock"
+        self._config = root / "dbus.xml"
+        self._config.write_text(
+            """<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-BUS Bus Configuration 1.0//EN"
+ "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
+<busconfig>
+  <type>system</type>
+  <listen>unix:path=%s</listen>
+  <auth>EXTERNAL</auth>
+  <policy context="default">
+    <allow own="*"/>
+    <allow send_destination="*"/>
+    <allow receive_sender="*"/>
+  </policy>
+</busconfig>
+"""
+            % self._socket
+        )
+        self._proc = None
+        self.address = None
+
+    def start(self):
+        """Start the bus and wait for it to come up."""
+        self._proc = subprocess.Popen(
+            [
+                "dbus-daemon",
+                "--nofork",
+                "--config-file=%s" % self._config,
+                "--print-address",
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        assert self._proc.stdout is not None
+        self.address = self._proc.stdout.readline().strip()
+        if not self.address:
+            raise RuntimeError("dbus-daemon did not print an address")
+        wait_for(self._socket.exists, 5, "DBus to come up")
+
+    def stop(self):
+        """Stop the bus."""
+        if self._proc is None or self._proc.poll() is not None:
+            return
+        self._proc.send_signal(signal.SIGTERM)
+        try:
+            self._proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            self._proc.kill()
+            self._proc.wait(timeout=5)