__main__: honour --device= in docker_shell_extra_arguments

test/unit/test_main.py

@@ -1,12 +1,15 @@
 import json
+import logging
 import os
+from pathlib import Path
 
 import pytest
 import shlex
 import yaml
 
 import tuxrun.__main__
-from tuxrun.__main__ import main, start
+from tuxrun.__main__ import bind_docker_shell_extra_args, main, start
+from tuxrun.runtimes import Runtime
 
 
 def touch(directory, name):
@@ -754,3 +757,52 @@ def test_boot_args(monkeypatch, mocker, tmpdir, args, cnt):
         # Invalid case
         with pytest.raises(Exception):
             main()
+
+
+class TestBindDockerShellExtraArgs:
+    def test_device_bound_read_write_by_default(self, tmp_path):
+        dev = touch(tmp_path, "i2c-3")
+        runtime = Runtime.select("null")(tmp_path)
+        bind_docker_shell_extra_args(runtime, [f"--device={dev}"])
+        assert runtime.__bindings__ == [(str(dev), dev, False, True)]
+
+    def test_device_honours_read_only_permission(self, tmp_path):
+        dev = touch(tmp_path, "i2c-3")
+        runtime = Runtime.select("null")(tmp_path)
+        bind_docker_shell_extra_args(runtime, [f"--device={dev}:{dev}:r"])
+        assert runtime.__bindings__ == [(str(dev), dev, True, True)]
+
+    def test_device_read_write_permission_kept(self, tmp_path):
+        dev = touch(tmp_path, "i2c-3")
+        runtime = Runtime.select("null")(tmp_path)
+        bind_docker_shell_extra_args(runtime, [f"--device={dev}:{dev}:rw"])
+        assert runtime.__bindings__ == [(str(dev), dev, False, True)]
+
+    def test_device_custom_container_path(self, tmp_path):
+        dev = touch(tmp_path, "i2c-3")
+        runtime = Runtime.select("null")(tmp_path)
+        bind_docker_shell_extra_args(runtime, [f"--device={dev}:/dev/i2c-9"])
+        assert runtime.__bindings__ == [(str(dev), Path("/dev/i2c-9"), False, True)]
+
+    def test_device_empty_host_path_skipped_with_warning(self, tmp_path, caplog):
+        runtime = Runtime.select("null")(tmp_path)
+        with caplog.at_level(logging.WARNING, logger="tuxrun"):
+            bind_docker_shell_extra_args(runtime, ["--device="])
+        assert runtime.__bindings__ == []
+        assert "empty host path" in caplog.text
+
+    def test_device_missing_host_path_skipped_with_warning(self, tmp_path, caplog):
+        missing = tmp_path / "does-not-exist"
+        runtime = Runtime.select("null")(tmp_path)
+        with caplog.at_level(logging.WARNING, logger="tuxrun"):
+            bind_docker_shell_extra_args(runtime, [f"--device={missing}"])
+        assert runtime.__bindings__ == []
+        assert "does not exist" in caplog.text
+
+    def test_volume_read_only_still_works(self, tmp_path):
+        cfg = touch(tmp_path, "config.csv")
+        runtime = Runtime.select("null")(tmp_path)
+        bind_docker_shell_extra_args(runtime, [f"--volume={cfg}:/etc/config.csv:ro"])
+        assert runtime.__bindings__ == [
+            (str(cfg), Path("/etc/config.csv"), True, False)
+        ]

test/unit/test_runtime.py

@@ -347,3 +347,35 @@ def test_host_network_and_skip_http_server_combined(tmp_path):
     assert "--network" in cmd
     assert "--entrypoint" in cmd
     assert cmd[-2:] == ["--device", "foo"]
+
+
+def test_bind_device_read_write_renders_rw(tmp_path):
+    runtime = Runtime.select("podman")(tmp_path)
+    runtime.name("name")
+    runtime.image("image")
+    runtime.bind("/dev/i2c-3", "/dev/i2c-3", device=True)
+    cmd = runtime.cmd(["hello"])
+    idx = cmd.index("/dev/i2c-3:/dev/i2c-3:rw")
+    assert cmd[idx - 1] == "--device"
+
+
+def test_bind_device_read_only_renders_valid_r(tmp_path):
+    runtime = Runtime.select("podman")(tmp_path)
+    runtime.name("name")
+    runtime.image("image")
+    runtime.bind("/dev/i2c-3", "/dev/i2c-3", ro=True, device=True)
+    cmd = runtime.cmd(["hello"])
+    assert "/dev/i2c-3:/dev/i2c-3:r" in cmd
+    assert "/dev/i2c-3:/dev/i2c-3:ro" not in cmd
+    idx = cmd.index("/dev/i2c-3:/dev/i2c-3:r")
+    assert cmd[idx - 1] == "--device"
+
+
+def test_bind_volume_read_only_still_renders_ro(tmp_path):
+    runtime = Runtime.select("podman")(tmp_path)
+    runtime.name("name")
+    runtime.image("image")
+    runtime.bind("/etc/config", "/etc/config", ro=True)
+    cmd = runtime.cmd(["hello"])
+    idx = cmd.index("/etc/config:/etc/config:ro")
+    assert cmd[idx - 1] == "-v"

tuxrun/__main__.py

@@ -172,6 +172,39 @@ def run_hacking_sesson(definition, case, test):
             )
 
 
+def bind_docker_shell_extra_args(runtime, extra_args):
+    """Bind the volumes and devices declared in a device dict's
+    docker_shell_extra_arguments onto the runtime."""
+    for arg in extra_args:
+        if arg.startswith("--volume="):
+            volume_spec = arg[len("--volume=") :]
+            parts = volume_spec.split(":")
+            if len(parts) >= 2:
+                host_path = Path(parts[0])
+                container_path = Path(parts[1])
+                ro = len(parts) > 2 and parts[2] == "ro"
+                if host_path.exists():
+                    runtime.bind(host_path, container_path, ro=ro)
+        elif arg.startswith("--device="):
+            device_spec = arg[len("--device=") :]
+            parts = device_spec.split(":")
+            if not parts[0]:
+                LOG.warning("Ignoring --device with empty host path: %r", arg)
+                continue
+            host_path = Path(parts[0])
+            container_path = (
+                Path(parts[1]) if len(parts) > 1 and parts[1] else host_path
+            )
+            # docker --device permissions are r/w/m; treat a bare "r" as
+            # read-only and anything else (rw, rwm, unset) as read-write, so we
+            # never grant more access than the device dict asked for.
+            ro = len(parts) > 2 and parts[2] == "r"
+            if host_path.exists():
+                runtime.bind(host_path, container_path, ro=ro, device=True)
+            else:
+                LOG.warning("Ignoring --device %s: host path does not exist", host_path)
+
+
 ##############
 # Entrypoint #
 ##############
@@ -323,16 +356,7 @@ def run(options, tmpdir: Path, cache_dir: Optional[Path], artefacts: dict) -> in
         runtime.skip_http_server()
         if job.d_dict_config:
             extra_args = job.d_dict_config.get("docker_shell_extra_arguments", [])
-            for arg in extra_args:
-                if arg.startswith("--volume="):
-                    volume_spec = arg[len("--volume=") :]
-                    parts = volume_spec.split(":")
-                    if len(parts) >= 2:
-                        host_path = Path(parts[0])
-                        container_path = Path(parts[1])
-                        ro = len(parts) > 2 and parts[2] == "ro"
-                        if host_path.exists():
-                            runtime.bind(host_path, container_path, ro=ro)
+            bind_docker_shell_extra_args(runtime, extra_args)
         control_binaries_param = options.parameters.get("DEVICE_CONTROL_BINARIES", "")
         control_binaries = (
             control_binaries_param.split(",") if control_binaries_param else []

tuxrun/runtimes.py

@@ -164,8 +164,12 @@ def cmd(self, args):
                 LOG.error("Duplicated mount destination %r", dst)
                 raise Exception("Duplicated mount destination %r" % dst)
             dsts.add(dst)
-            ro = "ro" if ro else "rw"
-            prefix.extend(["--device" if device else "-v", f"{src}:{dst}:{ro}"])
+            if device:
+                mode = "r" if ro else "rw"
+                prefix.extend(["--device", f"{src}:{dst}:{mode}"])
+            else:
+                mode = "ro" if ro else "rw"
+                prefix.extend(["-v", f"{src}:{dst}:{mode}"])
         prefix.extend(["--name", self.__name__])
         return prefix + [self.__image__] + args