diff --git a/scripts/tests/test_vm_disk_diagnostics.py b/scripts/tests/test_vm_disk_diagnostics.py index adce1fb149..02d58b36d0 100644 --- a/scripts/tests/test_vm_disk_diagnostics.py +++ b/scripts/tests/test_vm_disk_diagnostics.py @@ -14,6 +14,7 @@ import re import shutil import subprocess import tempfile +import textwrap import time import unittest @@ -26,6 +27,20 @@ REAL_TIMEOUT = shutil.which("timeout") PRIVATE = "synthetic-private-infrastructure-marker" READS = [["status", "100"], ["config", "100", "--current"]] + +def guide_section(guide, heading): + """Select one named H3, never another section's executable examples.""" + sections = re.findall(r"^### " + re.escape(heading) + + r"\n(.*?)(?=^#{1,3} |\Z)", guide, re.MULTILINE | re.DOTALL) + if len(sections) != 1: + raise AssertionError(f"Expected one guide section: {heading}") + return sections[0] + + +def bash_examples(section): + return [textwrap.dedent(block).strip() + for block in re.findall(r"```bash\n(.*?)```", section, re.DOTALL)] + # Record every attempted operation, including prohibited ones. The timeout # adapter checks the helper's real deadline arguments; one test uses real # coreutils timeout around the synthetic blocked qm child. @@ -250,7 +265,7 @@ class VMDiskDiagnosticsTest(unittest.TestCase): self.assert_passive(calls, operations=READS[:1]) def test_exact_documented_command_uses_the_passive_helper(self): - recipes = re.findall(r"```bash\n(.*?)```", DOC.read_text(), re.DOTALL) + recipes = bash_examples(guide_section(DOC.read_text(), "Passive host preflight")) self.assertEqual(len(recipes), 1) result, calls = self.exercise(recipe=recipes[0]) self.assertEqual(result.returncode, 0, result.stderr) @@ -272,6 +287,39 @@ class VMDiskDiagnosticsTest(unittest.TestCase): class VMDiskHelpTest(unittest.TestCase): + def assert_guide_commands(self, guide): + # The planned server pause is intentionally active, unlike the passive + # hypervisor diagnostic. Admit only the six reviewed server/timer blocks + # here; do not permit arbitrary systemctl or guest commands elsewhere. + precaution = guide_section(guide, "Pause Pulse for a planned freeze-enabled backup") + self.assertEqual(bash_examples(precaution), [ + "systemctl show pulse.service --property=LoadState,ActiveState,MainPID", + "systemctl show pulse-update.timer --property=LoadState,ActiveState", + "sudo systemctl stop pulse-update.timer\n" + "systemctl show pulse-update.timer pulse-update.service \\\n" + " --property=Id,LoadState,ActiveState,MainPID", + "sudo systemctl stop pulse.service\n" + "systemctl show pulse.service --property=LoadState,ActiveState,MainPID", + "sudo systemctl start pulse.service\nsystemctl is-active pulse.service", + "sudo systemctl start pulse-update.timer", + ]) + for block in bash_examples(guide.replace(precaution, "", 1)): + self.assertNotRegex(block, r"\b(curl|wget|qm agent|pveum|systemctl)\b") + + def test_command_boundaries_reject_active_diagnostics_and_unreviewed_pause_operations(self): + guide = DOC.read_text() + mutations = [ + guide.replace("sudo bash ./scripts/test-vm-disk.sh 100", + "systemctl restart pulse.service"), + guide.replace("sudo systemctl stop pulse.service", "sudo systemctl restart pulse.service"), + guide.replace("sudo systemctl start pulse-update.timer", "qm agent 100 ping"), + guide + "\n```bash\nsystemctl stop pulse.service\n```\n", + guide.replace("sudo bash ./scripts/test-vm-disk.sh 100", "curl https://example.invalid"), + ] + for index, mutated in enumerate(mutations): + with self.subTest(mutation=index), self.assertRaises(AssertionError): + self.assert_guide_commands(mutated) + def test_guides_are_mirrored_and_preserve_the_safety_and_proof_boundaries(self): for name in ("VM_DISK_MONITORING", "TROUBLESHOOTING"): source = (ROOT / f"docs/{name}.md").read_bytes() @@ -281,8 +329,7 @@ class VMDiskHelpTest(unittest.TestCase): "does not prove thaw", "a backup can", "older copies", "non-zero exit", "host-root diagnostic does not test", "not a claim", "bounded timeouts"): self.assertIn(phrase, guide) - for block in re.findall(r"```bash\n(.*?)```", guide, re.DOTALL): - self.assertNotRegex(block, r"\b(curl|wget|qm agent|pveum|systemctl)\b") + self.assert_guide_commands(guide) self.assertNotIn("GUEST_AGENT_FSINFO_TIMEOUT=", guide)