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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
"""Test the GRUB cmdline merge.
Bug (P2): configure_grub rewrote the whole GRUB_CMDLINE_LINUX_DEFAULT line
with a fixed string. A base install that had set cryptdevice=/resume=/zfs=
(or any other boot-critical token) lost it, and the following grub-mkconfig
baked an unbootable config.
The fix is a merge: every pre-existing token survives, archsetup's tokens are
added, and where both set the same key archsetup's value wins. A safety
assert refuses to write if any existing token's key would vanish.
Method: sed-extract merge_grub_cmdline (pure) and update_grub_cmdline
(file-level, run against temp grub files with a fake error_warn).
Run from repo root:
python3 -m unittest tests.installer-steps.test_grub_cmdline
"""
import os
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")
EXTRACT = (
"source <(sed -n '/^merge_grub_cmdline() {{/,/^}}/p;"
"/^update_grub_cmdline() {{/,/^}}/p' \"{a}\")"
).format(a=ARCHSETUP)
def run(body):
script = f"logfile=/dev/null\nerror_warn() {{ echo \"WARN: $1\"; return 1; }}\n{EXTRACT}\n{body}\n"
return subprocess.run(["bash", "-c", script],
capture_output=True, text=True, timeout=10)
def merge(existing, desired):
r = run(f'merge_grub_cmdline "{existing}" "{desired}"')
return r.stdout.strip()
class MergeGrubCmdline(unittest.TestCase):
DESIRED = "rw loglevel=2 quiet splash"
def test_empty_existing_yields_desired(self):
self.assertEqual(merge("", self.DESIRED), self.DESIRED)
def test_boot_critical_tokens_survive(self):
out = merge("cryptdevice=UUID=abc:root resume=/dev/nvme0n1p3 root=/dev/mapper/root",
self.DESIRED)
for tok in ("cryptdevice=UUID=abc:root", "resume=/dev/nvme0n1p3",
"root=/dev/mapper/root", "loglevel=2", "quiet", "splash"):
self.assertIn(tok, out.split(), f"{tok} missing from: {out}")
def test_same_key_desired_value_wins_once(self):
out = merge("loglevel=7 quiet", self.DESIRED).split()
self.assertIn("loglevel=2", out)
self.assertNotIn("loglevel=7", out)
self.assertEqual(out.count("loglevel=2"), 1)
self.assertEqual(out.count("quiet"), 1)
def test_zfs_token_survives(self):
out = merge("zfs=zroot/ROOT/default", self.DESIRED).split()
self.assertIn("zfs=zroot/ROOT/default", out)
def run_with_gpu(body, gpu):
# detect_gpu_vendors is stubbed to a chosen vendor list so the AMD-only
# amdgpu.runpm=0 addition can be exercised without real /sys reads.
stub = 'detect_gpu_vendors() { printf "%s\\n" %s; }\n' % ("%s", "'" + gpu + "'")
stub = 'detect_gpu_vendors() { printf "%s" "$GPU"; [ -n "$GPU" ] && echo; }\n'
script = ('logfile=/dev/null\nerror_warn() { echo "WARN: $1"; return 1; }\n'
+ stub + EXTRACT + "\n" + body + "\n")
return subprocess.run(["bash", "-c", script], capture_output=True, text=True,
timeout=10, env={**os.environ, "GPU": gpu})
class AmdRunpmCmdline(unittest.TestCase):
"""amdgpu.runpm=0 is added on AMD machines only: the AMD iGPU invalidates
hyprlock's GPU resources on a display power cycle, wedging the lock
(hyprlock#953). Disabling GPU runtime PM keeps them valid. A no-op on Intel
or NVIDIA."""
def _file(self, gpu):
import tempfile
fd, path = tempfile.mkstemp(prefix="grub-runpm-")
os.close(fd)
with open(path, "w") as f:
f.write('GRUB_CMDLINE_LINUX_DEFAULT="rw quiet"\n')
self.addCleanup(os.remove, path)
run_with_gpu(f'update_grub_cmdline "{path}"', gpu)
with open(path) as f:
return f.read()
def test_amd_gets_runpm_disabled(self):
self.assertIn("amdgpu.runpm=0", self._file("amd"))
def test_intel_does_not(self):
self.assertNotIn("amdgpu.runpm", self._file("intel"))
def test_no_gpu_detected_does_not(self):
self.assertNotIn("amdgpu.runpm", self._file(""))
def test_amd_with_nvidia_still_gets_it(self):
# A hybrid box with an AMD part present still wants the fix.
self.assertIn("amdgpu.runpm=0", self._file("amd\nnvidia"))
class UpdateGrubCmdline(unittest.TestCase):
def run_update(self, grub_body):
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "grub")
with open(path, "w") as f:
f.write(grub_body)
r = run(f'update_grub_cmdline "{path}"; echo "RC=$?"')
with open(path) as f:
return r, f.read()
def line(self, content):
return [ln for ln in content.splitlines()
if ln.startswith('GRUB_CMDLINE_LINUX_DEFAULT=')]
def test_existing_tokens_preserved_in_file(self):
_, content = self.run_update(textwrap.dedent("""\
GRUB_TIMEOUT=5
GRUB_CMDLINE_LINUX_DEFAULT="loglevel=3 cryptdevice=UUID=abc:root resume=/dev/sda2"
GRUB_DISABLE_RECOVERY=true
"""))
lines = self.line(content)
self.assertEqual(len(lines), 1)
val = lines[0]
self.assertIn("cryptdevice=UUID=abc:root", val)
self.assertIn("resume=/dev/sda2", val)
self.assertIn("loglevel=2", val) # archsetup's value wins
self.assertNotIn("loglevel=3", val)
self.assertIn("quiet", val)
# other lines untouched
self.assertIn("GRUB_TIMEOUT=5", content)
self.assertIn("GRUB_DISABLE_RECOVERY=true", content)
def test_quoting_stays_a_single_pair(self):
_, content = self.run_update('GRUB_CMDLINE_LINUX_DEFAULT="quiet"\n')
val = self.line(content)[0]
self.assertEqual(val.count('"'), 2)
self.assertRegex(val, r'^GRUB_CMDLINE_LINUX_DEFAULT="[^"]*"$')
def test_commented_line_gets_active_line_appended(self):
_, content = self.run_update('#GRUB_CMDLINE_LINUX_DEFAULT="quiet"\n')
self.assertEqual(len(self.line(content)), 1)
self.assertIn('#GRUB_CMDLINE_LINUX_DEFAULT="quiet"', content)
def test_idempotent_on_rerun(self):
body = 'GRUB_CMDLINE_LINUX_DEFAULT="cryptdevice=UUID=abc:root"\n'
_, once = self.run_update(body)
_, twice = self.run_update(once)
self.assertEqual(once, twice)
if __name__ == "__main__":
unittest.main()
|
