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
|
"""Tests for the run-test transport-independent polling/report core."""
import os
import subprocess
import tempfile
import textwrap
import unittest
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
RUN_COMMON = os.path.join(
REPO_ROOT, "scripts", "testing", "lib", "run-common.sh")
VM_RUNNER = os.path.join(REPO_ROOT, "scripts", "testing", "run-test.sh")
BARE_RUNNER = os.path.join(
REPO_ROOT, "scripts", "testing", "run-test-baremetal.sh")
def run_bash(body):
return subprocess.run(
["bash", "-c", f'source "{RUN_COMMON}"\n{body}'],
capture_output=True, text=True, cwd=REPO_ROOT, timeout=10,
)
class SharedCoreWiring(unittest.TestCase):
def test_both_runners_source_shared_core_and_avoid_ps_grep(self):
for runner in (VM_RUNNER, BARE_RUNNER):
with open(runner) as f:
source = f.read()
self.assertIn('source "$SCRIPT_DIR/lib/run-common.sh"', source)
self.assertNotIn("ps aux | grep '[b]ash archsetup'", source)
def test_vm_runner_adapts_the_password_taking_transport(self):
with open(VM_RUNNER) as f:
source = f.read()
self.assertIn('vm_exec "$ROOT_PASSWORD" "$@"', source)
self.assertIn("archsetup_process_running vm_transport", source)
self.assertIn("wait_for_archsetup vm_transport", source)
class PollArchsetup(unittest.TestCase):
def test_uses_one_pgrep_liveness_probe_until_process_stops(self):
proc = run_bash(textwrap.dedent("""\
checks=0
transport() {
[ "$1" = "pgrep -f '[b]ash archsetup' > /dev/null" ] ||
{ echo "bad-command:$1"; return 2; }
checks=$((checks + 1))
[ "$checks" -lt 3 ]
}
progress() { echo "progress:$1"; }
wait_for_archsetup transport 5 0 1 progress
echo "rc=$? checks=$checks polls=$ARCHSETUP_POLL_COUNT"
"""))
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertNotIn("bad-command", proc.stdout)
self.assertIn("progress:1", proc.stdout)
self.assertIn("progress:2", proc.stdout)
self.assertIn("rc=0 checks=3 polls=2", proc.stdout)
def test_returns_124_at_poll_limit(self):
proc = run_bash(textwrap.dedent("""\
transport() { return 0; }
set +e
wait_for_archsetup transport 2 0 10
rc=$?
set -e
echo "rc=$rc polls=$ARCHSETUP_POLL_COUNT"
"""))
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("rc=124 polls=2", proc.stdout)
class Report(unittest.TestCase):
def test_writes_common_report_with_transport_specific_details(self):
with tempfile.TemporaryDirectory() as d:
report = os.path.join(d, "report.txt")
proc = run_bash(textwrap.dedent(f"""\
write_archsetup_test_report "{report}" \
"Bare Metal ArchSetup Test Report" "stamp" \
"Bare Metal ZFS" "Target: ratio.local" \
yes true 12 1 2 "Results: {d}/"
"""))
self.assertEqual(proc.returncode, 0, proc.stderr)
with open(report) as f:
text = f.read()
self.assertIn("Bare Metal ArchSetup Test Report", text)
self.assertIn("Test Method: Bare Metal ZFS", text)
self.assertIn("Target: ratio.local", text)
self.assertIn("ArchSetup Completed: yes", text)
self.assertIn("Validation: PASSED", text)
self.assertIn("Passed: 12", text)
self.assertIn(f"Results: {d}/", text)
if __name__ == "__main__":
unittest.main()
|
