from __future__ import annotations import io import json import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from pathlib import Path from unittest.mock import patch from lumi_eggcracker.cli import main def doctor_response(**overrides: object) -> dict[str, object]: value: dict[str, object] = { "autonomous_discovery": True, "backend": "root-supervisor", "cgroup_v2": True, "catalogue": {"secret": "catalogue-canary"}, "discovery": { "healthy": True, "receipt_persistence_healthy": True, "path": "/private/discovery-canary", }, "execution_boundary": {"supported": True, "argv": ["argv-canary"]}, "incidents": {"healthy": True, "lockdown": False, "secret": "incident-canary"}, "installation": {"state": "HEALTHY", "path": "/private/install-canary"}, "network": { "cleanup_healthy": True, "mode": "offline", "primitives": {"supported": True, "uid": "uid-canary"}, }, "pidfd": True, "result": "PASS", "version": "1.0.10", "workload_identity": {"healthy": True, "uid": 4242}, "workload_uid": 4242, } value.update(overrides) return value def wait_status(*, state: str = "RUNNING", run_id: str = "a" * 24) -> dict[str, object]: return { "exec_policy_digest": "0" * 64, "exec_policy_id": "legacy", "incident_lockdown": False, "name": "demo", "network_mode": "offline", "run_id": run_id, "state": state, "unit": f"lumi-eggcracker-workload-{run_id}.service", "workload_uid": 4242, } class CliTests(unittest.TestCase): def test_version(self) -> None: output = io.StringIO() with redirect_stdout(output): self.assertEqual(0, main(["version"])) self.assertEqual("1.0.10", output.getvalue().strip()) def test_public_help_has_only_supported_commands(self) -> None: from lumi_eggcracker.cli import _parser help_text = _parser().format_help() for command in ( "start", "check-start", "kill", "status", "wait", "list", "history", "approve", "revoke", "approvals", "detections", "doctor", "support-bundle", "validate-support-bundle", "compare-detection-receipts", "version", ): self.assertIn(command, help_text) self.assertIn("exec-policy", help_text) self.assertNotIn("_supervisor", help_text) self.assertNotIn("network" + "-deny", help_text) self.assertIn( "retained terminal records in recorded monotonic order", " ".join(help_text.split()), ) self.assertNotIn("recent", help_text) self.assertNotIn("chronological", help_text) doctor_help = io.StringIO() with redirect_stdout(doctor_help), self.assertRaises(SystemExit) as raised: main(["doctor", "--help"]) self.assertEqual(0, raised.exception.code) self.assertIn("--recovery-plan", doctor_help.getvalue()) def test_history_defaults_to_twenty_and_forwards_explicit_limit(self) -> None: with patch("lumi_eggcracker.cli.request", return_value={"runs": []}) as request: output = io.StringIO() with redirect_stdout(output): self.assertEqual(0, main(["history"])) request.assert_called_once_with("history", limit=20) output.seek(0) self.assertEqual({"runs": []}, json.loads(output.read())) with patch("lumi_eggcracker.cli.request", return_value={"runs": []}) as request: with redirect_stdout(io.StringIO()): self.assertEqual(0, main(["history", "--limit", "7"])) request.assert_called_once_with("history", limit=7) def test_history_forwards_each_supported_state(self) -> None: for state in ( "COMPLETED_ALLOWED", "TERMINATED", "CONTAINMENT_FAILED", "CONTAINED_RECEIPT_FAILED", ): with self.subTest(state=state), patch( "lumi_eggcracker.cli.request", return_value={"runs": []} ) as request, redirect_stdout(io.StringIO()): self.assertEqual(0, main(["history", "--limit", "7", "--state", state])) request.assert_called_once_with("history", limit=7, state=state) def test_history_limit_rejects_invalid_duplicate_and_extra_arguments(self) -> None: cases = ( ["history", "--limit", "0"], ["history", "--limit", "21"], ["history", "--limit", "true"], ["history", "--limit", "1.0"], ["history", "--limit", "1", "--limit", "2"], ["history", "--limit", "1", "unexpected"], ["history", "--state", "RUNNING"], ["history", "--state", "terminated"], ["history", "--state"], ["history", "--state", "TERMINATED", "--state", "CONTAINMENT_FAILED"], ["history", "--state", "TERMINATED", "unexpected"], ) for values in cases: with self.subTest(values=values), patch( "lumi_eggcracker.cli.request" ) as request, redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as raised: main(values) self.assertEqual(2, raised.exception.code) request.assert_not_called() def test_doctor_summary_ready_is_fixed_and_nonidentifying(self) -> None: response = doctor_response() output = io.StringIO() error = io.StringIO() with ( patch("lumi_eggcracker.cli.request", return_value=response) as request, redirect_stdout(output), redirect_stderr(error), ): self.assertEqual(0, main(["doctor", "--summary"])) self.assertEqual( { "failed_checks": [], "next_action": "NONE", "result": "READY", "schema": "lumi-eggcracker.doctor-summary.v1", }, json.loads(output.getvalue()), ) self.assertEqual("", error.getvalue()) self.assertNotIn("canary", output.getvalue()) request.assert_called_once_with("doctor") def test_doctor_summary_mixed_failures_are_sorted_and_deduplicated(self) -> None: response = doctor_response( autonomous_discovery=False, cgroup_v2=False, execution_boundary={"supported": False}, discovery={"healthy": False, "receipt_persistence_healthy": False}, incidents={"healthy": False, "lockdown": True}, installation={"state": "DRIFT"}, network={ "cleanup_healthy": False, "mode": "unsupported", "primitives": {"supported": False}, }, pidfd=False, result="UNSUPPORTED", workload_identity={"healthy": False}, ) output = io.StringIO() with patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output): self.assertEqual(6, main(["doctor", "--summary"])) summary = json.loads(output.getvalue()) self.assertEqual( [ "cgroup_v2", "discovery", "execution_boundary", "incidents", "installation", "network_cleanup", "network_mode", "network_primitives", "pidfd", "receipt_persistence", "workload_identity", ], summary["failed_checks"], ) self.assertEqual(len(summary["failed_checks"]), len(set(summary["failed_checks"]))) self.assertEqual("NOT_READY", summary["result"]) self.assertEqual("REVIEW_SUPERVISOR_READINESS", summary["next_action"]) self.assertEqual( {"failed_checks", "next_action", "result", "schema"}, set(summary), ) self.assertNotIn("canary", output.getvalue()) def test_doctor_summary_unsupported_without_visible_failure_is_honest(self) -> None: response = doctor_response(autonomous_discovery=False, result="UNSUPPORTED") output = io.StringIO() with patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output): self.assertEqual(6, main(["doctor", "--summary"])) self.assertEqual( { "failed_checks": ["supervisor_readiness"], "next_action": "REVIEW_SUPERVISOR_READINESS", "result": "NOT_READY", "schema": "lumi-eggcracker.doctor-summary.v1", }, json.loads(output.getvalue()), ) def test_doctor_summary_malformed_or_contradictory_response_is_constant_error(self) -> None: malformed = [ {}, doctor_response(result=True), doctor_response(cgroup_v2="true"), doctor_response(network={"mode": "secret-path-canary"}), doctor_response(autonomous_discovery=False), doctor_response(autonomous_discovery=True, result="UNSUPPORTED"), doctor_response(incidents={"healthy": False, "lockdown": False}), doctor_response(installation={"state": "PRIVATE-CANARY"}), ] for response in malformed: with self.subTest(response=response): output = io.StringIO() error = io.StringIO() with ( patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output), redirect_stderr(error), ): self.assertEqual(6, main(["doctor", "--summary"])) self.assertEqual("", output.getvalue()) self.assertEqual("eggcracker: doctor summary unavailable\n", error.getvalue()) self.assertNotIn("canary", error.getvalue()) output = io.StringIO() error = io.StringIO() with patch("lumi_eggcracker.cli.request", side_effect=OSError("/private/CANARY")), redirect_stdout(output), redirect_stderr(error): self.assertEqual(6, main(["doctor", "--summary"])) self.assertEqual("", output.getvalue()) self.assertEqual("eggcracker: doctor summary unavailable\n", error.getvalue()) self.assertNotIn("CANARY", error.getvalue()) def test_default_doctor_output_and_exit_remain_unchanged(self) -> None: for result, expected_exit in (("PASS", 0), ("UNSUPPORTED", 6)): with self.subTest(result=result): response = doctor_response( autonomous_discovery=result == "PASS", result=result ) output = io.StringIO() error = io.StringIO() with patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output), redirect_stderr(error): self.assertEqual(expected_exit, main(["doctor"])) self.assertEqual(json.dumps(response, sort_keys=True) + "\n", output.getvalue()) self.assertEqual("", error.getvalue()) def test_doctor_recovery_plan_ready_is_closed_and_empty(self) -> None: response = doctor_response() output = io.StringIO() error = io.StringIO() with ( patch("lumi_eggcracker.cli.request", return_value=response) as request, redirect_stdout(output), redirect_stderr(error), ): self.assertEqual(0, main(["doctor", "--recovery-plan"])) self.assertEqual( { "actions": [], "failed_checks": [], "installation_state": "HEALTHY", "result": "READY", "schema": "lumi-eggcracker.doctor-recovery-plan.v1", }, json.loads(output.getvalue()), ) self.assertEqual("", error.getvalue()) self.assertNotIn("canary", output.getvalue()) request.assert_called_once_with("doctor") def test_doctor_recovery_plan_distinguishes_installation_states(self) -> None: expected_actions = { "NOT_INSTALLED": ["MANUAL_INSTALLATION_REQUIRED"], "DRIFT": ["MANUAL_INSTALLATION_REPAIR"], "RECOVERY_REQUIRED": ["MANUAL_INSTALLATION_RECOVERY"], } for state, actions in expected_actions.items(): with self.subTest(state=state): output = io.StringIO() response = doctor_response( autonomous_discovery=False, installation={"state": state, "path": "/private/install-canary"}, result="UNSUPPORTED", ) with ( patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output), ): self.assertEqual(6, main(["doctor", "--recovery-plan"])) plan = json.loads(output.getvalue()) self.assertEqual(state, plan["installation_state"]) self.assertEqual(["installation"], plan["failed_checks"]) self.assertEqual(actions, plan["actions"]) self.assertNotIn("canary", output.getvalue()) def test_doctor_recovery_plan_hidden_readiness_is_manual_supervisor_review(self) -> None: response = doctor_response(autonomous_discovery=False, result="UNSUPPORTED") output = io.StringIO() with ( patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output), ): self.assertEqual(6, main(["doctor", "--recovery-plan"])) self.assertEqual( { "actions": ["MANUAL_SUPERVISOR_REVIEW"], "failed_checks": ["supervisor_readiness"], "installation_state": "HEALTHY", "result": "NOT_READY", "schema": "lumi-eggcracker.doctor-recovery-plan.v1", }, json.loads(output.getvalue()), ) def test_doctor_recovery_plan_failures_are_sorted_and_deduplicated(self) -> None: response = doctor_response( autonomous_discovery=False, cgroup_v2=False, execution_boundary={"supported": False}, discovery={"healthy": False, "receipt_persistence_healthy": False}, incidents={"healthy": False, "lockdown": True}, installation={"state": "DRIFT"}, network={ "cleanup_healthy": False, "mode": "unsupported", "primitives": {"supported": False}, }, pidfd=False, result="UNSUPPORTED", workload_identity={"healthy": False}, ) output = io.StringIO() with ( patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output), ): self.assertEqual(6, main(["doctor", "--recovery-plan"])) plan = json.loads(output.getvalue()) self.assertEqual( [ "cgroup_v2", "discovery", "execution_boundary", "incidents", "installation", "network_cleanup", "network_mode", "network_primitives", "pidfd", "receipt_persistence", "workload_identity", ], plan["failed_checks"], ) self.assertEqual( [ "MANUAL_HOST_REVIEW", "MANUAL_IDENTITY_REVIEW", "MANUAL_INCIDENT_REVIEW", "MANUAL_INSTALLATION_REPAIR", "MANUAL_POLICY_REVIEW", "MANUAL_SUPERVISOR_REVIEW", ], plan["actions"], ) self.assertEqual(len(plan["failed_checks"]), len(set(plan["failed_checks"]))) self.assertEqual(len(plan["actions"]), len(set(plan["actions"]))) def test_doctor_recovery_plan_malformed_or_transport_response_is_constant_error(self) -> None: malformed = [ {}, doctor_response(result=True), doctor_response(installation={"state": "PRIVATE-CANARY"}), doctor_response(autonomous_discovery=True, result="UNSUPPORTED"), ] for response in malformed: with self.subTest(response=response): output = io.StringIO() error = io.StringIO() with ( patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output), redirect_stderr(error), ): self.assertEqual(6, main(["doctor", "--recovery-plan"])) self.assertEqual("", output.getvalue()) self.assertEqual( "eggcracker: doctor recovery plan unavailable\n", error.getvalue() ) self.assertNotIn("canary", error.getvalue()) output = io.StringIO() error = io.StringIO() with ( patch("lumi_eggcracker.cli.request", side_effect=OSError("/private/CANARY")), redirect_stdout(output), redirect_stderr(error), ): self.assertEqual(6, main(["doctor", "--recovery-plan"])) self.assertEqual("", output.getvalue()) self.assertEqual( "eggcracker: doctor recovery plan unavailable\n", error.getvalue() ) self.assertNotIn("CANARY", error.getvalue()) def test_doctor_projections_cannot_be_combined_before_transport(self) -> None: with ( patch("lumi_eggcracker.cli.request") as request, redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as raised, ): main(["doctor", "--summary", "--recovery-plan"]) self.assertEqual(2, raised.exception.code) request.assert_not_called() def test_compare_detection_receipts_help_uses_installed_command(self) -> None: output = io.StringIO() with redirect_stdout(output), self.assertRaises(SystemExit) as raised: main(["compare-detection-receipts", "--help"]) self.assertEqual(0, raised.exception.code) self.assertIn("usage: eggcracker compare-detection-receipts", output.getvalue()) self.assertIn("--before FILE", output.getvalue()) self.assertIn("--after FILE", output.getvalue()) def test_compare_detection_receipts_forwards_arguments_before_connected_commands( self, ) -> None: forwarded = [ "--before", "before-export.json", "--after", "after-export.json", "--output", "comparison.json", ] with ( patch( "lumi_eggcracker.cli.compare_detection_receipts_main", return_value=0 ) as compare, patch("lumi_eggcracker.cli.request") as request, patch("lumi_eggcracker.cli.supervisor_main") as supervisor, patch("lumi_eggcracker.cli.gate_main") as gate, patch("lumi_eggcracker.cli.watchdog_main") as watchdog, ): self.assertEqual(0, main(["compare-detection-receipts", *forwarded])) compare.assert_called_once_with( forwarded, prog="eggcracker compare-detection-receipts" ) request.assert_not_called() supervisor.assert_not_called() gate.assert_not_called() watchdog.assert_not_called() def test_compare_detection_receipts_success_and_existing_output_refusal(self) -> None: def synthetic_export(event_id: str) -> dict[str, object]: return { "export_schema": "lumi-eggcracker.redacted-detection-receipt-export.v1", "source_sha256": "f" * 64, "authentication": "NOT_AUTHENTICATED", "live_verification": "NOT_PERFORMED", "receipt": { "schema_version": "lumi-eggcracker.detection-receipt.v2", "event_id": event_id, "source_commit": "a" * 40, "version": "1.0.10", "catalogue_sha256": "b" * 64, "receipt_written_utc": "2026-10-02T00:00:00Z", "detector": { "profile": "content.gguf-llama", "detection_path": "CONTENT", "catalogue_schema": "lumi-eggcracker.detectors.v3", }, "trigger": {"kind": "UNAPPROVED_AI_MATCH"}, "recorded_result": "CONTAINMENT_FAILED", }, } with tempfile.TemporaryDirectory(prefix="cli-receipt-comparison-") as directory: root = Path(directory) before = root / "before.json" after = root / "after.json" before.write_text(json.dumps(synthetic_export("0" * 24)), encoding="utf-8") after.write_text(json.dumps(synthetic_export("1" * 24)), encoding="utf-8") output = root / "existing-comparison.json" original_output = b"preserve this synthetic report\n" output.write_bytes(original_output) stdout = io.StringIO() stderr = io.StringIO() with ( patch("lumi_eggcracker.cli.request") as request, patch("lumi_eggcracker.cli.supervisor_main") as supervisor, redirect_stdout(stdout), redirect_stderr(stderr), ): self.assertEqual( 0, main( [ "compare-detection-receipts", "--before", str(before), "--after", str(after), ] ), ) comparison = json.loads(stdout.getvalue()) self.assertEqual("RECORDED", comparison["comparison_status"]) self.assertEqual("", stderr.getvalue()) stdout.seek(0) stdout.truncate(0) stderr.seek(0) stderr.truncate(0) self.assertEqual( 1, main( [ "compare-detection-receipts", "--before", str(before), "--after", str(after), "--output", str(output), ] ), ) self.assertEqual("", stdout.getvalue()) self.assertIn("OUTPUT_ALREADY_EXISTS", stderr.getvalue()) request.assert_not_called() supervisor.assert_not_called() self.assertEqual(original_output, output.read_bytes()) def test_validate_support_bundle_dispatches_before_connected_commands(self) -> None: with ( patch("lumi_eggcracker.cli.validate_support_bundle_main", return_value=0) as validate, patch("lumi_eggcracker.cli.request") as request, patch("lumi_eggcracker.cli.supervisor_main") as supervisor, patch("lumi_eggcracker.cli.gate_main") as gate, patch("lumi_eggcracker.cli.watchdog_main") as watchdog, ): self.assertEqual( 0, main(["validate-support-bundle", "private-name-must-not-be-echoed.json"]), ) validate.assert_called_once_with(["private-name-must-not-be-echoed.json"]) request.assert_not_called() supervisor.assert_not_called() gate.assert_not_called() watchdog.assert_not_called() def test_internal_supervisor_dispatch_remains_available(self) -> None: with patch("lumi_eggcracker.cli.supervisor_main", return_value=7) as supervisor: self.assertEqual(main(["_supervisor", "--policy", "/tmp/policy.json"]), 7) supervisor.assert_called_once_with(["--policy", "/tmp/policy.json"]) def test_approve_forwards_only_exact_command_arguments(self) -> None: with patch("lumi_eggcracker.cli.os.geteuid", return_value=0, create=True), patch("lumi_eggcracker.cli.request", return_value={"result": "APPROVED"}) as request: self.assertEqual(0, main(["approve", "--name", "qwen", "--uid", "1001", "--", "/opt/llama-cli", "-m", "/models/qwen.gguf"])) request.assert_called_once_with( "approve", name="qwen", uid=1001, max_pids=64, max_memory_mib=2048, cpu_quota_percent=400, allow_interface_discovery=False, argv=["/opt/llama-cli", "-m", "/models/qwen.gguf"], ) def test_approve_can_grant_interface_discovery(self) -> None: with patch("lumi_eggcracker.cli.os.geteuid", return_value=0, create=True), patch("lumi_eggcracker.cli.request", return_value={"result": "APPROVED"}) as request: self.assertEqual(0, main(["approve", "--name", "qwen", "--uid", "1001", "--allow-interface-discovery", "--", "/opt/llama-cli", "-m", "/models/qwen.gguf"])) self.assertTrue(request.call_args.kwargs["allow_interface_discovery"]) def test_start_includes_resource_limits(self) -> None: with patch("lumi_eggcracker.cli.request", return_value={"result": "STARTED"}) as request: self.assertEqual(0, main(["start", "--name", "demo", "--max-pids", "8", "--", "/bin/sleep", "1"])) request.assert_called_once_with( "start", name="demo", max_pids=8, max_memory_mib=2048, cpu_quota_percent=400, argv=["/bin/sleep", "1"], ) def test_start_forwards_selected_execution_policy(self) -> None: with patch("lumi_eggcracker.cli.request", return_value={"result": "STARTED"}) as request: self.assertEqual(0, main(["start", "--name", "demo", "--exec-policy", "a" * 24, "--max-pids", "8", "--", "/bin/sleep", "1"])) request.assert_called_once_with( "start", name="demo", max_pids=8, max_memory_mib=2048, cpu_quota_percent=400, argv=["/bin/sleep", "1"], exec_policy="a" * 24, ) def test_start_can_require_exact_approval_at_admission_time(self) -> None: with patch("lumi_eggcracker.cli.request", return_value={"result": "STARTED"}) as request: self.assertEqual( 0, main( [ "start", "--name", "demo", "--max-pids", "8", "--require-approval", "--", "/bin/sleep", "1", ] ), ) request.assert_called_once_with( "start", name="demo", max_pids=8, max_memory_mib=2048, cpu_quota_percent=400, argv=["/bin/sleep", "1"], require_approval=True, ) def test_check_start_forwards_exact_tuple_and_emits_fixed_projection(self) -> None: response = { "next_action": ( "run start --require-approval immediately; admission is revalidated" ), "reason_code": "EXACT_APPROVAL_MATCHED", "result": "READY_TO_ATTEMPT", "schema": "lumi-eggcracker.check-start.v1", } output = io.StringIO() with patch("lumi_eggcracker.cli.request", return_value=response) as request, redirect_stdout(output): self.assertEqual( 0, main( [ "check-start", "--name", "demo", "--max-pids", "8", "--max-memory-mib", "4096", "--cpu-quota-percent", "700", "--exec-policy", "a" * 24, "--", "/bin/sleep", "1", ] ), ) request.assert_called_once_with( "check_start", name="demo", max_pids=8, max_memory_mib=4096, cpu_quota_percent=700, argv=["/bin/sleep", "1"], exec_policy="a" * 24, ) self.assertEqual(response, json.loads(output.getvalue())) def test_check_start_malformed_or_transport_response_is_constant_safe_error(self) -> None: cases = ( {"result": "READY_TO_ATTEMPT", "name": "/private/name"}, { "next_action": "raw /private/path", "reason_code": "EXACT_APPROVAL_MATCHED", "result": "READY_TO_ATTEMPT", "schema": "lumi-eggcracker.check-start.v1", }, ) for response in (*cases, OSError("/private/socket")): with self.subTest(response=response): output = io.StringIO() error = io.StringIO() with patch( "lumi_eggcracker.cli.request", side_effect=response if isinstance(response, BaseException) else None, return_value=None if isinstance(response, BaseException) else response, ) as request, redirect_stdout(output), redirect_stderr(error): self.assertEqual( 4, main( [ "check-start", "--name", "demo", "--max-pids", "8", "--", "/private/executable", ] ), ) self.assertEqual("", output.getvalue()) self.assertEqual( "eggcracker: check-start response unavailable\n", error.getvalue(), ) self.assertNotIn("/private", error.getvalue()) request.assert_called_once() def test_check_start_parse_failure_is_constant_and_excludes_canary(self) -> None: output = io.StringIO() error = io.StringIO() with patch("lumi_eggcracker.cli.request") as request, redirect_stdout(output), redirect_stderr(error): self.assertEqual( 4, main( [ "check-start", "--name", "demo", "--max-pids", "/private/parse-canary", "--", "/private/executable-canary", ] ), ) self.assertEqual("", output.getvalue()) self.assertEqual( "eggcracker: check-start response unavailable\n", error.getvalue(), ) self.assertNotIn("/private", error.getvalue()) request.assert_not_called() def test_check_start_not_ready_is_a_nonzero_snapshot_result(self) -> None: response = { "next_action": ( "create or restore the exact root approval for this tuple, then check again" ), "reason_code": "APPROVAL_MISSING", "result": "NOT_READY", "schema": "lumi-eggcracker.check-start.v1", } output = io.StringIO() with patch("lumi_eggcracker.cli.request", return_value=response), redirect_stdout(output): self.assertEqual( 6, main( [ "check-start", "--name", "demo", "--max-pids", "8", "--", "/bin/sleep", "1", ] ), ) self.assertEqual(response, json.loads(output.getvalue())) def test_wait_immediate_terminal_emits_fixed_marker_and_validated_state(self) -> None: response = wait_status(state="TERMINATED") output = io.StringIO() error = io.StringIO() with patch("lumi_eggcracker.cli.time.monotonic", return_value=10.0), patch( "lumi_eggcracker.cli.request_before", return_value=response ) as request_before, patch("lumi_eggcracker.cli.time.sleep") as sleep, redirect_stdout( output ), redirect_stderr(error): self.assertEqual(0, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) self.assertEqual("TERMINAL\n", output.getvalue().splitlines()[0] + "\n") self.assertEqual(response, json.loads(output.getvalue().splitlines()[1])) self.assertEqual("", error.getvalue()) request_before.assert_called_once_with( "status", deadline_monotonic=11.0, name="demo" ) sleep.assert_not_called() def test_wait_active_to_terminal_pins_run_id_and_uses_bounded_sleep(self) -> None: active = wait_status() terminal = wait_status(state="CONTAINMENT_FAILED") output = io.StringIO() with patch( "lumi_eggcracker.cli.time.monotonic", side_effect=[10.0, 10.0, 10.1, 10.1, 10.2, 10.2], ), patch( "lumi_eggcracker.cli.request_before", side_effect=[active, terminal] ) as request_before, patch("lumi_eggcracker.cli.time.sleep") as sleep, redirect_stdout( output ): self.assertEqual(0, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) self.assertEqual("TERMINAL", output.getvalue().splitlines()[0]) self.assertEqual(terminal, json.loads(output.getvalue().splitlines()[1])) self.assertEqual(2, request_before.call_count) self.assertEqual([((0.1,), {})], sleep.call_args_list) for call in request_before.call_args_list: self.assertEqual("status", call.args[0]) self.assertEqual(11.0, call.kwargs["deadline_monotonic"]) self.assertEqual("demo", call.kwargs["name"]) def test_wait_timeout_emits_last_validated_state_without_postdeadline_request(self) -> None: active = wait_status() output = io.StringIO() with patch( "lumi_eggcracker.cli.time.monotonic", side_effect=[0.0, 0.0, 0.1, 0.1, 1.0], ), patch("lumi_eggcracker.cli.request_before", return_value=active) as request_before, patch( "lumi_eggcracker.cli.time.sleep" ) as sleep, redirect_stdout(output): self.assertEqual(6, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) self.assertEqual("TIMED_OUT", output.getvalue().splitlines()[0]) self.assertEqual(active, json.loads(output.getvalue().splitlines()[1])) request_before.assert_called_once() sleep.assert_called_once_with(0.1) def test_wait_inflight_deadline_is_timeout_without_inventing_state(self) -> None: output = io.StringIO() with patch("lumi_eggcracker.cli.time.monotonic", return_value=10.0), patch( "lumi_eggcracker.cli.request_before", side_effect=TimeoutError ) as request_before, redirect_stdout(output), redirect_stderr(io.StringIO()): self.assertEqual(6, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) self.assertEqual("TIMED_OUT\n", output.getvalue()) request_before.assert_called_once() def test_wait_fails_closed_for_drift_malformed_unknown_and_nonmonotonic_results(self) -> None: cases = ( ({**wait_status(), "extra": True}, "extra field"), ({**wait_status(), "state": "UNKNOWN"}, "unknown state"), ({**wait_status(), "name": "other"}, "name drift"), ) for response, label in cases: with self.subTest(label=label): output = io.StringIO() error = io.StringIO() with patch("lumi_eggcracker.cli.time.monotonic", return_value=10.0), patch( "lumi_eggcracker.cli.request_before", return_value=response ), redirect_stdout(output), redirect_stderr(error): self.assertEqual( 4, main(["wait", "--name", "demo", "--timeout-seconds", "1"]) ) self.assertEqual("", output.getvalue()) self.assertEqual("ERROR\n", error.getvalue()) first = wait_status() second = wait_status(state="TERMINATED", run_id="b" * 24) output = io.StringIO() error = io.StringIO() with patch( "lumi_eggcracker.cli.time.monotonic", side_effect=[10.0, 10.0, 10.0, 10.0, 10.1, 10.1], ), patch( "lumi_eggcracker.cli.request_before", side_effect=[first, second] ) as request_before, patch("lumi_eggcracker.cli.time.sleep"), redirect_stdout( output ), redirect_stderr(error): self.assertEqual(4, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) self.assertEqual(2, request_before.call_count) self.assertEqual("", output.getvalue()) self.assertEqual("ERROR\n", error.getvalue()) output = io.StringIO() error = io.StringIO() with patch( "lumi_eggcracker.cli.time.monotonic", side_effect=[10.0, 9.0] ), patch("lumi_eggcracker.cli.request_before") as request_before, redirect_stdout( output ), redirect_stderr(error): self.assertEqual(4, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) request_before.assert_not_called() self.assertEqual("", output.getvalue()) self.assertEqual("ERROR\n", error.getvalue()) def test_wait_timeout_argument_is_bounded_and_does_not_request(self) -> None: for timeout in ("0", "301", "1.0", "true"): with self.subTest(timeout=timeout), patch( "lumi_eggcracker.cli.request_before" ) as request_before, redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as raised: main(["wait", "--name", "demo", "--timeout-seconds", timeout]) self.assertEqual(2, raised.exception.code) request_before.assert_not_called() with patch("lumi_eggcracker.cli.request_before") as request_before, redirect_stderr( io.StringIO() ): with self.assertRaises(SystemExit) as raised: main( [ "wait", "--name", "demo", "--timeout-seconds", "1", "--timeout-seconds", "2", ] ) self.assertEqual(2, raised.exception.code) request_before.assert_not_called() def test_wait_uses_only_status_action_and_existing_status_still_uses_request(self) -> None: terminal = wait_status(state="COMPLETED_ALLOWED") with patch("lumi_eggcracker.cli.time.monotonic", return_value=10.0), patch( "lumi_eggcracker.cli.request_before", return_value=terminal ) as request_before, patch( "lumi_eggcracker.cli.request", return_value={"state": "RUNNING"} ) as request: with redirect_stdout(io.StringIO()): self.assertEqual(0, main(["wait", "--name", "demo", "--timeout-seconds", "1"])) with redirect_stdout(io.StringIO()): self.assertEqual(0, main(["status", "--name", "demo"])) request_before.assert_called_once() self.assertEqual("status", request_before.call_args.args[0]) request.assert_called_once_with("status", name="demo") def test_execution_policy_create_requires_root(self) -> None: with patch("lumi_eggcracker.cli.os.geteuid", return_value=1001, create=True): self.assertEqual(4, main(["exec-policy", "create", "--name", "demo", "--", "/bin/sh"])) def test_duplicate_identifier_options_are_rejected_before_request(self) -> None: policy_a = "a" * 24 policy_b = "b" * 24 cases = ( ["start", "--name", "first", "--name", "second", "--max-pids", "8", "--", "/bin/true"], ["start", "--name", "demo", "--exec-policy", policy_a, "--exec-policy", policy_b, "--max-pids", "8", "--", "/bin/true"], ["kill", "--name", "first", "--name", "second", "--receipt", "/tmp/receipt.json"], ["status", "--name", "first", "--name", "second"], ["approve", "--name", "first", "--name", "second", "--uid", "1001", "--", "/bin/true"], ["revoke", "--name", "first", "--name", "second"], ["exec-policy", "create", "--name", "first", "--name", "second", "--", "/bin/true"], ["exec-policy", "revoke", "--policy-id", policy_a, "--policy-id", policy_b], ) for values in cases: with self.subTest(values=values), patch( "lumi_eggcracker.cli.request" ) as request, redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as raised: main(values) self.assertEqual(2, raised.exception.code) request.assert_not_called()