1from __future__ import annotations23import asyncio4import tempfile5import unittest6from pathlib import Path7from typing import Any8from unittest import mock910from goq_sigil.controller import Controller11from goq_sigil.errors import ControllerError12from goq_sigil.runner import CommandResult, Runner, strict_json131415REVISION_A = "sha256:" + "a" * 6416REVISION_B = "sha256:" + "b" * 6417TRANSACTION = "1" * 3218FINGERPRINT = "12345678…90abcdef"192021def request() -> dict[str, Any]:22 return {23 "schema_version": 1,24 "expected_revision": REVISION_A,25 "settings": {26 "resolution": {"mode": "native"},27 "framerate": 60,28 "rate_control": {"mode": "cbr", "bitrate_kbps": 12000},29 },30 }313233class FakeRunner:34 def __init__(self, root: Path):35 self.home = root36 self.sigil = root / "sigil"37 self.sigil.write_text("fixture", encoding="utf-8")38 self.config = root / "host.toml"39 self.systemctl = Path("/usr/bin/systemctl")40 self.unit = "sigil-host.service"41 self.active = True42 self.revision = REVISION_A43 self.instance = "2" * 3244 self.next_instance = 245 self.calls: list[tuple[str, Any]] = []46 self.fail_ready = False47 self.keep_stale_instance = False48 self.fail_start_number: int | None = None49 self.start_count = 050 self.pending: dict[str, Any] | None = None51 self.fail_rollback = False52 self.lose_commit_response = False5354 async def systemctl_command(55 self, action: str, *, check: bool = True, timeout: float = 15.056 ) -> CommandResult:57 self.calls.append(("systemctl", action))58 if action == "show":59 state = "active" if self.active else "inactive"60 output = (61 "LoadState=loaded\n"62 f"ActiveState={state}\n"63 f"SubState={'running' if self.active else 'dead'}\n"64 "UnitFileState=enabled\n"65 ).encode()66 return CommandResult(0, output, b"")67 if action in {"start", "restart"}:68 self.start_count += 169 if self.fail_start_number == self.start_count:70 raise ControllerError("command_failed")71 self.active = True72 if not self.keep_stale_instance:73 self.next_instance += 174 self.instance = f"{self.next_instance:032x}"75 elif action == "stop":76 self.active = False77 return CommandResult(0, b"", b"")7879 async def sigil_json(80 self,81 arguments: list[str],82 *,83 stdin_value: dict[str, Any] | None = None,84 schema_version: int = 1,85 timeout: float = 10.0,86 ) -> dict[str, Any]:87 operation = tuple(arguments[:3])88 self.calls.append(("sigil", operation))89 if operation == ("appliance", "status", "--config"):90 ready = self.active and not (91 self.fail_ready and self.revision == REVISION_B92 )93 return {94 "schema_version": 2,95 "sigil_version": "0.1.0",96 "overall": "ready" if ready else "degraded",97 "identity": {"host_fingerprint": FINGERPRINT},98 "enrollment": {"state": "none", "grants": [], "epoch": 0},99 "runtime": {100 "state": "fresh",101 "daemon": "ready" if ready else "degraded",102 "session": "inactive",103 "instance_id": self.instance,104 "loaded_config_revision": self.revision,105 },106 "config": {"revision": self.revision, "pending_transaction": None},107 }108 if operation == ("appliance", "config", "show"):109 return {110 "schema_version": 1,111 "revision": self.revision,112 "settings": request()["settings"],113 "pending_transaction": self.pending,114 }115 if operation == ("appliance", "config", "validate"):116 return {117 "schema_version": 1,118 "base_revision": REVISION_A,119 "candidate_revision": REVISION_B,120 "changed": True,121 "settings": stdin_value["settings"],122 }123 if operation == ("appliance", "config", "set"):124 self.revision = REVISION_B125 self.pending = {126 "transaction": TRANSACTION,127 "base_revision": REVISION_A,128 "candidate_revision": REVISION_B,129 "state": "pending_validation",130 }131 return {132 "schema_version": 1,133 "transaction": TRANSACTION,134 "candidate_revision": REVISION_B,135 "changed": True,136 "restart_required": True,137 }138 if operation == ("appliance", "config", "commit"):139 self.pending = None140 if self.lose_commit_response:141 raise ControllerError("command_timeout")142 return {143 "schema_version": 1,144 "operation": "config_commit",145 "transaction": TRANSACTION,146 "revision": REVISION_B,147 }148 if operation == ("appliance", "config", "rollback"):149 if self.fail_rollback:150 raise ControllerError("command_failed")151 self.revision = REVISION_A152 self.pending = None153 return {154 "schema_version": 1,155 "operation": "config_rollback",156 "transaction": TRANSACTION,157 "restored_revision": REVISION_A,158 }159 if operation == ("appliance", "enrollment-reset", "--config"):160 return {161 "schema_version": 1,162 "operation": "enrollment_reset",163 "host_fingerprint": FINGERPRINT,164 "had_enrollment": True,165 "previous_epoch": 4,166 "current_epoch": 5,167 "invitations_invalidated": True,168 }169 raise AssertionError(arguments)170171172class FastController(Controller):173 async def _wait_ready(self, **kwargs):174 if self.runner.fail_ready and self.runner.revision == REVISION_B:175 raise ControllerError("service_ready_timeout")176 if (177 self.runner.keep_stale_instance178 and kwargs.get("previous_instance") == self.runner.instance179 ):180 raise ControllerError("service_ready_timeout")181 return await super()._wait_ready(**kwargs)182183184class ControllerTests(unittest.IsolatedAsyncioTestCase):185 def setUp(self):186 self.temporary = tempfile.TemporaryDirectory()187 self.runner = FakeRunner(Path(self.temporary.name))188 self.controller = FastController(self.runner)189190 def tearDown(self):191 self.temporary.cleanup()192193 async def test_snapshot_combines_fixed_service_and_redacted_appliance_contract(self):194 snapshot = await self.controller.get_snapshot()195 self.assertTrue(snapshot["installed"])196 self.assertTrue(snapshot["service"]["active"])197 self.assertEqual(198 snapshot["appliance"]["identity"]["host_fingerprint"], FINGERPRINT199 )200 self.assertFalse(snapshot["capabilities"]["factory_reset"])201202 async def test_apply_uses_health_bound_commit_and_restores_active_service(self):203 result = await self.controller.apply_config(request())204 self.assertTrue(result["changed"])205 self.assertEqual(result["revision"], REVISION_B)206 significant = [207 value208 for kind, value in self.runner.calls209 if kind == "systemctl" and value != "show"210 ]211 self.assertEqual(significant, ["stop", "start", "stop", "start"])212 sigil_operations = [213 value for kind, value in self.runner.calls if kind == "sigil"214 ]215 self.assertIn(("appliance", "config", "validate"), sigil_operations)216 self.assertIn(("appliance", "config", "set"), sigil_operations)217 self.assertIn(("appliance", "config", "commit"), sigil_operations)218 self.assertNotIn(("appliance", "config", "rollback"), sigil_operations)219220 async def test_candidate_health_failure_rolls_back_and_restores_service(self):221 self.runner.fail_ready = True222 with self.assertRaisesRegex(ControllerError, "apply_rolled_back"):223 await self.controller.apply_config(request())224 self.assertEqual(self.runner.revision, REVISION_A)225 self.assertTrue(self.runner.active)226 self.assertIn(227 ("sigil", ("appliance", "config", "rollback")), self.runner.calls228 )229230 async def test_stale_snapshot_cannot_prove_restart(self):231 self.runner.keep_stale_instance = True232 with self.assertRaisesRegex(ControllerError, "service_ready_timeout"):233 await self.controller.restart_service()234235 async def test_wait_ready_timeout_reports_last_polling_error(self):236 controller = Controller(self.runner)237238 async def failing_status() -> dict[str, Any]:239 raise ControllerError("command_timeout")240241 controller._status = failing_status242 with (243 mock.patch("goq_sigil.controller.READY_TIMEOUT_SECONDS", 0.02),244 mock.patch("goq_sigil.controller.POLL_SECONDS", 0.0),245 self.assertRaises(ControllerError) as caught,246 ):247 await controller._wait_ready(previous_instance="0" * 32)248 self.assertEqual(caught.exception.code, "service_ready_timeout")249 self.assertEqual(caught.exception.detail, "command_timeout")250 self.assertIsInstance(caught.exception.__cause__, ControllerError)251 self.assertEqual(caught.exception.__cause__.code, "command_timeout")252253 async def test_wait_ready_timeout_reports_unmet_condition(self):254 controller = Controller(self.runner)255256 async def degraded_status() -> dict[str, Any]:257 return {"overall": "degraded", "runtime": {}}258259 controller._status = degraded_status260 with (261 mock.patch("goq_sigil.controller.READY_TIMEOUT_SECONDS", 0.02),262 mock.patch("goq_sigil.controller.POLL_SECONDS", 0.0),263 self.assertRaises(ControllerError) as caught,264 ):265 await controller._wait_ready(previous_instance="0" * 32)266 self.assertEqual(caught.exception.code, "service_ready_timeout")267 self.assertEqual(caught.exception.detail, "overall='degraded'")268 self.assertIsNone(caught.exception.__cause__)269270 async def test_restart_from_proven_absent_runtime_accepts_first_instance(self):271 self.runner.instance = None272 self.runner.active = False273 result = await self.controller.restart_service()274 self.assertEqual(result["operation"], "restart")275 self.assertTrue(self.runner.active)276277 async def test_lost_commit_response_is_recovered_without_rollback(self):278 self.runner.lose_commit_response = True279 result = await self.controller.apply_config(request())280 self.assertTrue(result["commit_response_recovered"])281 self.assertEqual(result["revision"], REVISION_B)282 self.assertTrue(self.runner.active)283 self.assertNotIn(284 ("sigil", ("appliance", "config", "rollback")), self.runner.calls285 )286287 async def test_post_commit_start_failure_does_not_attempt_impossible_rollback(self):288 self.runner.fail_start_number = 2289 with self.assertRaisesRegex(ControllerError, "post_commit_service_unhealthy"):290 await self.controller.apply_config(request())291 self.assertEqual(self.runner.revision, REVISION_B)292 self.assertNotIn(293 ("sigil", ("appliance", "config", "rollback")), self.runner.calls294 )295 self.assertTrue(self.runner.active)296297 async def test_failed_explicit_rollback_leaves_unproven_candidate_stopped(self):298 self.runner.revision = REVISION_B299 self.runner.pending = {300 "transaction": TRANSACTION,301 "base_revision": REVISION_A,302 "candidate_revision": REVISION_B,303 "state": "pending_validation",304 }305 self.runner.fail_rollback = True306 with self.assertRaisesRegex(ControllerError, "rollback_failed_safe"):307 await self.controller.rollback_pending(TRANSACTION)308 self.assertFalse(self.runner.active)309310 async def test_reset_stops_before_mutation_and_restores_prior_state(self):311 result = await self.controller.reset_enrollment(FINGERPRINT)312 self.assertEqual(result["current_epoch"], 5)313 stop = self.runner.calls.index(("systemctl", "stop"))314 reset = self.runner.calls.index(315 ("sigil", ("appliance", "enrollment-reset", "--config"))316 )317 start = self.runner.calls.index(("systemctl", "start"))318 self.assertLess(stop, reset)319 self.assertLess(reset, start)320321 async def test_mutations_are_serialized(self):322 entered = asyncio.Event()323 release = asyncio.Event()324325 async def hold_restart():326 async with self.controller._mutation_lock:327 entered.set()328 await release.wait()329330 first = asyncio.create_task(hold_restart())331 await entered.wait()332 second = asyncio.create_task(self.controller.reset_enrollment(FINGERPRINT))333 await asyncio.sleep(0)334 self.assertFalse(second.done())335 release.set()336 await first337 await second338339 async def test_invalid_frontend_shapes_never_reach_sigil(self):340 invalid = request()341 invalid["settings"]["resolution"] = {342 "mode": "fixed",343 "width": 1279,344 "height": 800,345 }346 with self.assertRaisesRegex(ControllerError, "invalid_request"):347 await self.controller.validate_config(invalid)348349350class JsonTests(unittest.TestCase):351 def test_duplicate_keys_are_rejected(self):352 with self.assertRaisesRegex(ControllerError, "invalid_response"):353 strict_json(b'{"schema_version":1,"schema_version":1}', 1)354355 def test_wrong_schema_is_rejected(self):356 with self.assertRaisesRegex(ControllerError, "incompatible_sigil"):357 strict_json(b'{"schema_version":2}', 1)358359360class HangingStream:361 async def read(self, _size):362 await asyncio.Future()363364365class FakeProcess:366 def __init__(self):367 self.returncode = None368 self.stdin = None369 self.stdout = HangingStream()370 self.stderr = HangingStream()371 self.terminated = False372 self._done = asyncio.Event()373374 def terminate(self):375 self.terminated = True376 self.returncode = -15377 self._done.set()378379 def kill(self):380 self.returncode = -9381 self._done.set()382383 async def wait(self):384 await self._done.wait()385 return self.returncode386387388class TestRunner(Runner):389 def environment(self):390 return {}391392393class RunnerTests(unittest.IsolatedAsyncioTestCase):394 async def test_cancellation_terminates_and_reaps_child(self):395 process = FakeProcess()396397 async def create(*_args, **_kwargs):398 return process399400 runner = TestRunner("/fixture", uid=1000)401 with mock.patch("asyncio.create_subprocess_exec", create):402 task = asyncio.create_task(runner.run(["/fixed-command"], timeout=60))403 await asyncio.sleep(0)404 task.cancel()405 with self.assertRaises(asyncio.CancelledError):406 await task407 self.assertTrue(process.terminated)408 self.assertIsNotNone(process.returncode)409410411if __name__ == "__main__":412 unittest.main()