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
164
165
166
167
168
169
170
171
172
|
"""Tests for the run_task / enable_service helpers in the archsetup installer.
run_task is the installer's describe-run-warn primitive. It replaces the
hand-written idiom that recurs ~100 times across the script:
action="enabling rngd service" && display "task" "$action"
systemctl enable rngd >> "$logfile" 2>&1 || error_warn "$action" "$?"
as a single call:
run_task "enabling rngd service" systemctl enable rngd
It announces the task via display, runs the command with stdout+stderr
appended to $logfile, and on failure calls error_warn with the command's
real exit code (non-fatal). enable_service is a thin wrapper that enables
one or more systemd units with the conventional "enabling <unit> service"
wording.
These tests exercise the REAL function bodies, extracted from the
`archsetup` script at run time (not a copy), with recording stubs standing
in for display, error_warn, and systemctl. The command run by run_task is
genuinely executed.
Run from repo root:
python3 -m unittest tests.run-task.test_run_task
"""
import os
import shutil
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")
# A bash harness that sources the real run_task + enable_service out of the
# installer, with recording stubs for their dependencies. Each stub appends a
# tab-separated record to a file named by an env var, so the Python side can
# assert what was called. The real command passed to run_task still runs.
WRAPPER = r"""#!/bin/bash
ARCHSETUP="$1"; shift
logfile="$LOGFILE"
display() { printf '%s\t%s\n' "$1" "$2" >> "$DISPLAY_LOG"; }
error_warn() { printf '%s\t%s\n' "$1" "$2" >> "$ERRWARN_LOG"; return 1; }
systemctl() { printf 'systemctl %s\n' "$*"; }
source <(sed -n '/^run_task() {/,/^}/p' "$ARCHSETUP")
source <(sed -n '/^enable_service() {/,/^}/p' "$ARCHSETUP")
"$@"
"""
class RunTaskHarness(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="run-task-test-")
self.wrapper = os.path.join(self.tmp, "run.sh")
with open(self.wrapper, "w") as f:
f.write(WRAPPER)
os.chmod(self.wrapper, 0o755)
self.logfile = os.path.join(self.tmp, "install.log")
self.display_log = os.path.join(self.tmp, "display.log")
self.errwarn_log = os.path.join(self.tmp, "errwarn.log")
def tearDown(self):
shutil.rmtree(self.tmp, ignore_errors=True)
def call(self, *args):
env = dict(os.environ)
env["LOGFILE"] = self.logfile
env["DISPLAY_LOG"] = self.display_log
env["ERRWARN_LOG"] = self.errwarn_log
return subprocess.run(
["bash", self.wrapper, ARCHSETUP, *args],
capture_output=True, text=True, timeout=10, env=env,
)
def read(self, path):
if not os.path.exists(path):
return ""
with open(path) as f:
return f.read()
# --- Normal cases -----------------------------------------------------
def test_run_task_success_announces_and_runs(self):
result = self.call("run_task", "doing a thing", "true")
self.assertEqual(result.returncode, 0, result.stderr)
# Announced as a "task" with the exact description.
self.assertEqual(self.read(self.display_log), "task\tdoing a thing\n")
# No warning on success.
self.assertEqual(self.read(self.errwarn_log), "")
def test_run_task_captures_command_output_to_logfile(self):
result = self.call("run_task", "echo something", "echo", "hello-from-cmd")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("hello-from-cmd", self.read(self.logfile))
# Command output is logged, not printed to the terminal.
self.assertNotIn("hello-from-cmd", result.stdout)
def test_run_task_captures_stderr_to_logfile(self):
# `ls` of a missing path writes to stderr; it must land in the logfile.
missing = os.path.join(self.tmp, "no-such-path")
self.call("run_task", "listing", "ls", missing)
self.assertIn("no-such-path", self.read(self.logfile))
def test_run_task_preserves_multiple_arguments(self):
self.call("run_task", "multi-arg", "printf", "%s|%s|%s", "a", "b", "c")
self.assertIn("a|b|c", self.read(self.logfile))
def test_run_task_preserves_arguments_with_spaces(self):
self.call("run_task", "spacey", "printf", "[%s]", "two words")
self.assertIn("[two words]", self.read(self.logfile))
# --- enable_service ---------------------------------------------------
def test_enable_service_single_unit(self):
self.call("enable_service", "rngd")
self.assertEqual(self.read(self.display_log), "task\tenabling rngd service\n")
self.assertIn("systemctl enable rngd", self.read(self.logfile))
def test_enable_service_multiple_units(self):
self.call("enable_service", "foo", "bar", "baz")
disp = self.read(self.display_log)
self.assertIn("task\tenabling foo service\n", disp)
self.assertIn("task\tenabling bar service\n", disp)
self.assertIn("task\tenabling baz service\n", disp)
log = self.read(self.logfile)
self.assertIn("systemctl enable foo", log)
self.assertIn("systemctl enable bar", log)
self.assertIn("systemctl enable baz", log)
# --- Error cases ------------------------------------------------------
def test_run_task_failure_warns_with_description(self):
result = self.call("run_task", "failing thing", "false")
self.assertNotEqual(result.returncode, 0)
self.assertEqual(self.read(self.errwarn_log), "failing thing\t1\n")
def test_run_task_failure_propagates_real_exit_code(self):
# `bash -c 'exit 42'` must surface 42 to error_warn, not a clobbered 0.
self.call("run_task", "exit-42", "bash", "-c", "exit 42")
self.assertEqual(self.read(self.errwarn_log), "exit-42\t42\n")
def test_enable_service_failure_warns_per_unit(self):
# Override systemctl to fail; each unit should produce a warning.
env = dict(os.environ)
env["LOGFILE"] = self.logfile
env["DISPLAY_LOG"] = self.display_log
env["ERRWARN_LOG"] = self.errwarn_log
# Re-create wrapper with a failing systemctl stub for this case.
failing = os.path.join(self.tmp, "run-fail.sh")
with open(failing, "w") as f:
f.write(WRAPPER.replace(
"systemctl() { printf 'systemctl %s\\n' \"$*\"; }",
"systemctl() { printf 'systemctl %s\\n' \"$*\"; return 1; }",
))
os.chmod(failing, 0o755)
subprocess.run(
["bash", failing, ARCHSETUP, "enable_service", "alpha", "beta"],
capture_output=True, text=True, timeout=10, env=env,
)
warns = self.read(self.errwarn_log)
self.assertIn("enabling alpha service\t1\n", warns)
self.assertIn("enabling beta service\t1\n", warns)
if __name__ == "__main__":
unittest.main()
|
