scripts/test_macos_repair_qa.py
Rama main · réplica de lectura del repositorio original.
"""Guard and rollback contracts; these tests never call the user's service."""
import importlib.util
from contextlib import redirect_stdout
import io
import json
from pathlib import Path
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import patch
spec = importlib.util.spec_from_file_location("repair_qa", Path(__file__).with_name("ci-qa-macos-repair.py"))
qa = importlib.util.module_from_spec(spec)
spec.loader.exec_module(qa)
class RepairQA(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory()
self.root = Path(self.directory.name).resolve()
self.home, self.install = self.root / "micromanager", self.root / "bin"
self.addCleanup(self.directory.cleanup)
def assets(self):
for path in [self.install / "firecrab", self.install / "firecrab-micromanager-macos"] + [
self.home / file for file in (
"system/debian-system.raw", "data/firecrab-data.raw", "system/Image", "system/initrd.img",
"runtime/manager_ed25519", "runtime/provisioned", "runtime/known_hosts", "downloads/archive",
)
]:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("preserved\n")
def test_progress_requires_all_six_stages_in_order(self):
stages = [f"[REPAIR {step}/6] operation" for step in range(1, 7)]
output = "\n[WAIT] pending\n".join(stages) + "\n[PASS] ready\n"
qa.require_repair_progress(output)
for lines in (stages[:-1], stages[::-1], stages + [stages[-1]], []):
with self.assertRaisesRegex(qa.QAError, "G7b"):
qa.require_repair_progress("\n".join(lines))
def test_successful_repair_without_progress_fails_the_runtime_qa(self):
runner = qa.QA("cli", self.home, self.install, self.root / "results")
self.addCleanup(runner.log.close)
with patch.object(runner, "service", return_value=SimpleNamespace(stdout="ready\n")):
with redirect_stdout(io.StringIO()), self.assertRaises(qa.QAError):
runner.repair()
check = runner.summary["checks"][-1]
self.assertEqual(check["id"], "G7b")
self.assertEqual(check["status"], "FAILED")
def test_running_transitional_and_unknown_vms_block_mutation(self):
for state in ("running", "starting", "stopping", "unknown", None):
with self.assertRaises(qa.QAError):
qa.require_idle([{"id": "vm", "state": state}])
qa.require_idle([])
qa.require_idle([{"state": state} for state in ("created", "stopped", "error")])
def test_disk_replacement_binary_change_and_host_key_removal_are_detected(self):
self.assets()
baseline = qa.snapshot(self.home, self.install)
qa.assert_preserved(baseline, qa.snapshot(self.home, self.install))
disk = self.home / "data/firecrab-data.raw"
disk.rename(disk.with_suffix(".backup"))
disk.write_text("preserved\n")
with self.assertRaisesRegex(qa.QAError, "disks"):
qa.assert_preserved(baseline, qa.snapshot(self.home, self.install))
disk.unlink()
disk.with_suffix(".backup").rename(disk)
(self.install / "firecrab").write_text("replaced")
with self.assertRaisesRegex(qa.QAError, "hashes"):
qa.assert_preserved(baseline, qa.snapshot(self.home, self.install))
(self.install / "firecrab").write_text("preserved\n")
(self.home / "runtime/known_hosts").write_text("")
with self.assertRaisesRegex(qa.QAError, "host keys"):
qa.assert_preserved(baseline, qa.snapshot(self.home, self.install))
def test_boot_writes_and_new_host_pins_do_not_count_as_disk_replacement(self):
self.assets()
baseline = qa.snapshot(self.home, self.install)
(self.home / "system/debian-system.raw").write_text("bootwrite\n")
(self.home / "runtime/known_hosts").write_text("preserved\nnew host\n")
qa.assert_preserved(baseline, qa.snapshot(self.home, self.install))
def test_missing_optional_host_pins_are_allowed_and_symlinks_rejected(self):
self.assets()
known_hosts = self.home / "runtime/known_hosts"
known_hosts.unlink()
baseline = qa.snapshot(self.home, self.install)
self.assertEqual(baseline["known_hosts"], set())
known_hosts.write_text("new pinned host\n")
qa.assert_preserved(baseline, qa.snapshot(self.home, self.install))
known_hosts.unlink()
known_hosts.symlink_to(self.home / "runtime/manager_ed25519")
with self.assertRaisesRegex(qa.QAError, "symlinked SSH host pins"):
qa.snapshot(self.home, self.install)
def test_real_gate_snapshots_and_backs_up_idle_registration(self):
self.assets()
runner = qa.QA("cli", self.home, self.install, self.root / "results")
self.addCleanup(runner.log.close)
runner.plist = self.root / "LaunchAgents/service.plist"
runner.plist.parent.mkdir()
runner.plist.write_text("original plist")
runner.wrapper.write_text("original wrapper")
def command(args, **_kwargs):
if args[0] == "/bin/launchctl":
return SimpleNamespace(stdout=str(runner.wrapper))
return SimpleNamespace(stdout="revision\n" if args[-1] == "HEAD" else "")
with patch.object(runner, "run", side_effect=command), \
patch.object(runner, "service") as service, patch.object(runner, "api", return_value=[]), \
patch.object(runner, "config_digest", return_value="guest config"):
runner.gate()
service.assert_called_once_with("status")
self.assertFalse(runner.mutated)
self.assertEqual(runner.summary["revision"], "revision")
self.assertFalse(runner.summary["dirty"])
self.assertEqual(runner.config_before, "guest config")
self.assertEqual(runner.files_before, qa.snapshot(self.home, self.install))
self.assertEqual(len(runner.backups), 2)
for original, backup in runner.backups.items():
self.assertEqual(original.read_bytes(), backup.read_bytes())
def test_real_gate_rejects_active_vm_before_registration_backup(self):
runner = qa.QA("cli", self.home, self.install, self.root / "results")
self.addCleanup(runner.log.close)
runner.plist = self.root / "LaunchAgents/service.plist"
with patch.object(runner, "run", return_value=SimpleNamespace(stdout="")), \
patch.object(runner, "api", return_value=[{"id": "vm", "state": "running"}]), \
patch.object(runner, "service") as service:
with self.assertRaisesRegex(qa.QAError, "stop workloads first"):
runner.gate()
service.assert_not_called()
self.assertFalse(runner.mutated)
self.assertEqual(runner.backups, {})
self.assertFalse((runner.results / "registration-backup").exists())
def test_failed_gate_never_stops_or_restores_service(self):
runner = qa.QA("cli", self.home, self.install, self.root / "results")
with patch.object(runner, "gate", side_effect=qa.QAError("active VM")), \
patch.object(runner, "service") as service, patch.object(runner, "restore") as restore, \
redirect_stdout(io.StringIO()):
self.assertEqual(runner.execute(), 1)
service.assert_not_called()
restore.assert_not_called()
summary = json.loads((runner.results / "summary.json").read_text())
self.assertEqual(summary["status"], "FAILED")
self.assertEqual(summary["checks"][0]["status"], "FAILED")
def test_failed_repair_restores_registration_and_records_failure(self):
runner = qa.QA("cli", self.home, self.install, self.root / "results")
wrapper = self.root / "daemon.sh"
backup = self.root / "backup"
wrapper.write_text("damaged")
backup.write_text("original")
runner.backups[wrapper] = backup
def fail():
runner.mutated = True
raise qa.QAError("repair failed")
with patch.object(runner, "gate"), patch.object(runner, "restart", side_effect=fail), \
patch.object(runner, "service") as service, patch.object(runner, "api"), \
redirect_stdout(io.StringIO()):
self.assertEqual(runner.execute(), 1)
self.assertEqual(wrapper.read_text(), "original")
self.assertEqual([call.args[0] for call in service.call_args_list], ["stop", "start", "status"])
self.assertEqual(runner.summary["restoration"], "PASS")
self.assertEqual(runner.summary["status"], "FAILED")
def test_failed_runtime_records_preservation_even_when_recovery_fails(self):
for changed in (False, True):
with self.subTest(changed=changed):
self.assets()
runner = qa.QA("cli", self.home, self.install, self.root / f"results-{changed}")
def gate():
runner.files_before = qa.snapshot(self.home, self.install)
runner.record_snapshot("preservation-before.json", runner.files_before)
def fail():
runner.mutated = True
if changed:
(self.install / "firecrab").write_text("unexpected replacement")
raise qa.QAError("runtime failed")
with patch.object(runner, "gate", side_effect=gate), \
patch.object(runner, "restart", side_effect=fail), \
patch.object(runner, "restore", side_effect=qa.QAError("recovery failed")), \
patch.object(runner, "service") as service, redirect_stdout(io.StringIO()):
self.assertEqual(runner.execute(), 1)
self.assertEqual(runner.summary["error"], "runtime failed")
self.assertIn("recovery failed", runner.summary["restoration"])
service.assert_called_once_with("stop", timeout=60)
self.assertTrue(runner.summary["cleanup"].startswith("PASS"))
expected = "FAILED" if changed else "PASS"
self.assertTrue(runner.summary["preservation"].startswith(expected))
for name in ("preservation-before.json", "preservation-after.json"):
evidence = json.loads((runner.results / name).read_text())
self.assertNotIn("preserved", evidence["known_hosts"])
self.assertEqual(len(evidence["known_hosts"][0]), 64)
def test_interrupted_qa_still_restores_the_service(self):
runner = qa.QA("cli", self.home, self.install, self.root / "results")
def interrupt():
runner.mutated = True
raise KeyboardInterrupt()
with patch.object(runner, "gate"), patch.object(runner, "restart", side_effect=interrupt), \
patch.object(runner, "restore") as restore, redirect_stdout(io.StringIO()):
self.assertEqual(runner.execute(), 1)
restore.assert_called_once()
self.assertEqual(runner.summary["error"], "interrupted")
if __name__ == "__main__":
unittest.main()