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
|
"""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)
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()
|
