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
|
"""Test the resume-idempotency cluster: three extracted installer helpers.
Each fix targets a step that misbehaves on a resume re-run or with unexpected
system state:
- crontab_append_once: the log-cleanup cron line was appended unconditionally,
so a resume re-run of essential_services stacked a duplicate line each time.
The helper appends only when no existing line matches.
- zfs_scrub_timer_units: the scrub timer used `zpool list | head -1`, which
picks an arbitrary pool when several exist and yields the malformed unit
`zfs-scrub-weekly@.timer` when none do. The helper emits one unit per pool
and nothing for no pools.
- enable_user_service: gamemode was enabled via `systemctl --user enable`,
which the script itself documents fails at install time (no user bus). The
helper creates the wants symlink directly, the pattern syncthing already uses.
Method: sed-extract each helper from the real `archsetup` and drive it with
plain stdin/args (the first two are pure) or a temp HOME (the third).
Run from repo root:
python3 -m unittest tests.installer-steps.test_idempotency_cluster
"""
import os
import subprocess
import tempfile
import unittest
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
ARCHSETUP = os.path.join(REPO_ROOT, "archsetup")
ME = os.environ.get("USER") or __import__("getpass").getuser()
def extract(func):
return (
"source <(sed -n '/^{f}() {{/,/^}}/p' \"{a}\")".format(f=func, a=ARCHSETUP)
)
def run(func, body, stdin=None):
script = f"{extract(func)}\n{body}\n"
return subprocess.run(
["bash", "-c", script], capture_output=True, text=True, timeout=10,
input=stdin,
)
class CrontabAppendOnce(unittest.TestCase):
def call(self, existing, match, line):
body = f'crontab_append_once {match!r} {line!r}'
return run("crontab_append_once", body, stdin=existing).stdout
def test_appends_when_absent(self):
out = self.call("", "log-cleanup", "0 12 * * * run-cleanup")
self.assertIn("0 12 * * * run-cleanup", out)
def test_preserves_existing_and_appends(self):
out = self.call("0 0 * * * other-job\n", "log-cleanup", "0 12 * * * cleanup")
self.assertIn("other-job", out)
self.assertIn("0 12 * * * cleanup", out)
def test_does_not_duplicate_when_present(self):
existing = "0 12 * * * $HOME/.local/bin/cron/log-cleanup\n"
out = self.call(existing, "cron/log-cleanup", "0 12 * * * dup")
self.assertNotIn("0 12 * * * dup", out)
self.assertEqual(out.count("log-cleanup"), 1)
class ZfsScrubTimerUnits(unittest.TestCase):
def units(self, pools_text):
out = run("zfs_scrub_timer_units", "zfs_scrub_timer_units", stdin=pools_text).stdout
return [ln for ln in out.splitlines() if ln]
def test_single_pool(self):
self.assertEqual(self.units("zroot\n"),
["zfs-scrub-weekly@zroot.timer"])
def test_no_pool_yields_no_unit(self):
self.assertEqual(self.units(""), [])
def test_multiple_pools_each_get_a_unit(self):
self.assertEqual(
self.units("zroot\ntank\n"),
["zfs-scrub-weekly@zroot.timer", "zfs-scrub-weekly@tank.timer"],
)
def test_blank_lines_are_skipped(self):
self.assertEqual(self.units("zroot\n\n"),
["zfs-scrub-weekly@zroot.timer"])
class EnableUserService(unittest.TestCase):
def test_creates_wants_symlink_in_user_config(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'link="{home}/.config/systemd/user/default.target.wants/gamemoded.service"\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/gamemoded.service", out)
def test_idempotent_on_rerun(self):
with tempfile.TemporaryDirectory() as home:
svc = "/usr/lib/systemd/user/gamemoded.service"
one = (
f'enable_user_service {ME!r} gamemoded.service {svc!r} {home!r}\n'
f'enable_user_service {ME!r} gamemoded.service {svc!r} {home!r}\n'
f'echo "RC=$?"'
)
out = run("enable_user_service", one).stdout
self.assertIn("RC=0", out)
if __name__ == "__main__":
unittest.main()
|
