From 169dab71a063f4fab6c46f95e11d6e85b140cb8f Mon Sep 17 00:00:00 2001 From: Craig Jennings Date: Sun, 9 Aug 2026 11:20:43 -0500 Subject: feat: enable the podman API socket and ship the camera udev grant Two one-time machine setups from the winvm handoffs now happen at install time. - The devops podman block enables the rootless podman API socket. Socket-activated, so it costs nothing idle, and API clients like Pods fail with an empty window without it. enable_user_service grew an optional wants-target argument because a socket unit's [Install] is WantedBy=sockets.target. The old default.target link would never socket-activate. - install_camera_passthrough_rules ships 72-usb-passthrough-cameras.rules: GROUP="video", MODE="0660" plus the uaccess tag on the OBSBOT and BRIO USB IDs, so usbredirect can claim them for VM passthrough. The filename is load-bearing: logind's ACL is applied by 73-seat-late.rules, so the tag only works from a file sorting below 73. A test pins that property. Both are live on ratio (the old 99- rules file is retired there). Whether uaccess alone would suffice from the corrected position is untested and stays documented as a hypothesis. --- tests/installer-steps/test_idempotency_cluster.py | 27 +++++ .../test_install_camera_passthrough_rules.py | 118 +++++++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 tests/installer-steps/test_install_camera_passthrough_rules.py (limited to 'tests/installer-steps') diff --git a/tests/installer-steps/test_idempotency_cluster.py b/tests/installer-steps/test_idempotency_cluster.py index ecb279d..0cca750 100644 --- a/tests/installer-steps/test_idempotency_cluster.py +++ b/tests/installer-steps/test_idempotency_cluster.py @@ -114,6 +114,33 @@ class EnableUserService(unittest.TestCase): out = run("enable_user_service", one).stdout self.assertIn("RC=0", out) + def test_socket_unit_lands_in_sockets_target_wants(self): + # A socket unit's [Install] is WantedBy=sockets.target, so enabling it + # via default.target.wants would never socket-activate. The optional + # fifth arg names the wants target. + with tempfile.TemporaryDirectory() as home: + body = ( + f'enable_user_service {ME!r} podman.socket ' + f'/usr/lib/systemd/user/podman.socket {home!r} sockets.target\n' + f'link="{home}/.config/systemd/user/sockets.target.wants/podman.socket"\n' + f'[ -L "$link" ] && echo "LINK=yes" || echo "LINK=no"\n' + f'echo "TARGET=$(readlink "$link")"' + ) + out = run("enable_user_service", body).stdout + self.assertIn("LINK=yes", out) + self.assertIn("TARGET=/usr/lib/systemd/user/podman.socket", out) + + def test_omitted_target_still_defaults_to_default_target(self): + with tempfile.TemporaryDirectory() as home: + body = ( + f'enable_user_service {ME!r} gamemoded.service ' + f'/usr/lib/systemd/user/gamemoded.service {home!r}\n' + f'[ -L "{home}/.config/systemd/user/default.target.wants/gamemoded.service" ] ' + f'&& echo "DEFAULT=yes" || echo "DEFAULT=no"' + ) + out = run("enable_user_service", body).stdout + self.assertIn("DEFAULT=yes", out) + if __name__ == "__main__": unittest.main() diff --git a/tests/installer-steps/test_install_camera_passthrough_rules.py b/tests/installer-steps/test_install_camera_passthrough_rules.py new file mode 100644 index 0000000..ab28575 --- /dev/null +++ b/tests/installer-steps/test_install_camera_passthrough_rules.py @@ -0,0 +1,118 @@ +"""Test install_camera_passthrough_rules — the VM camera udev grant. + +usbredirect must open a camera's raw USB node read-write to claim it for the +Windows VM; the node defaults to root-owned with no group write, so the +attach fails with a bare "Failed to open device!". The rule grants +GROUP="video", MODE="0660" (the verified fix) and keeps TAG+="uaccess". + +The file NUMBER is load-bearing (winvm correction, 2026-08-08): the uaccess +ACL is applied by 73-seat-late.rules, so a 99- file adds the tag after that +already ran. The shipped filename must sort below 73. + +Method: sed-extract install_camera_passthrough_rules from the real +`archsetup`, point it at a temp rules path, fake display / error_warn. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_install_camera_passthrough_rules +""" + +import os +import re +import stat +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") + + +def run(rules_path, pre=""): + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^install_camera_passthrough_rules() {{/,/^}}/p' "{ARCHSETUP}") + {pre} + install_camera_passthrough_rules "{rules_path}" + echo "RC=$?" + echo "RULES:[$(cat "{rules_path}" 2>/dev/null)]" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rules_body(r): + return r.stdout.split("RULES:[")[1].split("]")[0] + + +class InstallCameraPassthroughRules(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_writes_both_camera_rules_with_group_mode_and_tag(self): + with tempfile.TemporaryDirectory() as d: + r = run(os.path.join(d, "72-test.rules")) + body = rules_body(r) + for vendor, product in (("3564", "ff02"), ("046d", "085e")): + line = next((ln for ln in body.splitlines() + if f'ATTR{{idVendor}}=="{vendor}"' in ln), None) + assert line is not None, f"no rule line for {vendor}:{product}" + self.assertIn(f'ATTR{{idProduct}}=="{product}"', line) + self.assertIn('GROUP="video"', line) + self.assertIn('MODE="0660"', line) + self.assertIn('TAG+="uaccess"', line) + self.assertIn("RC=0", r.stdout) + + def test_default_filename_sorts_below_seat_late(self): + # The rule file's default install path must sort before + # 73-seat-late.rules or the uaccess tag lands too late to be ACLed. + with open(ARCHSETUP) as f: + src = f.read() + func = re.search( + r'^install_camera_passthrough_rules\(\)\s*{.*?^}', src, re.S | re.M) + assert func is not None, "function not found in archsetup" + m = re.search(r'\$\{1:-(/etc/udev/rules\.d/[^}]+)\}', func.group(0)) + assert m is not None, "default rules path not found in the function" + basename = os.path.basename(m.group(1)) + self.assertLess(basename, "73-seat-late.rules", + "the rules file must sort below 73-seat-late.rules") + + # ---------------------------------------------------------- boundary ---- + def test_rerun_is_idempotent(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "72-test.rules") + first = run(path) + body_one = rules_body(first) + second = run(path) + self.assertEqual(body_one, rules_body(second)) + self.assertIn("RC=0", second.stdout) + + def test_overwrites_a_stale_existing_file(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "72-test.rules") + with open(path, "w") as f: + f.write("# stale content that must not survive\n") + r = run(path) + self.assertNotIn("stale content", rules_body(r)) + self.assertIn('GROUP="video"', rules_body(r)) + + # ------------------------------------------------------------- error ---- + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_dir_warns_and_does_not_crash(self): + with tempfile.TemporaryDirectory() as d: + locked = os.path.join(d, "locked") + os.makedirs(locked) + os.chmod(locked, stat.S_IRUSR | stat.S_IXUSR) + r = run(os.path.join(locked, "72-test.rules")) + os.chmod(locked, stat.S_IRWXU) + self.assertIn("WARN:", r.stdout) + self.assertIn("RC=", r.stdout, + "the harness must reach its RC line — the function " + "returned rather than aborting the script") + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3