"""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()