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
|
"""Test run-net-scenarios.sh scenario accounting.
Bug: the scenario_diagnose_expect else-branch printed a FAIL line but never
forced a non-zero subshell exit, so a scenario whose only failure was the
diagnose check counted as passed and the run printed "all scenarios passed"
with exit 0. A net-doctor diagnosis regression would ship behind a green run.
Method: run the real script against a temp scenario dir (NET_SCENARIO_DIR)
with stub ssh/rsync/jq on PATH, driving one fake scenario whose check results
are controlled per test. Assert on the script's exit code and summary line.
Run from repo root:
python3 -m unittest tests.net-scenarios.test_run_net_scenarios
"""
import os
import shutil
import subprocess
import tempfile
import textwrap
import unittest
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
SCRIPT = os.path.join(REPO_ROOT, "scripts", "testing", "run-net-scenarios.sh")
STUB = "#!/bin/sh\ncat >/dev/null 2>&1 || true\nexit 0\n"
class RunNetScenarios(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="net-scen-test-")
self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
self.bindir = os.path.join(self.tmp, "bin")
os.mkdir(self.bindir)
for tool in ("ssh", "rsync", "jq"):
p = os.path.join(self.bindir, tool)
with open(p, "w") as f:
f.write(STUB)
os.chmod(p, 0o755)
self.scen_dir = os.path.join(self.tmp, "scenarios")
os.mkdir(self.scen_dir)
def write_scenario(self, *, brk=0, diagnose=0, fix=0, assert_rc=0,
with_diagnose=True):
diag_fn = (
f"scenario_diagnose_expect() {{ return {diagnose}; }}\n"
if with_diagnose else ""
)
body = textwrap.dedent(f"""\
SCENARIO_DESC="fake scenario for harness tests"
scenario_break() {{ return {brk}; }}
{diag_fn}scenario_fix() {{ return {fix}; }}
scenario_assert() {{ return {assert_rc}; }}
""")
with open(os.path.join(self.scen_dir, "fake-scenario.sh"), "w") as f:
f.write(body)
def run_script(self):
env = dict(os.environ)
env["PATH"] = self.bindir + os.pathsep + env["PATH"]
env["NET_SCENARIO_DIR"] = self.scen_dir
env["DOTFILES"] = self.tmp # rsync is stubbed; path just has to exist
return subprocess.run(
["bash", SCRIPT, "--target", "root@fake-vm"],
capture_output=True, text=True, timeout=20, env=env,
)
def test_all_checks_pass_exits_zero(self):
self.write_scenario()
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
self.assertIn("all scenarios passed", r.stdout)
def test_diagnose_failure_alone_fails_the_run(self):
# The bug: break/fix/assert all pass, only the diagnose check fails.
self.write_scenario(diagnose=1)
r = self.run_script()
self.assertIn("diagnose did NOT name it", r.stdout)
self.assertNotEqual(r.returncode, 0,
"a diagnose regression must not exit green")
self.assertIn("1 scenario(s) failed", r.stdout)
def test_assert_failure_fails_the_run(self):
self.write_scenario(assert_rc=1)
r = self.run_script()
self.assertNotEqual(r.returncode, 0)
self.assertIn("NOT repaired", r.stdout)
def test_break_failure_fails_the_run(self):
self.write_scenario(brk=1)
r = self.run_script()
self.assertNotEqual(r.returncode, 0)
self.assertIn("break failed", r.stdout)
def test_scenario_without_diagnose_hook_still_passes(self):
self.write_scenario(with_diagnose=False)
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
if __name__ == "__main__":
unittest.main()
|
