1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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()
|
