From c4a72995c6bf2465a6540a1b546a9dcd732944af Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 4 Oct 2026 18:40:58 +0100 Subject: [PATCH] Keep passive VM diagnostics separate from planned server pause controls The completed PR smoke job demonstrates two obsolete whole-guide assertions after the reviewed manual backup precaution added six server-control examples. Select the unique passive preflight section for helper execution, constrain planned pause examples exactly, and reject active probes or service operations outside that section. Production code and safe guidance remain unchanged. Change-source: pulse-maintainer --- scripts/tests/test_vm_disk_diagnostics.py | 53 +++++++++++++++++++++-- 1 file changed, 50 insertions(+), 3 deletions(-) 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)