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
|
"""Test run_step's state-marker and scope behavior.
Bug (Major): run_step marked a step complete only when its function returned 0.
But error_fatal exits the script outright, so a step function that *returns* has
already survived every fatal check -- a non-zero return can only come from a
trailing non-fatal warning (error_warn returns 1). The old code read that as a
step failure and withheld the state marker, so the step re-ran on every resume.
run_step now records the marker whenever the function returns.
Bug (Minor): run_step's step_name/step_func leaked to global scope.
Method: sed-extract run_step + step_completed + mark_complete from the real
`archsetup`, point state_dir at a temp dir, and drive it with fake step
functions.
Run from repo root:
python3 -m unittest tests.installer-steps.test_run_step
"""
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 '/^run_step() {{/,/^}}/p;"
"/^step_completed() {{/,/^}}/p;"
"/^mark_complete() {{/,/^}}/p' \"{archsetup}\")"
).format(archsetup=ARCHSETUP)
def run(step_body, pre_marked=False):
"""Run run_step against a fake step function whose body is step_body."""
with tempfile.TemporaryDirectory() as state_dir:
marker_setup = (
f'mkdir -p "{state_dir}"; echo pre > "{state_dir}/mystep"\n'
if pre_marked else ""
)
script = textwrap.dedent(f"""\
state_dir="{state_dir}"
{marker_setup}
mystep() {{
{step_body}
}}
{EXTRACT}
run_step "mystep" "mystep"
rc=$?
echo "RUN_STEP_RC=$rc"
# marker present?
[ -f "$state_dir/mystep" ] && echo "MARKED=yes" || echo "MARKED=no"
# locals must not leak
echo "LEAK_step_name=[${{step_name:-}}]"
echo "LEAK_step_func=[${{step_func:-}}]"
""")
return subprocess.run(
["bash", "-c", script], capture_output=True, text=True, timeout=10,
)
class RunStep(unittest.TestCase):
def test_success_marks_complete(self):
r = run("return 0")
self.assertIn("RUN_STEP_RC=0", r.stdout)
self.assertIn("MARKED=yes", r.stdout)
def test_nonfatal_warning_return_still_marks_complete(self):
# The bug fix: a step ending in a non-fatal warning returns 1 but has
# done its work; it must be marked so resume does not re-run it.
r = run("return 1")
self.assertIn("MARKED=yes", r.stdout,
"a returned-non-zero step must still record its marker")
self.assertIn("RUN_STEP_RC=0", r.stdout,
"run_step returns 0 so the main loop treats it as handled")
def test_already_completed_skips_and_does_not_rerun(self):
# Body would create a sentinel if it ran; pre-marked step must skip it.
r = run("echo RAN-BODY", pre_marked=True)
self.assertIn("Skipping", r.stdout)
self.assertNotIn("RAN-BODY", r.stdout)
def test_locals_do_not_leak_to_global_scope(self):
r = run("return 0")
self.assertIn("LEAK_step_name=[]", r.stdout)
self.assertIn("LEAK_step_func=[]", r.stdout)
if __name__ == "__main__":
unittest.main()
|
