import base64 import importlib.util import http.client import io import json import os from pathlib import Path import tempfile import threading import unittest from unittest.mock import MagicMock, call, mock_open, patch spec = importlib.util.spec_from_file_location("ec_agent", Path(__file__).resolve().parents[1] / "ec_agent.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) Agent, AgentError, Held, Journal = module.Agent, module.AgentError, module.Held, module.Journal INSTANCE = "a53d99da-a1eb-4f01-9b10-a6268b556542" class FakeDocker: def __init__(self): self.containers = {} self.calls = [] self.fail_restart = False self.lose_reply_after = None self.started = 0 self.image_id = "sha256:" + "1" * 64 def inspect(self, name): return self.containers.get(name) def call(self, *args, **kwargs): self.calls.append(args) if args[0] == "info": return json.dumps({"ServerVersion": "27.5.1"}) if args[:2] == ("image", "ls"): return "ghcr.io/ayymoss/plutainer:latest\n" if args[:4] == ("image", "inspect", "--format", "{{.Id}}"): return self.image_id if args[:2] == ("container", "create"): labels = {args[index + 1].split("=", 1)[0]: args[index + 1].split("=", 1)[1] for index, arg in enumerate(args) if arg == "--label"} name = args[args.index("--name") + 1] self.containers[name] = {"Id": "container-123", "Config": {"Labels": labels}, "State": {"Status": "created", "Running": False}} if args[:2] in (("container", "start"), ("container", "restart")): if self.fail_restart and args[1] == "restart": raise AgentError("Docker command unavailable or timed out") self.started += 1 self.containers[args[-1]]["State"] = {"Status": "running", "Running": True, "StartedAt": f"started-{self.started}"} if args[:2] == ("container", "stop"): self.containers[args[-1]]["State"] = {"Status": "exited", "Running": False} if args[:2] == ("container", "rm"): self.containers.pop(args[-1]) if args[0] == "container" and args[1] == self.lose_reply_after: raise AgentError("Docker command unavailable or timed out") return "" class Response(io.BytesIO): def __init__(self, content, status=200, headers=None): super().__init__(content) self.status = status self.headers = headers or {} class JournalRetentionTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.path = Path(self.temp.name) / "operations.sqlite3" self.journal = Journal(self.path) self.now = 2000000 self.sequence = 0 def tearDown(self): self.journal.close() self.temp.cleanup() def record(self, kind="files.stat", payload=None, result=None, state="completed", error=None, at=None, confirmed=True): self.sequence += 1 operation = {"id": f"operation-{self.sequence}", "kind": kind, "payload": payload or {}} self.journal.accept(operation) with patch.object(module.time, "time", return_value=self.now - 172800 if at is None else at): row = self.journal.update(operation["id"], state, result, error) if confirmed: self.journal.confirm_completed(row) return row def prune(self, days=1): with patch.object(module.time, "time", return_value=self.now): return self.journal.prune_observations(days) def test_cutoff_requires_an_older_confirmed_success(self): old = self.record(at=self.now - 86401) boundary = self.record(at=self.now - 86400) recent = self.record(at=self.now - 10) unconfirmed = self.record(confirmed=False) self.assertEqual(self.prune(), 1) with self.assertRaisesRegex(AgentError, "Unknown operation"): self.journal.get(old["id"]) for row in (boundary, recent, unconfirmed): self.assertEqual(self.journal.get(row["id"])["state"], "completed") def test_zero_and_invalid_retention_preserve_every_record(self): row = self.record() for days in (0, -1, 36501, True, False, None, "1", 1.0, 1.5): with self.subTest(days=days): self.assertEqual(self.prune(days), 0) self.assertEqual(self.journal.get(row["id"])["state"], "completed") def test_all_read_kinds_are_eligible_but_mutations_keep_their_replay_journal(self): for kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"): self.record(kind, result={"healthy": True}) for command in ("status", "getstatus", "version"): self.record("rcon.execute", {"command": command}, {"ok": True}) mutations = [self.record(kind) for kind in ("instance.restart", "instance.stop", "instance.create", "files.stage", "files.activate", "credentials.rotate")] mutations.append(self.record("rcon.execute", {"command": "map_restart"}, {"ok": True})) self.assertEqual(self.prune(), 10) for row in mutations: saved, accepted = self.journal.accept(json.loads(row["request"])) self.assertFalse(accepted) self.assertEqual(saved["id"], row["id"]) def test_completed_batch_with_failed_targets_is_pruned_after_acknowledgment(self): result = {"ok": True, "results": [{"instanceId": INSTANCE, "generation": 1, "reply": {"ok": False, "reason": "timeout", "text": ""}}]} self.record("rcon.status-batch", {"instances": [{"instanceId": INSTANCE, "generation": 1}]}, result) self.assertEqual(self.prune(), 1) def test_errors_failures_and_unfinished_operations_are_retained(self): retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})] retained.append(self.record(error="read did not finish")) retained.extend(self.record(state=state) for state in ("accepted", "running", "unknown", "failed")) self.assertEqual(self.prune(), 0) for row in retained: self.assertEqual(self.journal.get(row["id"])["state"], row["state"]) def test_old_receipt_cannot_confirm_a_changed_or_reexecuted_result(self): old = self.record(confirmed=False) with patch.object(module.time, "time", return_value=self.now - 90000): current = self.journal.update(old["id"], "completed", {"new": "result"}) self.journal.confirm_completed(old) self.assertIsNone(self.journal.get(old["id"])["reported_at"]) self.assertEqual(self.prune(), 0) self.journal.confirm_completed(current) self.assertEqual(self.prune(), 1) def test_pruning_uses_bounded_indexed_batches(self): template = self.record() self.journal.db.executemany("""INSERT INTO operations(id,fingerprint,request,state,result,updated,reported_at,retention_eligible) VALUES(?,?,?,'completed',NULL,?,?,1)""", [ (f"batch-{index}", template["fingerprint"], template["request"], self.now - 172800, self.now - 172800) for index in range(504) ]) self.journal.db.commit() self.assertEqual(self.prune(), 500) self.assertEqual(self.journal.db.execute("SELECT COUNT(*) FROM operations").fetchone()[0], 5) self.assertEqual(self.prune(), 5) plan = self.journal.db.execute("""EXPLAIN QUERY PLAN SELECT id FROM operations WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1 AND updated 0 for frame in details["frames"])) self.assertIn("validating the container specification", result["error"]) self.assertIn("HTTP 404", result["error"]) self.assertIn("at artifact_open:", result["error"]) self.assertIn("artifact is still available", result["error"]) self.assertFalse(any(call[:2] == ("container", "create") for call in self.docker.calls)) self.assertIsNone(self.agent.operation_phase) self.agent.journal.close() self.agent = Agent(self.config, self.docker) self.assertEqual(self.agent.execute(operation), result) with patch.object(self.agent, "request", return_value={}) as report: self.agent.report(result) receipt = report.call_args.args[2] self.assertEqual(receipt["result"]["unexpectedError"], details) safe_output = json.dumps(receipt) for hidden in (secret, "controller.example", "Authorization", self.temp.name, "contentSha256", "rconPassword", "Traceback"): self.assertNotIn(hidden, safe_output) def test_unexpected_create_disk_failure_identifies_phase_before_mutation(self): payload = {"game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960} with patch.object(self.agent, "seed_plutonium", side_effect=OSError(module.errno.ENOSPC, "private-error", "/private/secret-path")): result = self.agent.execute(self.operation("instance.create", payload)) self.assertEqual(result["state"], "failed", result) self.assertIsNone(result["checkpoint"]) details = json.loads(result["result"])["unexpectedError"] self.assertEqual(details["phase"], "seeding-runtime") self.assertEqual(details["errno"], module.errno.ENOSPC) self.assertEqual(details["errnoName"], "ENOSPC") self.assertIn("seeding the runtime cache", result["error"]) self.assertIn("free disk space and inodes", result["error"]) self.assertNotIn("private", result["error"] + result["result"]) self.assertFalse(any(call[:2] == ("container", "create") for call in self.docker.calls)) def test_unexpected_create_failure_after_checkpoint_remains_unknown_without_replay(self): payload = {"game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960} docker_call = self.docker.call creates = [] def failed_create(*args, **kwargs): if args[:2] == ("container", "create"): creates.append(args) raise RuntimeError("private-runtime-data") from ValueError("private-cause") return docker_call(*args, **kwargs) operation = self.operation("instance.create", payload) with patch.object(self.docker, "call", side_effect=failed_create): result = self.agent.execute(operation) self.assertEqual(result["state"], "unknown", result) self.assertEqual(json.loads(result["checkpoint"])["phase"], "creating") details = json.loads(result["result"])["unexpectedError"] self.assertEqual(details["type"], "RuntimeError") self.assertEqual(details["phase"], "creating-container") self.assertEqual(self.agent.execute(operation), result) self.assertEqual(len(creates), 1) self.assertNotIn("private", result["error"] + result["result"]) def test_unexpected_error_guidance_uses_safe_exception_metadata(self): for error, category, hint in ( (PermissionError(module.errno.EACCES, "private", "/private/path"), "PermissionError", "permissions"), (OSError(module.errno.EDQUOT, "private"), "OSError", "quota"), (OSError(module.errno.EROFS, "private"), "OSError", "read-only"), (module.urllib.error.URLError(ConnectionRefusedError(module.errno.ECONNREFUSED, "private")), "URLError", "connectivity"), (TimeoutError("private"), "TimeoutError", "connectivity"), (http.client.IncompleteRead(b"private", 100), "IncompleteRead", "connectivity"), (module.urllib.error.HTTPError("https://private", 403, "private", {}, None), "HTTPError", "enrollment"), (module.urllib.error.HTTPError("https://private", 503, "private", {}, None), "HTTPError", "controller availability"), ): with self.subTest(category=category, hint=hint): message, details = module.unexpected_operation_error(error, "executing") self.assertEqual(details["type"], category) self.assertIn(hint, message) self.assertNotIn("private", message + json.dumps(details)) def test_unexpected_error_diagnostics_handle_unfamiliar_attributes_without_rendering_them(self): class UnsafeValue: def __str__(self): raise AssertionError("private attribute was rendered") def __eq__(self, other): raise AssertionError("private attribute was compared") class UnavailableTraceback(Exception): def __getattribute__(self, name): if name == "__traceback__": raise RuntimeError("private traceback access") return super().__getattribute__(name) filesystem_error = OSError("private error") filesystem_error.errno = UnsafeValue() errors = [filesystem_error, UnavailableTraceback("private exception"), type("private class", (Exception,), {})("private args"), module.urllib.error.URLError(UnsafeValue()), module.urllib.error.HTTPError("https://private", UnsafeValue(), "private reason", {}, None)] for error in errors: error.__cause__ = RuntimeError("private cause") message, details = module.unexpected_operation_error(error, "private phase") self.assertEqual(details["phase"], "executing") self.assertNotIn("httpStatus", details) self.assertNotIn("errno", details) self.assertNotIn("private", message + json.dumps(details)) def test_read_and_health_lanes_retry_unexpected_failures_without_logging_secrets(self): for lane in ("read", "health"): with self.subTest(lane=lane): self.agent.stopping.clear() polls = [] def request(*args): polls.append(args) if len(polls) == 1: raise RuntimeError("secret controller response") self.agent.stopping.set() return {"operations": []} with patch.object(self.agent, "request", side_effect=request), patch.object(self.agent.stopping, "wait", return_value=False), patch.object(module.sys, "stderr", new_callable=io.StringIO) as logs: self.agent.read_loop(lane) self.assertEqual(len(polls), 2) self.assertIn("RuntimeError", logs.getvalue()) self.assertNotIn("secret", logs.getvalue()) def test_heartbeat_loop_survives_unexpected_failures(self): with patch.object(self.agent.stopping, "wait", side_effect=[False, False, True]), patch.object(self.agent, "heartbeat", side_effect=[RuntimeError("secret"), None]) as heartbeat, patch.object(module.sys, "stderr", new_callable=io.StringIO) as logs: self.agent.heartbeat_loop() self.assertEqual(heartbeat.call_count, 2) self.assertIn("RuntimeError", logs.getvalue()) self.assertNotIn("secret", logs.getvalue()) def test_startup_and_write_loop_survive_unexpected_failures(self): polls = [] def poll(): polls.append(True) if len(polls) == 1: raise RuntimeError("secret") self.agent.stopping.set() with patch.object(module.threading, "Thread"), patch.object(self.agent.stopping, "wait", return_value=False), patch.object(self.agent, "heartbeat", side_effect=[RuntimeError("secret"), None]) as heartbeat, patch.object(self.agent, "poll_operations", side_effect=poll), patch.object(module.sys, "stderr", new_callable=io.StringIO) as logs: self.agent.run() self.assertEqual(heartbeat.call_count, 2) self.assertEqual(len(polls), 2) self.assertEqual(logs.getvalue().count("RuntimeError"), 2) self.assertNotIn("secret", logs.getvalue()) def test_lane_inventory_tracks_each_health_worker_and_detects_dead_threads(self): def poll(): self.agent.lane_started("health-1") self.agent.lane_polled("health-1") with patch.object(module.time, "monotonic", return_value=100): thread = threading.Thread(target=poll) thread.start() thread.join(timeout=2) self.assertFalse(thread.is_alive()) with patch.object(module.time, "monotonic", return_value=150): self.agent.lane_started("health-2") self.agent.lane_polled("health-2") self.agent.lane_busy("health-2", True) with patch.object(module.time, "monotonic", return_value=200): lanes = {lane["name"]: lane for lane in self.agent.lane_inventory()} self.assertEqual(set(lanes), {"health-1", "health-2", "read", "write"}) self.assertFalse(lanes["health-1"]["alive"]) self.assertEqual(lanes["health-1"]["pollAgeSeconds"], 100) self.assertTrue(lanes["health-2"]["alive"]) self.assertTrue(lanes["health-2"]["busy"]) self.assertEqual(lanes["health-2"]["pollAgeSeconds"], 50) self.assertIsNotNone(lanes["health-2"]["lastPollAt"]) self.assertEqual(len(self.agent.inventory()["agentLanes"]), 4) def test_health_lane_times_out_on_instance_lock_and_processes_next_check(self): blocked = self.operation("instance.inspect", op_id="blocked-health") fresh = {**self.operation("instance.inspect", op_id="fresh-health"), "instanceId": "other-instance"} pending = iter([blocked, fresh]) failures = [] def request(method, path, payload=None): if method == "POST": return {"operation": {"id": path.rsplit("/", 1)[1], "state": payload["state"]}} operation = next(pending, None) if operation is None: self.agent.stopping.set() return {"operations": [operation] if operation else []} def read(): try: self.agent.read_loop("health") except BaseException as error: failures.append(error) with patch.object(module, "HEALTH_LOCK_TIMEOUT_SECONDS", 0.01), patch.object(self.agent, "request", side_effect=request), patch.object(self.agent, "dispatch", return_value={"healthy": True}) as dispatch: with self.agent.instance_lock(blocked): thread = threading.Thread(target=read) thread.start() thread.join(timeout=2) finished_while_locked = not thread.is_alive() self.agent.stopping.set() thread.join(timeout=2) self.assertTrue(finished_while_locked) self.assertEqual(failures, []) self.assertEqual(self.agent.journal.get(blocked["id"])["state"], "failed") self.assertIn("waiting for the instance", self.agent.journal.get(blocked["id"])["error"]) self.assertEqual(self.agent.journal.get(fresh["id"])["state"], "completed") dispatch.assert_called_once_with(fresh) self.assertEqual(self.agent.read_claims, set()) def test_inventory_reuses_manifests_until_mtime_changes(self): marker = self.library / "manifest.json" saved = json.loads(marker.read_text()) original_read = Path.read_text reads = [] def read(path, *args, **kwargs): if path == marker: reads.append(path) return original_read(path, *args, **kwargs) with patch.object(Path, "read_text", read): self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9) self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 9) self.assertEqual(len(reads), 1) metadata = marker.stat() marker.write_text(json.dumps({**saved, "size": 8})) module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns + 1_000_000_000)) self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8) self.assertEqual(len(reads), 2) def test_inventory_reloads_atomic_replacement_with_same_size_and_mtime(self): marker = self.library / "manifest.json" saved = json.loads(marker.read_text()) self.agent.inventory() metadata = marker.stat() module.atomic_json(marker, {**saved, "size": 8}) module.os.utime(marker, ns=(metadata.st_atime_ns, metadata.st_mtime_ns)) self.assertEqual(marker.stat().st_size, metadata.st_size) self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8) def test_inventory_discards_invalid_and_removed_manifests(self): marker = self.library / "manifest.json" saved = json.loads(marker.read_text()) self.agent.inventory() marker.write_text("invalid JSON") self.assertEqual(self.agent.inventory()["libraryCaches"], []) self.assertNotIn(marker, self.agent.inventory_manifest_cache) module.atomic_json(marker, saved) self.assertEqual(len(self.agent.inventory()["libraryCaches"]), 1) marker.unlink() self.assertEqual(self.agent.inventory()["libraryCaches"], []) self.assertNotIn(marker, self.agent.inventory_manifest_cache) module.atomic_json(marker, {**saved, "size": 8}) self.assertEqual(self.agent.inventory()["libraryCaches"][0]["bytesTotal"], 8) def test_cached_inventory_preserves_transfer_progress_and_integrity_checks(self): self.agent.inventory() progress = {"libraryVersionId": self.library_id, "status": "verifying", "bytesVerified": 3} module.atomic_json(self.library / "transfer.json", progress) self.assertEqual(self.agent.inventory()["libraryCaches"][0]["status"], "missing") module.atomic_json(self.library / "transfer.json", {**progress, "status": "failed", "bytesVerified": 5}) cache = self.agent.inventory()["libraryCaches"][0] self.assertEqual(cache["status"], "failed") self.assertEqual(cache["bytesVerified"], 5) (self.library / "content/game.ff").write_bytes(b"bad-data!") with self.assertRaises(AgentError): self.agent.verify_library(self.library) (self.library / "manifest.json").write_text("invalid JSON") with self.assertRaises(ValueError): self.agent.verify_library(self.library) def test_library_remove_preserves_other_versions_and_is_idempotent(self): other = self.agent.library_dir("t5", "other-library") other.mkdir() staging = self.library.with_name("." + self.library_id + ".partial") staging.mkdir() (staging / "partial").write_bytes(b"partial") payload = {"libraryVersionId": self.library_id, "game": "t5"} self.assertTrue(self.agent.library_remove(payload)["removed"]) self.assertFalse(self.library.exists()) self.assertFalse(staging.exists()) self.assertTrue(other.exists()) self.assertTrue(self.agent.library_remove(payload)["removed"]) def test_library_remove_refuses_stopped_container_mounts_and_docker_errors(self): payload = {"libraryVersionId": self.library_id, "game": "t5"} with patch.object(self.docker, "call", return_value="stopped-container"), patch.object(self.docker, "inspect", return_value={ "State": {"Running": False}, "Mounts": [{"Source": str(self.library / "content")}] }): with self.assertRaisesRegex(AgentError, "still mounted"): self.agent.library_remove(payload) with patch.object(self.docker, "call", side_effect=AgentError("Docker unavailable")): with self.assertRaises(AgentError): self.agent.library_remove(payload) self.assertTrue((self.library / "content" / "game.ff").exists()) def test_library_remove_recovers_after_interrupted_deletion(self): payload = {"libraryVersionId": self.library_id, "game": "t5"} operation = {"id": "library-cleanup", "kind": "library.remove", "payload": payload} self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") self.agent.journal.checkpoint(operation["id"], {"phase": "removing-library"}) (self.library / "manifest.json").unlink() row = self.agent.reconcile(operation, self.agent.journal.get(operation["id"])) self.assertEqual(row["state"], "completed") self.assertFalse(self.library.exists()) def test_library_remove_ignores_unrelated_and_tmpfs_mounts(self): with patch.object(self.docker, "call", return_value="other-container"), patch.object(self.docker, "inspect", return_value={ "Mounts": [{"Type": "tmpfs", "Source": ""}, {"Source": str(self.agent.root / "instances")}] }): self.agent.library_remove({"libraryVersionId": self.library_id, "game": "t5"}) self.assertFalse(self.library.exists()) def test_library_remove_refuses_symlink_storage(self): linked = self.agent.library_dir("t5", "linked-library") linked.symlink_to(self.library, target_is_directory=True) with self.assertRaisesRegex(AgentError, "symlink"): self.agent.library_remove({"libraryVersionId": "linked-library", "game": "t5"}) self.assertTrue(self.library.exists()) def cpu_sample(self, line): with patch.object(Path, "open", mock_open(read_data=line)): return self.agent.cpu_usage_percent() def test_cpu_usage_uses_interval_and_excludes_guest_double_counting(self): self.assertIsNone(self.cpu_sample("cpu 100 0 50 800 50 0 0 0 20 0\n")) # 30 busy ticks (including steal), 60 idle, 10 I/O wait = 30%. self.assertEqual(self.cpu_sample("cpu 120 0 55 860 60 0 0 5 30 0\n"), 30.0) self.assertEqual(self.cpu_sample("cpu 120 0 55 960 60 0 0 5 30 0\n"), 0.0) self.assertEqual(self.cpu_sample("cpu 220 0 55 960 60 0 0 5 30 0\n"), 100.0) def test_cpu_usage_rebases_after_reset_or_invalid_sample(self): baseline = "cpu 100 0 50 800 50 0 0 0\n" self.assertIsNone(self.cpu_sample(baseline)) self.assertIsNone(self.cpu_sample(baseline)) self.assertIsNone(self.cpu_sample("cpu 10 0 5 80 5 0 0 0\n")) self.assertIsNone(self.cpu_sample("cpu 20 0 5 90 4 0 0 0\n")) for invalid in ("", "cpu0 1 2 3 4 5 6 7 8", "cpu 1 2", "cpu x 0 0 0 0 0 0 0", "cpu -1 0 0 0 0 0 0 0"): self.assertIsNone(self.cpu_sample(invalid)) self.assertIsNone(self.cpu_sample(baseline)) with patch.object(Path, "open", side_effect=OSError("unavailable")): self.assertIsNone(self.agent.cpu_usage_percent()) self.assertIsNone(self.cpu_sample(baseline)) def test_inventory_reports_cpu_only_when_available(self): for usage in (None, 0.0, 42.3, 100.0): with patch.object(self.agent, "cpu_usage_percent", return_value=usage): inventory = self.agent.inventory() if usage is None: self.assertNotIn("cpuUsagePercent", inventory) else: self.assertEqual(inventory["cpuUsagePercent"], usage) def resource_container(self): return {"Name": self.agent.container_name(INSTANCE, 1), "Config": {"Labels": {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE, module.LABEL + ".generation": "1"}}, "State": {"Running": True, "Status": "running", "StartedAt": "2000-01-01T00:00:00.000000000Z"}, "Mounts": [{"Type": "bind", "Source": str(self.agent.instance_dir(INSTANCE, 1) / "app"), "Destination": "/home/plutainer/app"}]} def test_instance_resources_batches_owned_containers_and_normalizes_core_cpu_and_memory(self): name = self.agent.container_name(INSTANCE, 1) rows = json.dumps({"Name": name, "CPUPerc": "150.00%", "MemUsage": "384.5MiB / 8GiB", "PIDs": "12"}) with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([self.resource_container()]), rows]) as call: samples = self.agent.instance_resources() self.assertEqual(len(samples), 1) self.assertEqual(samples[0]["instanceId"], INSTANCE) self.assertEqual(samples[0]["generation"], 1) self.assertEqual(samples[0]["cpuCores"], 1.5) self.assertEqual(samples[0]["memoryBytes"], int(384.5 * 1024 ** 2)) self.assertGreater(samples[0]["uptimeSeconds"], 300) self.assertEqual(call.call_args_list[0].args, ("container", "ls", "--all", "--filter", "label=io.electric-cherry.managed=true", "--no-trunc", "--format", "{{.ID}}")) self.assertEqual(call.call_args_list[-1].args, ("stats", "--no-stream", "--no-trunc", "--format", "{{json .}}", name)) def test_instance_resources_ignores_stopped_foreign_and_mismatched_containers(self): for change in ("stopped", "foreign", "generation", "name"): info = self.resource_container() if change == "stopped": info["State"]["Running"] = False elif change == "foreign": info["Mounts"][0]["Source"] = "/another-installation/app" elif change == "generation": info["Config"]["Labels"][module.LABEL + ".generation"] = "2" else: info["Name"] = "unrelated" with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([info])]) as call: self.assertEqual(self.agent.instance_resources(), []) self.assertEqual(call.call_count, 2) def test_instance_resources_skips_unavailable_stats_instead_of_reporting_zero_usage(self): good = {"Name": self.agent.container_name(INSTANCE, 1), "CPUPerc": "0.00%", "MemUsage": "200MiB / 8GiB", "PIDs": "10"} for override in ({"CPUPerc": "NaN%"}, {"CPUPerc": "-1%"}, {"CPUPerc": "--"}, {"MemUsage": "bad"}, {"MemUsage": "0B / 0B"}, {"PIDs": "0"}, {"Name": "foreign"}): with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([self.resource_container()]), json.dumps({**good, **override})]): self.assertEqual(self.agent.instance_resources(), []) with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([self.resource_container()]), json.dumps(good)]): self.assertEqual(self.agent.instance_resources()[0]["cpuCores"], 0) with patch.object(self.agent, "instance_resources", side_effect=AgentError("timed out")): inventory = self.agent.inventory() self.assertNotIn("instanceResources", inventory) self.assertIn("sampling failed", inventory["instanceResourcesError"]) with patch.object(self.agent, "instance_resources", return_value=[]): inventory = self.agent.inventory() self.assertEqual(inventory["instanceResources"], []) self.assertNotIn("instanceResourcesError", inventory) def test_heartbeat_states_share_inspection_and_include_stopped_and_unhealthy_containers(self): for status, running, health in (("exited", False, "healthy"), ("running", True, "unhealthy"), ("running", True, "healthy"), ("restarting", True, "healthy"), ("paused", True, "healthy")): info = self.resource_container() info["State"].update({"Status": status, "Running": running, "Health": {"Status": health}}) observations = [] outputs = ["container-id\n", json.dumps([info])] + ([""] if running else []) with patch.object(self.docker, "call", side_effect=outputs) as docker: self.agent.instance_resources(observations) self.assertEqual(observations[0]["instanceId"], INSTANCE) self.assertEqual(observations[0]["generation"], 1) self.assertEqual(observations[0]["state"], health if running and health != "healthy" else status) self.assertEqual(observations[0]["healthy"], status == "running" and running and health == "healthy") self.docker.containers[self.agent.container_name(INSTANCE, 1)] = info self.assertEqual(self.agent.inspect(INSTANCE, 1)["healthy"], observations[0]["healthy"]) self.assertIn("observedAt", observations[0]) self.assertEqual(sum(args.args[:2] == ("container", "inspect") for args in docker.call_args_list), 1) info["Mounts"][0]["Source"] = "/foreign/app" observations = [] with patch.object(self.docker, "call", side_effect=["container-id", json.dumps([info])]): self.agent.instance_resources(observations) self.assertEqual(observations, []) def test_status_batch_limits_concurrency_and_isolates_target_failures(self): operation = {"kind": "rcon.status-batch", "payload": {"instances": [ {"instanceId": f"00000000-0000-4000-8000-{i:012d}", "generation": 1} for i in range(5)]}} self.assertTrue(module.read_operation(operation)) self.assertTrue(self.agent.health_operation(operation)) ready, release = threading.Event(), threading.Event() lock = threading.Lock() active, peak = 0, 0 def rcon(instance_id, generation, command, deadline, status_attempts): nonlocal active, peak with lock: active += 1 peak = max(peak, active) if active == 2: ready.set() try: self.assertEqual(command, "status") self.assertEqual(status_attempts, 1) self.assertTrue(release.wait(2)) if instance_id == "00000000-0000-4000-8000-000000000002": raise AgentError("local failure") return {"ok": True, "text": "status"} finally: with lock: active -= 1 results = [] with patch.object(self.agent, "rcon", side_effect=rcon) as probe: thread = threading.Thread(target=lambda: results.append(self.agent.dispatch(operation))) thread.start() try: self.assertTrue(ready.wait(2)) self.assertEqual(probe.call_count, 2) finally: release.set() thread.join(timeout=3) self.assertFalse(thread.is_alive()) self.assertEqual(peak, 2) self.assertEqual(len(results[0]["results"]), 5) self.assertTrue(results[0]["ok"]) self.assertEqual(results[0]["results"][2]["reply"]["reason"], "unreachable") self.assertTrue(results[0]["results"][4]["reply"]["ok"]) def test_batch_uses_one_attempt_but_individual_status_still_retries(self): saved = {"rconPassword": "secret", "port": 28960} timeout = {"ok": False, "reason": "timeout", "text": ""} operation = {"kind": "rcon.status-batch", "payload": {"instances": [{"instanceId": INSTANCE, "generation": 1}]}} with patch.object(self.agent, "spec", return_value=saved), patch.object(self.agent, "rcon_attempt", return_value=timeout) as attempt, patch.object(module.time, "sleep") as sleep: self.assertEqual(self.agent.dispatch(operation)["results"][0]["reply"], timeout) attempt.assert_called_once() sleep.assert_not_called() attempt.reset_mock() self.assertEqual(self.agent.rcon(INSTANCE, 1, "status"), timeout) self.assertEqual(attempt.call_count, 3) self.assertEqual(sleep.call_count, 2) attempt.reset_mock() self.assertEqual(self.agent.rcon(INSTANCE, 1, "status", deadline=0, status_attempts=1)["reason"], "deadline") attempt.assert_not_called() def test_batch_reaches_healthy_servers_after_four_unresponsive_servers(self): clock = [0] checked = [] targets = [{"instanceId": f"00000000-0000-4000-8000-{i:012d}", "generation": 1} for i in range(16)] def attempt(family, host, port, password, command, deadline): checked.append(port) if port < 28964: clock[0] += 3 return {"ok": False, "reason": "timeout", "text": ""} return {"ok": True, "text": "status"} # Serial execution is a conservative budget test; concurrency is tested separately. with patch.object(module, "ThreadPoolExecutor") as pool, patch.object(module.time, "monotonic", side_effect=lambda: clock[0]), patch.object(self.agent, "spec", side_effect=lambda instance, generation: {"port": 28960 + int(instance[-12:]), "rconPassword": "secret"}), patch.object(self.agent, "rcon_attempt", side_effect=attempt), patch.object(module.time, "sleep") as sleep: pool.return_value.__enter__.return_value.map.side_effect = map result = self.agent.rcon_status_batch({"payload": {"instances": targets}}) self.assertEqual(len(checked), 16) self.assertEqual(sum(row["reply"]["ok"] for row in result["results"]), 12) self.assertEqual(clock[0], 12) sleep.assert_not_called() def test_status_batch_rejects_old_generations_and_bounds_expired_work(self): marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" module.atomic_json(marker, {"generation": 2}) operation = {"kind": "rcon.status-batch", "payload": {"instances": [{"instanceId": INSTANCE, "generation": 1}]}} with patch.object(self.agent, "rcon") as probe: result = self.agent.dispatch(operation) self.assertEqual(result["results"][0]["reply"]["reason"], "superseded") probe.assert_not_called() marker.unlink() with patch.object(module.time, "monotonic", side_effect=[0] + [21] * 20): result = self.agent.dispatch(operation) self.assertEqual(result["results"][0]["reply"]["reason"], "deadline") probe.assert_not_called() for targets in ([], [None], [{"instanceId": INSTANCE, "generation": 1}] * 2): with self.assertRaises(AgentError): self.agent.dispatch({"kind": "rcon.status-batch", "payload": {"instances": targets}}) def operation(self, kind, payload=None, op_id="operation-1", generation=1): return {"id": op_id, "kind": kind, "instanceId": INSTANCE, "generation": generation, "payload": payload or {}} def create(self): payload = {"name": "t5_zombie_test_1", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {"PLUTAINER_PORT": "28960"}, "rconPassword": "rcon-secret", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"original").decode()}]} result = self.agent.execute(self.operation("instance.create", payload, "create")) self.assertEqual(result["state"], "completed", result) return payload def test_hash_file_downloads_are_verified_and_staged_as_writes(self): self.create() content = b"downloaded plugin" sha = module.digest(content) file = {"id": 1, "path": "configs/server.cfg", "contentSha256": sha, "sha256": sha} with patch.object(self.agent, "artifact_open", return_value=Response(content)) as fetch: result = self.agent.stage(INSTANCE, 1, {"stageId": "hash-stage", "files": [file]}, "hash-operation") self.assertEqual(result["state"], "staged") self.assertTrue(fetch.call_args.args[1].endswith("/api/agent/blobs/" + sha)) manifest = json.loads((self.agent.instance_dir(INSTANCE, 1) / "stages/hash-stage/manifest.json").read_text()) self.assertFalse(manifest["files"][0]["delete"]) self.assertEqual(base64.b64decode(manifest["files"][0]["contentBase64"]), content) with patch.object(self.agent, "artifact_open", return_value=Response(b"wrong bytes")): with self.assertRaisesRegex(AgentError, "checksum mismatch"): self.agent.decode_file(file) with patch.object(self.agent, "artifact_open") as fetch: with self.assertRaisesRegex(AgentError, "Invalid file content hash"): self.agent.decode_file({"contentSha256": "../../secret"}) fetch.assert_not_called() def test_same_id_reports_saved_result_without_executing_again(self): self.create() operation = self.operation("instance.restart") self.agent.rcon = lambda *_: {"ok": True, "players": 0} first = self.agent.execute(operation) self.assertEqual(first["state"], "completed") second = self.agent.execute({**operation, "state": "running", "updatedAt": "later", "result": None}) self.assertEqual(second["result"], first["result"]) self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1) def test_creation_uses_cached_runtime_without_allowing_docker_to_pull(self): payload = self.create() creation = next(args for args in self.docker.calls if args[:2] == ("container", "create")) self.assertIn("--pull=never", creation) self.assertEqual(creation[-1], self.docker.image_id) self.assertEqual(self.agent.spec(INSTANCE, 1)["image"], payload["image"]) self.assertFalse(any(args[:2] == ("image", "pull") for args in self.docker.calls)) def test_image_disappearing_after_validation_cannot_be_pulled_or_started(self): original = self.docker.call def missing_image(*args, **kwargs): if args[:2] == ("container", "create"): self.assertIn("--pull=never", args) raise AgentError("No such image") return original(*args, **kwargs) payload = {"game": "t5", "libraryId": self.library_id, "image": "prepared:runtime", "port": 28960} with patch.object(self.docker, "call", side_effect=missing_image): with self.assertRaisesRegex(AgentError, "No such image"): self.agent.create(INSTANCE, 1, payload) self.assertEqual(self.docker.started, 0) self.assertFalse(any(args[:2] == ("image", "pull") for args in self.docker.calls)) def test_retargeted_prepared_image_requires_a_new_admin_pull_before_creation(self): image = "registry.example/custom-runtime:stable" payload = {"game": "t5", "libraryId": self.library_id, "image": image, "port": 28960, "rconPassword": "rcon-secret", "env": {"PLUTAINER_KEY": "server-secret"}} self.assertEqual(self.agent.execute(self.operation("runtime.pull", {"image": image}, "prepare"))["state"], "completed") prepared_id = self.docker.image_id self.docker.image_id = "sha256:" + "2" * 64 original = self.docker.call def call(*args, **kwargs): if args[:2] == ("image", "ls"): # The old image still exists, so cached-ID inventory alone # cannot prove that this mutable tag is safe to launch. return (f"registry.example/custom-runtime|stable||{self.docker.image_id}\n" f"|||{prepared_id}\n") return original(*args, **kwargs) with patch.object(self.docker, "call", side_effect=call): self.assertIn(image, self.agent.inventory()["preparedRuntimeImages"]) result = self.agent.execute(self.operation("instance.create", payload, "retargeted")) self.assertEqual(result["state"], "failed") self.assertIn("Prepared runtime image has changed", result["error"]) self.assertFalse(any(args[:2] == ("container", "create") for args in self.docker.calls)) self.assertEqual(self.docker.started, 0) self.assertFalse((self.agent.instance_dir(INSTANCE, 1) / "environment").exists()) self.assertEqual(self.agent.execute(self.operation("runtime.pull", {"image": image}, "prepare-again"))["state"], "completed") self.assertEqual(self.agent.execute(self.operation("instance.create", payload, "retry"))["state"], "completed") creation = next(args for args in self.docker.calls if args[:2] == ("container", "create")) self.assertEqual(creation[-1], self.docker.image_id) def test_retag_after_validation_cannot_replace_the_image_used_for_creation(self): image = "ghcr.io/ayymoss/plutainer:latest" self.agent.dispatch(self.operation("runtime.pull", {"image": image})) prepared_id = self.docker.image_id original = self.agent.write_initial_files def retag(*args): self.docker.image_id = "sha256:" + "2" * 64 return original(*args) with patch.object(self.agent, "write_initial_files", side_effect=retag): payload = self.create() creation = next(args for args in self.docker.calls if args[:2] == ("container", "create")) self.assertIn("--pull=never", creation) self.assertEqual(creation[-1], prepared_id) self.assertNotEqual(creation[-1], self.docker.image_id) self.assertEqual(self.agent.spec(INSTANCE, 1), payload) # Retagging must not invalidate an already created immutable instance. calls = list(self.docker.calls) self.assertTrue(self.agent.create(INSTANCE, 1, payload)["exists"]) self.assertEqual(self.docker.calls, calls) def test_rebuild_rejects_retargeted_image_before_stopping_existing_container(self): image = "ghcr.io/ayymoss/plutainer:latest" self.agent.dispatch(self.operation("runtime.pull", {"image": image})) payload = self.create() self.docker.image_id = "sha256:" + "2" * 64 self.docker.calls.clear() result = self.agent.execute(self.operation("instance.rebuild", {"nextGeneration": 2, "spec": payload}, "rebuild")) self.assertEqual(result["state"], "failed") self.assertIn("Prepared runtime image has changed", result["error"]) self.assertFalse(any(args[:2] in (("container", "stop"), ("container", "rm"), ("container", "create")) for args in self.docker.calls)) self.assertTrue(self.docker.inspect(self.agent.container_name(INSTANCE, 1))["State"]["Running"]) def test_launch_rejects_invalid_image_ids_and_preparation_records(self): payload = {"game": "t5", "libraryId": self.library_id, "image": "prepared:runtime", "port": 28960} for image_id in ("", "prepared:runtime", "sha256:short"): with self.subTest(image_id=image_id): self.docker.image_id = image_id with self.assertRaisesRegex(AgentError, "valid runtime image ID"): self.agent.create(INSTANCE, 1, payload) self.docker.image_id = "sha256:" + "1" * 64 aliases = self.agent.root / "agent/runtime-images.json" for raw in ("invalid json", "[]", "null"): with self.subTest(records=raw): aliases.write_text(raw) with self.assertRaisesRegex(AgentError, "Runtime preparation records"): self.agent.create(INSTANCE, 1, payload) self.assertFalse(any(args[:2] == ("container", "create") for args in self.docker.calls)) self.assertEqual(self.docker.started, 0) def test_manifest_is_advertised_and_reconciles_as_a_read_operation(self): self.create() operation = self.operation("files.manifest", {"roots": ["configs", "scripts"], "ignoreNames": [], "paths": ["configs/server.cfg"]}) self.assertIn("files.manifest", self.agent.inventory()["capabilities"]) self.assertTrue(self.agent.read_operation(operation)) self.assertFalse(self.agent.health_operation(operation)) self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "unknown") calls = list(self.docker.calls) result = self.agent.execute(operation) self.assertEqual(result["state"], "completed") self.assertEqual(json.loads(result["result"]), {"files": [{"path": "configs/server.cfg", "size": 8, "sha256": module.digest(b"original")}]}) self.assertEqual(self.docker.calls, calls) def test_reusing_id_with_different_payload_is_rejected(self): self.agent.execute(self.operation("host.inspect")) with self.assertRaisesRegex(AgentError, "different content"): self.agent.execute(self.operation("host.inspect", {"changed": True})) def test_agent_restart_marks_running_unknown_and_never_replays(self): operation = self.operation("instance.restart") self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") self.agent.journal.close() self.agent = Agent(self.config, self.docker) result = self.agent.execute(operation) self.assertEqual(result["state"], "unknown") self.assertEqual(self.docker.calls, []) def test_agent_restart_resumes_runtime_pull_even_when_an_older_tag_is_cached(self): image = "ghcr.io/ayymoss/plutainer:latest" operation = self.operation("runtime.pull", {"image": image}) self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") aliases = self.agent.root / "agent/runtime-images.json" module.atomic_json(aliases, {image: "sha256:older-image"}) self.agent.journal.close() self.agent = Agent(self.config, self.docker) original = self.docker.call def call(*args, **kwargs): result = original(*args, **kwargs) return "sha256:new-image" if args[:2] == ("image", "inspect") else result self.docker.call = call result = self.agent.execute(operation) self.assertEqual(result["state"], "completed") self.assertEqual(json.loads(result["result"]), {"image": image, "ready": True, "reconciled": True}) self.assertEqual(json.loads(aliases.read_text())[image], "sha256:new-image") self.agent.execute(operation) self.assertEqual(sum(args[:2] == ("image", "pull") for args in self.docker.calls), 1) def test_interrupted_runtime_pull_failure_is_reported_for_automatic_retry(self): operation = self.operation("runtime.pull", {"image": "ghcr.io/ayymoss/plutainer:latest"}) self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") self.agent.journal.close() self.agent = Agent(self.config, self.docker) with patch.object(self.docker, "call", side_effect=AgentError("Registry unavailable")): result = self.agent.execute(operation) self.assertEqual(result["state"], "failed") self.assertIn("runtime download", result["error"]) def test_occupied_and_unknown_wait_then_revalidate_same_operation(self): self.create() operation = self.operation("instance.restart") for players in (2, None, 7): self.agent.rcon = lambda *_, count=players: {"ok": True, "players": count} result = self.agent.execute(operation) self.assertEqual(result["state"], "accepted") self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.assertEqual(self.agent.execute(operation)["state"], "completed") def autoscaling_operation(self, op_id="autoscaling-remove"): return self.operation("instance.remove", {"autoscaling": {"policyId": "policy", "revision": 1, "expiresAt": 2000000}}, op_id) def test_autoscaling_occupied_or_unknown_removal_never_retries_later(self): self.create() for index, players in enumerate((2, None)): operation = self.autoscaling_operation(f"autoscaling-{index}") self.agent.rcon = lambda *_, count=players: {"ok": True, "players": count} with patch.object(self.agent, "request", side_effect=AssertionError("Occupied removal requested authorization")): result = self.agent.execute(operation) self.assertEqual(result["state"], "failed") self.assertTrue(json.loads(result["result"])["autoscalingInvalidated"]) self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.assertEqual(self.agent.execute(operation)["state"], "failed") self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) def test_autoscaling_removal_requires_fresh_one_shot_controller_authorization(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} with patch.object(self.agent, "request", return_value={"authorized": False}) as request: self.assertEqual(self.agent.execute(self.autoscaling_operation("revoked"))["state"], "failed") self.assertTrue(request.call_args.args[1].endswith("/revoked/authorize-removal")) with patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}): self.assertEqual(self.agent.execute(self.autoscaling_operation("valid"))["state"], "completed") self.assertEqual(sum(args[:2] == ("container", "rm") for args in self.docker.calls), 1) def test_autoscaling_rechecks_players_after_authorization_and_rejects_expired_claim(self): self.create() with patch.object(self.agent, "rcon", side_effect=[{"ok": True, "players": 0}, {"ok": True, "players": 1}]), patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}): self.assertEqual(self.agent.execute(self.autoscaling_operation("joined"))["state"], "failed") self.agent.rcon = lambda *_: {"ok": True, "players": 0} with patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}), patch.object(module.time, "monotonic", side_effect=[0, 6]): self.assertEqual(self.agent.execute(self.autoscaling_operation("expired"))["state"], "failed") self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) def test_autoscaling_restart_before_mutation_invalidates_instead_of_replaying(self): self.create() operation = self.autoscaling_operation() self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") self.agent.journal.close() self.agent = Agent(self.config, self.docker) with patch.object(self.agent, "request", side_effect=AssertionError("Recovered command requested execution")): result = self.agent.execute(operation) self.assertEqual(result["state"], "failed") self.assertTrue(json.loads(result["result"])["autoscalingInvalidated"]) self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) def test_autoscaling_cannot_use_admin_override(self): self.create() operation = self.autoscaling_operation() operation["payload"]["safetyOverride"] = True self.assertEqual(self.agent.execute(operation)["state"], "failed") self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) def test_occupancy_hold_explains_local_rcon_failure_without_exposing_reply(self): self.create() operation = self.operation("instance.restart") for reason, detail in ( ("unconfigured", "password is not configured"), ("unreachable", "endpoint is unreachable"), ("invalid-reply", "invalid packet prefix"), ("timeout", "query timed out"), ("authentication", "authentication failed"), (None, "no recognized player count"), ): with self.subTest(reason=reason): self.agent.rcon = lambda *_: {"ok": reason is None, "reason": reason, "players": None, "text": "sensitive raw reply"} result = self.agent.execute(operation) self.assertEqual(result["state"], "accepted") message = json.loads(result["result"])["message"] self.assertIn(detail, message) self.assertNotIn("sensitive raw reply", message) self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.assertEqual(self.agent.execute(operation)["state"], "completed") def test_unknown_container_health_cannot_disrupt(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"] = {"Status": "unknown", "Running": False} self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.assertEqual(self.agent.execute(self.operation("instance.stop"))["state"], "accepted") self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls)) def assert_unhealthy_recovery(self, kind): spec = self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} payload = {"unhealthyRecovery": True} if kind == "instance.rebuild": payload.update({**spec, "nextGeneration": 2}) operation = self.operation(kind, payload) with patch.object(self.agent, "rcon", side_effect=AssertionError("Unhealthy recovery queried player status")): result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) calls = list(self.docker.calls) self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertEqual(self.docker.calls, calls) target_generation = 2 if kind == "instance.rebuild" else 1 self.assertTrue(self.agent.inspect(INSTANCE, target_generation)["healthy"]) self.assertNotIn("unhealthyRecovery", self.agent.spec(INSTANCE, target_generation)) def test_unhealthy_recovery_restarts_without_player_status(self): self.assert_unhealthy_recovery("instance.restart") self.assertIn("instance.unhealthy-recovery", self.agent.inventory()["capabilities"]) def test_unhealthy_recovery_rebuilds_without_player_status(self): self.assert_unhealthy_recovery("instance.rebuild") def test_unhealthy_recovery_keeps_safety_for_other_health_states(self): spec = self.create() container = self.docker.containers[self.agent.container_name(INSTANCE, 1)] for kind in ("instance.restart", "instance.rebuild"): payload = {"unhealthyRecovery": True} if kind == "instance.rebuild": payload.update({"spec": spec, "nextGeneration": 2}) for state in ("healthy", "starting", "restarting", "unknown"): with self.subTest(kind=kind, state=state): container["State"] = {"Status": "unknown" if state == "unknown" else "running", "Running": state != "unknown", "Health": {"Status": state}} with patch.object(self.agent, "rcon", return_value={"ok": True, "players": 2}) as rcon: result = self.agent.execute(self.operation(kind, payload, f"{kind.split('.')[-1]}-{state}")) self.assertEqual(result["state"], "accepted", result) self.assertEqual(rcon.call_count, 0 if state == "unknown" else 1) self.assertFalse(any(call[:2] in (("container", "restart"), ("container", "stop"), ("container", "rm")) for call in self.docker.calls)) def test_unhealthy_recovery_rechecks_health_before_disruption(self): spec = self.create() container = self.docker.containers[self.agent.container_name(INSTANCE, 1)] inspect = self.docker.inspect for kind in ("instance.restart", "instance.rebuild"): container["State"]["Health"] = {"Status": "unhealthy"} observations = [] def recovered(name): observations.append(name) if len(observations) == 2: container["State"]["Health"] = {"Status": "healthy"} return inspect(name) payload = {"unhealthyRecovery": True} if kind == "instance.rebuild": payload.update({"spec": spec, "nextGeneration": 2}) with patch.object(self.docker, "inspect", side_effect=recovered), patch.object(self.agent, "rcon", return_value={"ok": True, "players": 2}) as rcon: result = self.agent.execute(self.operation(kind, payload, kind.split(".")[-1])) self.assertEqual(result["state"], "accepted", result) rcon.assert_called_once() self.assertFalse(any(call[:2] in (("container", "restart"), ("container", "stop"), ("container", "rm")) for call in self.docker.calls)) def test_unhealthy_recovery_requires_boolean_true(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} for index, value in enumerate((False, None, "true", 1, {})): with self.subTest(value=value), patch.object(self.agent, "rcon", return_value={"ok": False, "players": None, "reason": "unreachable"}) as rcon: result = self.agent.execute(self.operation("instance.restart", {"unhealthyRecovery": value}, f"attempt-{index}")) self.assertEqual(result["state"], "accepted", result) rcon.assert_called_once() self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) def test_unhealthy_recovery_rejects_other_lifecycle_actions_and_activation(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} attempts = [(kind, {}) for kind in ("instance.stop", "instance.remove", "instance.start", "instance.create")] attempts.extend([("instance.restart", {"stageId": "campaign"}), ("instance.restart", {"autoscaling": {}})]) calls = list(self.docker.calls) for index, (kind, payload) in enumerate(attempts): with self.subTest(kind=kind, payload=payload): result = self.agent.execute(self.operation(kind, {**payload, "unhealthyRecovery": True}, f"attempt-{index}")) self.assertEqual(result["state"], "failed", result) self.assertIn("Unhealthy recovery only permits", result["error"]) self.assertEqual(self.docker.calls, calls) with patch.object(self.agent, "rcon", return_value={"ok": True, "players": 2}) as rcon: result = self.agent.execute(self.operation("files.activate", {"stageId": "campaign", "unhealthyRecovery": True})) self.assertEqual(result["state"], "accepted", result) rcon.assert_called_once() def test_unhealthy_recovery_preserves_ownership_and_generation_checks(self): spec = self.create() container = self.docker.containers[self.agent.container_name(INSTANCE, 1)] container["State"]["Health"] = {"Status": "unhealthy"} labels = container["Config"]["Labels"] container["Config"]["Labels"] = {} result = self.agent.execute(self.operation("instance.restart", {"unhealthyRecovery": True}, "unowned")) self.assertEqual(result["state"], "failed", result) container["Config"]["Labels"] = labels result = self.agent.execute(self.operation("instance.rebuild", {"unhealthyRecovery": True, "spec": spec, "nextGeneration": 1}, "same-generation")) self.assertEqual(result["state"], "failed", result) result = self.agent.execute(self.operation("instance.restart", {"unhealthyRecovery": True}, "stale", generation=0)) self.assertEqual(result["state"], "failed", result) self.assertFalse(any(call[:2] in (("container", "restart"), ("container", "stop"), ("container", "rm")) for call in self.docker.calls)) def assert_override_disruption(self, kind): spec = self.create() payload = {"safetyOverride": True} if kind == "instance.rebuild": payload.update({**spec, "nextGeneration": 2}) operation = self.operation(kind, payload) with patch.object(self.agent, "rcon", side_effect=AssertionError("Override queried player status")): result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) calls = list(self.docker.calls) self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertEqual(self.docker.calls, calls) self.assertTrue((self.library / "content/game.ff").exists()) def test_admin_override_restarts_without_player_status(self): self.assert_override_disruption("instance.restart") def test_admin_override_stops_without_player_status(self): self.assert_override_disruption("instance.stop") def test_admin_override_removes_without_player_status(self): self.assert_override_disruption("instance.remove") def test_admin_override_rebuilds_without_player_status(self): self.assert_override_disruption("instance.rebuild") def test_admin_override_requires_boolean_true(self): self.create() self.assertIn("instance.safety-override", self.agent.inventory()["capabilities"]) for index, value in enumerate((False, None, "true", 1, {})): with self.subTest(value=value): self.agent.rcon = lambda *_: {"ok": True, "players": 2} result = self.agent.execute(self.operation("instance.restart", {"safetyOverride": value}, f"attempt-{index}")) self.assertEqual(result["state"], "accepted", result) self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) def test_admin_override_preserves_ownership_and_generation_checks(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {} self.assertEqual(self.agent.execute(self.operation("instance.remove", {"safetyOverride": True}))["state"], "failed") self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls)) self.assertEqual(self.agent.execute(self.operation("instance.stop", {"safetyOverride": True}, "stale", generation=0))["state"], "failed") def test_admin_override_activation_keeps_file_validation(self): self.create() target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg" staged = {"stageId": "campaign", "files": [{"id": 9, "path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "checkBefore": True, "expectedBeforeSha": module.digest(b"original")}]} self.agent.execute(self.operation("files.stage", staged, "stage")) self.agent.rcon = lambda *_: {"ok": True, "players": 3} target.write_bytes(b"changed") failed = self.agent.execute(self.operation("instance.restart", {"stageId": "campaign", "safetyOverride": True}, "changed")) self.assertEqual(failed["state"], "failed", failed) self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) target.write_bytes(b"original") result = self.agent.execute(self.operation("instance.restart", {"stageId": "campaign", "safetyOverride": True}, "confirmed")) self.assertEqual(result["state"], "completed", result) self.assertEqual(target.read_bytes(), b"new") def test_files_stage_is_nondisruptive_activation_snapshots_and_rollback(self): self.create() target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg" staged = {"stageId": "campaign", "files": [{"id": 9, "path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "sha256": module.digest(b"new"), "checkBefore": True, "expectedBeforeSha": module.digest(b"original")}]} self.agent.execute(self.operation("files.stage", staged, "stage")) self.assertEqual(target.read_bytes(), b"original") self.agent.rcon = lambda *_: {"ok": True, "players": 3} operation = self.operation("instance.restart", {"stageId": "campaign"}) self.assertEqual(self.agent.execute(operation)["state"], "accepted") self.assertEqual(target.read_bytes(), b"original") self.agent.rcon = lambda *_: {"ok": True, "players": 0} result = json.loads(self.agent.execute(operation)["result"]) self.assertEqual(base64.b64decode(result["before"][0]["contentBase64"]), b"original") self.assertEqual(target.read_bytes(), b"new") rollback = self.agent.execute(self.operation("files.rollback", {"stageId": "campaign", "restart": False}, "rollback")) self.assertEqual(rollback["state"], "completed") self.assertEqual(target.read_bytes(), b"original") def test_changed_target_fails_before_activation_or_restart(self): self.create() target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg" self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "checkBefore": True, "expectedBeforeSha": "0" * 64}]}, "stage")) self.agent.rcon = lambda *_: {"ok": True, "players": 0} result = self.agent.execute(self.operation("instance.restart", {"stageId": "campaign"})) self.assertEqual(result["state"], "failed") self.assertEqual(target.read_bytes(), b"original") self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) def test_ambiguous_restart_keeps_before_images_on_disk_and_is_not_replayed(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage")) self.docker.fail_restart = True operation = self.operation("instance.restart", {"stageId": "campaign"}) self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.agent.execute(operation) self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1) saved = json.loads((self.agent.instance_dir(INSTANCE, 1) / "stages/campaign/manifest.json").read_text()) self.assertEqual(saved["state"], "activated") self.assertEqual(base64.b64decode(saved["before"][0]["contentBase64"]), b"original") def test_remove_only_own_instance_and_preserves_shared_library(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} result = self.agent.execute(self.operation("instance.remove")) self.assertEqual(result["state"], "completed") self.assertTrue((self.library / "content/game.ff").exists()) self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists()) def test_cancel_creation_bypasses_occupancy_and_preserves_library(self): self.create() self.agent.rcon = lambda *_: self.fail("Cancellation must not wait for player evidence") operation = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel") self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists()) self.assertTrue((self.library / "content/game.ff").exists()) self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertEqual(sum(call[:2] == ("container", "rm") for call in self.docker.calls), 1) def test_heartbeat_cancellation_stops_create_before_starting_container(self): original_call = self.docker.call def cancel_during_create(*args, **kwargs): result = original_call(*args, **kwargs) if args[:2] == ("container", "create"): self.agent.cancelled_creations = frozenset(["create"]) return result self.docker.call = cancel_during_create payload = {"name": "test", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []} self.assertEqual(self.agent.execute(self.operation("instance.create", payload, "create"))["state"], "unknown") self.assertFalse(any(call[:2] == ("container", "start") for call in self.docker.calls)) result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"})) self.assertEqual(result["state"], "completed") self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) def test_cancellation_cleanup_resumes_after_agent_restart(self): self.create() self.docker.lose_reply_after = "rm" cancel = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel") self.assertEqual(self.agent.execute(cancel)["state"], "unknown") self.agent.journal.db.close() self.agent = Agent(self.config, self.docker) self.assertEqual(self.agent.execute(cancel)["state"], "completed") self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists()) def test_cancel_before_creation_delivery_fences_delayed_create(self): result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "late-create"})) self.assertEqual(result["state"], "completed", result) result = self.agent.execute(self.operation("instance.create", {}, "late-create")) self.assertEqual(result["state"], "failed") self.assertFalse(any(call[:2] == ("container", "create") for call in self.docker.calls)) def test_cancel_removes_partial_files_without_a_container(self): directory = self.agent.instance_dir(INSTANCE, 1) (directory / "app").mkdir(parents=True) (directory / "app/partial.cfg").write_text("partial") self.agent.journal.accept(self.operation("instance.create", {}, "create")) result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"})) self.assertEqual(result["state"], "completed", result) self.assertFalse(directory.exists()) self.assertTrue((self.library / "content/game.ff").exists()) def test_cancel_rejects_a_creation_id_from_another_instance(self): other = self.operation("instance.create", {}, "other-create") other["instanceId"] = "another-instance" self.agent.journal.accept(other) result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "other-create"})) self.assertEqual(result["state"], "failed", result) self.assertFalse((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").exists()) def test_cancel_creation_rejects_other_creation_and_newer_generation(self): self.create() result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "wrong"})) self.assertEqual(result["state"], "failed") module.atomic_json(self.agent.instance_dir(INSTANCE, 1).parent / "current.json", {"generation": 2}) result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}, "old-generation")) self.assertEqual(result["state"], "failed") self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"]) self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) def test_cancel_creation_reconciles_lost_removal_reply_and_fences_unknown_create(self): self.docker.lose_reply_after = "start" payload = {"name": "test", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []} create = self.operation("instance.create", payload, "create") self.assertEqual(self.agent.execute(create)["state"], "unknown") self.docker.lose_reply_after = "rm" cancel = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel") self.assertEqual(self.agent.execute(cancel)["state"], "unknown") self.assertEqual(self.agent.execute(cancel)["state"], "completed") self.assertEqual(self.agent.execute(create)["state"], "failed") self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists()) self.assertTrue(json.loads((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").read_text())["removed"]) def test_cancel_creation_rejects_unowned_container(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {} result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"})) self.assertEqual(result["state"], "failed") self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) def test_cancel_creation_removes_completed_recovery_rebuild(self): spec = self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} rebuild = self.operation("instance.rebuild", {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True}, "rebuild") self.assertEqual(self.agent.execute(rebuild)["state"], "completed") cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": "rebuild", "recoveryGeneration": 2}, "cancel") with patch.object(self.agent, "rcon", side_effect=AssertionError("Cancellation queried player status")): result = self.agent.execute(cancel) self.assertEqual(result["state"], "completed", result) self.assertFalse(json.loads(result["result"])["exists"]) for generation in (1, 2): self.assertFalse(self.agent.inspect(INSTANCE, generation)["exists"]) self.assertFalse(self.agent.instance_dir(INSTANCE, generation).exists()) self.assertEqual(json.loads((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").read_text()), {"generation": 2, "removed": True}) calls = list(self.docker.calls) self.assertEqual(self.agent.execute(cancel)["state"], "completed") self.assertEqual(self.docker.calls, calls) def test_cancel_creation_fences_undelivered_recovery_rebuild(self): spec = self.create() cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": "late-rebuild", "recoveryGeneration": 2}, "cancel") self.assertEqual(self.agent.execute(cancel)["state"], "completed") rebuild = self.operation("instance.rebuild", {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True}, "late-rebuild") result = self.agent.execute(rebuild) self.assertEqual(result["state"], "failed", result) self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) self.assertEqual(sum(call[:2] == ("container", "create") for call in self.docker.calls), 1) def test_cancel_creation_reconciles_recovery_rebuild_removal(self): spec = self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} self.docker.lose_reply_after = "start" rebuild = self.operation("instance.rebuild", {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True}, "rebuild") self.assertEqual(self.agent.execute(rebuild)["state"], "unknown") self.docker.lose_reply_after = "rm" cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": "rebuild", "recoveryGeneration": 2}, "cancel") self.assertEqual(self.agent.execute(cancel)["state"], "unknown") self.agent.journal.close() self.agent = Agent(self.config, self.docker) self.assertEqual(self.agent.execute(cancel)["state"], "completed") for generation in (1, 2): self.assertFalse(self.agent.inspect(INSTANCE, generation)["exists"]) self.assertFalse(self.agent.instance_dir(INSTANCE, generation).exists()) def test_cancel_creation_rejects_mismatched_recovery_rebuild(self): spec = self.create() for index, (kind, instance_id, generation, changes) in enumerate(( ("instance.restart", INSTANCE, 1, {}), ("instance.rebuild", "another-instance", 1, {}), ("instance.rebuild", INSTANCE, 2, {}), ("instance.rebuild", INSTANCE, 1, {"nextGeneration": 3}), ("instance.rebuild", INSTANCE, 1, {"unhealthyRecovery": False}), )): rebuild = self.operation(kind, {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True, **changes}, f"rebuild-{index}", generation=generation) rebuild["instanceId"] = instance_id self.agent.journal.accept(rebuild) cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": rebuild["id"], "recoveryGeneration": 2}, f"cancel-{index}") result = self.agent.execute(cancel) self.assertEqual(result["state"], "failed", result) self.assertIn("does not match", result["error"]) self.assertTrue(self.agent.inspect(INSTANCE, 1)["exists"]) self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) def test_cancel_creation_rejects_unrelated_recovery_container_or_newer_generation(self): self.create() original = self.docker.containers[self.agent.container_name(INSTANCE, 1)] unrelated = json.loads(json.dumps(original)) unrelated["Config"]["Labels"].update({module.LABEL + ".generation": "2", module.LABEL + ".operation": "unrelated"}) self.docker.containers[self.agent.container_name(INSTANCE, 2)] = unrelated payload = {"createOperationId": "create", "recoveryRebuildOperationId": "rebuild", "recoveryGeneration": 2} result = self.agent.execute(self.operation("instance.cancel-create", payload, "unrelated-container")) self.assertEqual(result["state"], "failed", result) self.assertIn("different creation", result["error"]) marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" module.atomic_json(marker, {"generation": 3}) result = self.agent.execute(self.operation("instance.cancel-create", payload, "newer-generation")) self.assertEqual(result["state"], "failed", result) self.assertIn("superseded", result["error"]) self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) self.assertEqual(json.loads(marker.read_text()), {"generation": 3}) def test_unowned_container_never_mutated(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {} self.assertEqual(self.agent.execute(self.operation("instance.remove"))["state"], "failed") self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls)) def test_private_bank_file_survives_stale_file_plans_without_snapshots(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" value = "configs/electric_cherry_bank.cfg" target = app / value secret = b'set tws_bank_token "ec_bank_private_regression"\n' target.write_bytes(secret) files = [{"id": 9, "path": value, "delete": True, "contentBase64": None}] with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.stage(INSTANCE, 1, {"stageId": "private", "files": files}, "stage") marker = self.agent.instance_dir(INSTANCE, 1) / "stages/private/manifest.json" # A stage written by a previous agent must be rejected on activation too. module.atomic_json(marker, {"state": "staged", "files": files}) with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.activate(INSTANCE, 1, "private", override_safety=True) with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.rebuild_before(INSTANCE, 1, "private", {"files": []}) with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.restore_files(app, [{"path": value, "contentBase64": None}]) with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.file_operation("files.read", INSTANCE, 1, {"path": value}, "read") with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.read_log(INSTANCE, 1, {"path": value}) manifest = self.agent.file_operation("files.manifest", INSTANCE, 1, {"roots": ["configs"], "paths": [value], "ignoreNames": []}, "manifest") self.assertNotIn(value, [item["path"] for item in manifest["files"]]) self.assertEqual(target.read_bytes(), secret) self.assertNotIn("before", json.loads(marker.read_text())) with self.assertRaisesRegex(AgentError, "Private bank"): self.agent.initial_files({"files": [{"path": value, "contentBase64": base64.b64encode(secret).decode()}]}) def prepare_t6_library(self): target = self.agent.library_dir("t6", self.library_id) module.shutil.copytree(self.library, target) manifest = json.loads((target / "manifest.json").read_text()) manifest["game"] = "t6" module.atomic_json(target / "manifest.json", manifest) def test_t6_create_materializes_private_bank_config(self): self.prepare_t6_library() self.test_create_materializes_private_bank_config_from_instance_credential(game="t6") def test_create_materializes_private_bank_config_from_instance_credential(self, game="t5"): secret = "ec_bank_" + "a" * 43 payload = {"name": "server", "game": game, "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "bank": {"token": secret, "configPath": "configs/electric_cherry_bank.cfg"}} operation = self.operation("instance.create", payload, "create") self.assertEqual(self.agent.execute(operation)["state"], "completed") path = self.agent.instance_dir(INSTANCE, 1) / "app/configs/electric_cherry_bank.cfg" self.assertEqual(path.read_text(), f'set tws_bank_token "{secret}"\n') self.assertEqual(self.agent.config["controller"], "https://controller.example") self.assertIn("bank.settlement", self.agent.inventory()["capabilities"]) payload["bank"]["token"] = secret + '\nset bad "1"' with self.assertRaises(AgentError): self.agent.initial_files(payload) def bank_journal(self, decisions=("commit", "cancel"), game="t5", endpoint=module.BANK_GAME_ENDPOINT): path = self.agent.instance_dir(INSTANCE, 1) / f"app/runtime/tws-bank/{game}/{module.digest(endpoint.encode())}.json" reports = [json.dumps({"server": "t5_zombie_test_1", "guid": "ace", "match": "match-one", "request_id": f"{game}-{i}", "operation": "deposit", "amount": 2000, "capacity": 1000000, "decision": decision}) for i, decision in enumerate(decisions)] module.atomic_json(path, {"version": 1, "server": "t5_zombie_test_1", "endpoint": endpoint, "reports": reports}) return path def bank_acknowledgment(self, method, path, payload): self.assertEqual((method, path), ("POST", "/api/agent/bank/settle")) self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) return {"state": "committed" if payload["action"] == "commit" else "cancelled", "request_id": payload["transfer"]["request_id"]} def test_remove_settles_both_game_journals_after_stop_before_deleting_storage(self): self.create() paths = [self.bank_journal(game=game) for game in ("t5", "t6")] def acknowledge(*args): self.assertTrue(all(path.exists() for path in paths)) self.assertEqual(args[2]["operationId"], "remove") return self.bank_acknowledgment(*args) with patch.object(self.agent, "request", side_effect=acknowledge) as request: result = self.agent.execute(self.operation("instance.remove", {"safetyOverride": True}, "remove")) self.assertEqual(result["state"], "completed", result) self.assertEqual(request.call_count, 4) self.assertTrue(all(not path.exists() for path in paths)) def test_remove_retains_bank_journal_during_outage_and_resumes_after_agent_restart(self): self.create() path = self.bank_journal() operation = self.operation("instance.remove", {"safetyOverride": True}, "remove") with patch.object(self.agent, "request", side_effect=AgentError("Controller request failed")): self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(path.exists()) self.agent.journal.close() self.agent = Agent(self.config, self.docker) calls = list(self.docker.calls) with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment): self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertFalse(path.exists()) self.assertEqual(self.docker.calls, calls) def test_t6_rebuild_preserves_private_bank_configuration(self): self.prepare_t6_library() self.test_endpoint_upgrade_rebuild_retries_original_decisions_before_starting_public_bank(game="t6") def test_endpoint_upgrade_rebuild_retries_original_decisions_before_starting_public_bank(self, game="t5"): secret = "ec_bank_" + "a" * 43 payload = self.create() payload["game"] = game payload["bank"] = {"token": secret, "configPath": "configs/electric_cherry_bank.cfg"} paths = [self.bank_journal(game=game, endpoint=endpoint) for game in ("t5", "t6") for endpoint in (module.BANK_GAME_ENDPOINT, "https://controller.example/api/bank")] originals = {path: path.read_bytes() for path in paths} operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2, "safetyOverride": True}, "rebuild") applied, calls = {}, [] def acknowledge(method, path, data): self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) self.assertEqual(self.agent.config["controller"], "https://controller.example") for journal, raw in originals.items(): self.assertEqual(journal.read_bytes(), raw) transfer = data["transfer"] request_id = transfer["request_id"] calls.append(request_id) if request_id in applied: self.assertEqual(applied[request_id], transfer) applied[request_id] = transfer response = self.bank_acknowledgment(method, path, data) if len(calls) == 1: raise AgentError("Lost reply after settlement") return response with patch.object(self.agent, "request", side_effect=acknowledge): self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.agent.journal.close() self.agent = Agent(self.config, self.docker) with patch.object(self.agent, "request", side_effect=acknowledge): self.assertEqual(self.agent.execute(operation)["state"], "failed") retry = self.operation("instance.rebuild", operation["payload"], "replacement") result = self.agent.execute(retry) self.assertEqual(result["state"], "completed", result) # Reconciliation drains once, then the replacement operation drains # again before deleting the unchanged journals. All retries keep IDs. self.assertEqual(len(calls), 17) self.assertEqual(len(applied), 4) self.assertEqual({item["decision"] for item in applied.values()}, {"commit", "cancel"}) self.assertTrue(all(not path.exists() for path in paths)) self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"]) config = self.agent.instance_dir(INSTANCE, 2) / "app/configs/electric_cherry_bank.cfg" self.assertEqual(config.read_text(), f'set tws_bank_token "{secret}"\n') def test_remove_drains_legacy_and_public_endpoints_without_following_journal_urls(self): self.create() paths = [self.bank_journal(game=game, endpoint=endpoint) for game in ("t5", "t6") for endpoint in (module.BANK_GAME_ENDPOINT, "https://controller.example/api/bank")] with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment) as request: result = self.agent.execute(self.operation("instance.remove", {"safetyOverride": True}, "remove")) self.assertEqual(result["state"], "completed", result) self.assertEqual(request.call_count, 8) self.assertTrue(all(not path.exists() for path in paths)) def test_unrecognized_bank_endpoints_retain_storage_without_settlement(self): self.create() path = self.bank_journal() operation = self.operation("instance.remove", {"safetyOverride": True}, "remove") for endpoint in ("https://untrusted.example/api/bank", "http://ec.thewrightserver.net/api/bank", "https://ec.thewrightserver.net.evil/api/bank", module.BANK_GAME_ENDPOINT + "?redirect=evil", "https://controller.example/other/api/bank", "", None): with self.subTest(endpoint=endpoint): journal = json.loads(path.read_text()) journal["endpoint"] = endpoint module.atomic_json(path, journal) with patch.object(self.agent, "request") as request: self.assertEqual(self.agent.execute(operation)["state"], "unknown") request.assert_not_called() self.assertTrue(path.exists()) def test_rebuild_settles_before_starting_replacement_and_retains_unknown_decisions(self): payload = self.create() path = self.bank_journal(("commit",)) operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2, "safetyOverride": True}, "rebuild") with patch.object(self.agent, "request", return_value={"state": "committed", "request_id": "wrong"}): self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(path.exists()) self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment): # Existing rebuild recovery records the interrupted create as failed; # a fresh operation can then rebuild without replaying the old command. self.assertEqual(self.agent.execute(operation)["state"], "failed") retry = self.operation("instance.rebuild", operation["payload"], "replacement") self.assertEqual(self.agent.execute(retry)["state"], "completed") self.assertFalse(path.exists()) self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"]) def test_cancel_creation_retries_bank_settlement_before_deleting_runtime(self): self.create() path = self.bank_journal(("cancel",)) operation = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel") with patch.object(self.agent, "request", side_effect=AgentError("offline")): self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(path.exists()) with patch.object(self.agent, "request", side_effect=self.bank_acknowledgment): self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertFalse(path.exists()) def test_malformed_or_symlinked_bank_journal_blocks_storage_deletion(self): self.create() path = self.bank_journal() path.write_text('{broken') operation = self.operation("instance.remove", {"safetyOverride": True}, "remove") with patch.object(self.agent, "request") as request: self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(path.exists()) outside = Path(self.temp.name) / "outside.json" outside.write_text('{"reports": []}') self.bank_journal() journal = json.loads(path.read_text()) journal["endpoint"] = "https://untrusted.example/api/bank" path.write_text(json.dumps(journal)) self.assertEqual(self.agent.execute(operation)["state"], "unknown") path.unlink() path.symlink_to(outside) self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(path.is_symlink()) request.assert_not_called() def staged_rebuild(self, deleting=False): spec = self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} file = {"id": 9, "path": "configs/server.cfg", "contentBase64": None if deleting else base64.b64encode(b"replacement").decode(), "checkBefore": True, "expectedBeforeSha": module.digest(b"original")} staged = self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [file]}, "stage")) self.assertEqual(staged["state"], "completed") spec["files"] = [] if deleting else [file] return self.operation("instance.rebuild", {"spec": spec, "stageId": "campaign", "nextGeneration": 2}, "rebuild") def test_file_rebuild_replaces_generation_and_retains_before_images(self): operation = self.staged_rebuild() result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) receipt = json.loads(result["result"]) self.assertEqual(receipt["before"], [{"id": 9, "path": "configs/server.cfg", "contentBase64": base64.b64encode(b"original").decode()}]) self.assertEqual(receipt["generation"], 2) self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists()) self.assertEqual((self.agent.instance_dir(INSTANCE, 2) / "app/configs/server.cfg").read_bytes(), b"replacement") self.assertIn("instance.rebuild-files", self.agent.inventory()["capabilities"]) calls = list(self.docker.calls) self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertEqual(self.docker.calls, calls) def test_file_rebuild_removes_requested_files(self): operation = self.staged_rebuild(deleting=True) result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) self.assertFalse((self.agent.instance_dir(INSTANCE, 2) / "app/configs/server.cfg").exists()) self.assertEqual(len(json.loads(result["result"])["before"]), 1) def test_file_rebuild_rejects_changed_targets_and_mismatched_replacement(self): operation = self.staged_rebuild() target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg" target.write_bytes(b"changed") self.assertEqual(self.agent.execute(operation)["state"], "failed") self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"]) target.write_bytes(b"original") operation["id"] = "mismatched" operation["payload"]["spec"]["files"] = [] self.assertEqual(self.agent.execute(operation)["state"], "failed") self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"]) self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) def test_file_rebuild_rechecks_players_after_snapshot_preparation(self): operation = self.staged_rebuild() with patch.object(self.agent, "rcon", side_effect=[{"ok": True, "players": 0}, {"ok": True, "players": 1}]): self.assertEqual(self.agent.execute(operation)["state"], "accepted") self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"]) self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) def test_file_rebuild_reconciles_snapshots_after_old_storage_was_removed(self): operation = self.staged_rebuild() self.assertEqual(self.agent.execute(operation)["state"], "completed") self.agent.journal.update(operation["id"], "unknown", error="Lost completion") calls = list(self.docker.calls) result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) self.assertEqual(json.loads(result["result"])["before"][0]["contentBase64"], base64.b64encode(b"original").decode()) self.assertEqual(self.docker.calls, calls) def test_rebuild_moves_generation_and_rejects_old_commands(self): payload = self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} result = self.agent.execute(self.operation("instance.rebuild", {**payload, "nextGeneration": 2})) self.assertEqual(result["state"], "completed", result) self.assertEqual(json.loads(result["result"])["generation"], 2) self.assertEqual(self.agent.execute(self.operation("instance.stop", op_id="old-stop"))["state"], "failed") self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"]) def assert_missing_instance_rebuild(self, remove_directory=False): payload = self.create() self.docker.containers.pop(self.agent.container_name(INSTANCE, 1)) old = self.agent.instance_dir(INSTANCE, 1) if remove_directory: module.shutil.rmtree(old) operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}) with patch.object(self.agent, "rcon", side_effect=AssertionError("Missing container queried player status")): result = self.agent.execute(operation) self.assertEqual(result["state"], "completed", result) self.assertEqual(json.loads(result["result"])["generation"], 2) self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"]) self.assertFalse(old.exists()) self.assertEqual(json.loads((old.parent / "current.json").read_text()), {"generation": 2}) self.assertFalse(any(call[:2] in (("container", "stop"), ("container", "rm")) for call in self.docker.calls)) calls = list(self.docker.calls) self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertEqual(self.docker.calls, calls) def test_rebuild_recreates_missing_container(self): self.assert_missing_instance_rebuild() def test_rebuild_completes_when_old_instance_directory_is_also_missing(self): self.assert_missing_instance_rebuild(remove_directory=True) def test_rebuild_keeps_real_old_directory_cleanup_failures_unresolved(self): payload = self.create() self.docker.containers.pop(self.agent.container_name(INSTANCE, 1)) old = self.agent.instance_dir(INSTANCE, 1) operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}) with patch.object(module.shutil, "rmtree", side_effect=PermissionError("Cleanup denied")): self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"]) self.assertTrue(old.exists()) self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertFalse(old.exists()) def interrupted_rebuild_before_create(self): payload = self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}, "interrupted-rebuild") def interrupted_seed(app, image): (app / "partial-runtime").write_text("incomplete") raise OSError(module.errno.ENOSPC, "disk full") with patch.object(self.agent, "seed_plutonium", side_effect=interrupted_seed): row = self.agent.execute(operation) self.assertEqual(row["state"], "unknown", row) self.assertEqual(json.loads(row["checkpoint"])["phase"], "disrupting") self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) return operation, row def test_rebuild_reconciles_removal_before_create_without_replaying(self): operation, interrupted = self.interrupted_rebuild_before_create() original = self.agent.instance_dir(INSTANCE, 1) partial = self.agent.instance_dir(INSTANCE, 2) module.atomic_json(partial / "spec.json", operation["payload"]["spec"]) self.agent.journal.close() self.agent = Agent(self.config, self.docker) calls = list(self.docker.calls) failed = self.agent.execute(operation) self.assertEqual(failed["state"], "failed", failed) result = json.loads(failed["result"]) self.assertEqual(result["rebuildFailure"], {"phase": "disrupting", "generation": 1, "nextGeneration": 2, "oldContainerMissing": True, "replacementContainerMissing": True, "replacementFilesRemoved": True}) self.assertTrue(result["reconciled"]) self.assertEqual(result["unexpectedError"], json.loads(interrupted["result"])["unexpectedError"]) self.assertEqual(result["originalError"], interrupted["error"]) self.assertTrue(failed["error"].startswith("Old container removed; replacement creation failed.")) self.assertIn(interrupted["error"], failed["error"]) self.assertEqual((original / "app/configs/server.cfg").read_bytes(), b"original") self.assertFalse(partial.exists()) self.assertEqual(json.loads((original.parent / "current.json").read_text()), {"generation": 1}) self.assertEqual(self.agent.execute(operation), failed) self.assertEqual(self.docker.calls, calls) replacement = self.operation("instance.rebuild", operation["payload"], "replacement-rebuild") self.assertEqual(self.agent.execute(replacement)["state"], "completed") self.assertFalse((partial / "app/partial-runtime").exists()) self.assertEqual(self.agent.inspect(INSTANCE, 2)["creationOperation"], replacement["id"]) calls = list(self.docker.calls) self.agent.journal.close() self.agent = Agent(self.config, self.docker) self.assertEqual(self.agent.execute(operation), failed) self.assertEqual(self.docker.calls, calls) def test_rebuild_removal_reconciles_when_marker_is_absent(self): operation, _ = self.interrupted_rebuild_before_create() marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" marker.unlink() row = self.agent.execute(operation) self.assertEqual(row["state"], "failed", row) self.assertIn("rebuildFailure", json.loads(row["result"])) self.assertFalse(marker.exists()) self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) def test_rebuild_lost_removal_receipt_settles_without_recreating_or_deleting_twice(self): payload = self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.docker.lose_reply_after = "rm" operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}, "rebuild") self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertFalse(self.agent.instance_dir(INSTANCE, 2).exists()) calls = list(self.docker.calls) result = self.agent.execute(operation) self.assertEqual(result["state"], "failed", result) self.assertTrue(json.loads(result["result"])["rebuildFailure"]["replacementFilesRemoved"]) self.assertEqual(self.docker.calls, calls) self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) def test_rebuild_removal_requires_both_containers_positively_missing(self): operation, interrupted = self.interrupted_rebuild_before_create() partial = self.agent.instance_dir(INSTANCE, 2) for generation in (1, 2): for status, foreign in (("running", False), ("created", False), ("exited", False), ("unknown", False), ("running", True)): with self.subTest(generation=generation, status=status, foreign=foreign): name = self.agent.container_name(INSTANCE, generation) labels = {} if foreign else {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE, module.LABEL + ".generation": str(generation)} self.docker.containers[name] = {"Config": {"Labels": labels}, "State": {"Status": status, "Running": status == "running"}} self.assertEqual(self.agent.execute(operation), interrupted) self.assertTrue(partial.exists()) self.docker.containers.pop(name) with patch.object(self.agent, "inspect", return_value={"exists": False, "running": False, "state": "unknown"}): self.assertEqual(self.agent.execute(operation), interrupted) with patch.object(self.agent, "inspect", side_effect=AgentError("Docker unavailable")): self.assertEqual(self.agent.execute(operation), interrupted) self.assertTrue(partial.exists()) def test_rebuild_removal_does_not_settle_after_a_create_checkpoint(self): operation, interrupted = self.interrupted_rebuild_before_create() for phase in ("creating", "starting", "unknown"): with self.subTest(phase=phase): self.agent.journal.checkpoint(operation["id"], {"phase": phase, "generation": 2}) row = self.agent.execute(operation) self.assertEqual(row["state"], "unknown", row) self.assertEqual(row["result"], interrupted["result"]) self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists()) self.agent.journal.db.execute("UPDATE operations SET checkpoint=NULL WHERE id=?", (operation["id"],)) self.agent.journal.db.commit() self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists()) def test_rebuild_removal_rejects_ambiguous_generation_markers(self): operation, interrupted = self.interrupted_rebuild_before_create() marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" for current in ({"generation": 0}, {"generation": 2}, {"generation": 3}, {"generation": True}, {"generation": "1"}, {"generation": 1, "removed": True}, {"generation": 1, "removed": None}, {}, []): with self.subTest(current=current): module.atomic_json(marker, current) self.assertEqual(self.agent.execute(operation), interrupted) self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists()) self.assertEqual(json.loads(marker.read_text()), current) marker.write_text("invalid JSON") self.assertEqual(self.agent.execute(operation), interrupted) def test_rebuild_removal_keeps_unknown_for_unrelated_partial_files_and_symlinks(self): operation, interrupted = self.interrupted_rebuild_before_create() partial = self.agent.instance_dir(INSTANCE, 2) module.atomic_json(partial / "spec.json", {**operation["payload"]["spec"], "name": "unrelated"}) self.assertEqual(self.agent.execute(operation), interrupted) self.assertTrue(partial.exists()) (partial / "spec.json").unlink() outside = self.agent.root / "retained-files" partial.rename(outside) partial.symlink_to(outside, target_is_directory=True) self.assertEqual(self.agent.execute(operation), interrupted) self.assertEqual((outside / "app/partial-runtime").read_text(), "incomplete") partial.unlink() outside.rename(partial) marker = partial.parent / "current.json" retained_marker = self.agent.root / "retained-current.json" marker.rename(retained_marker) marker.symlink_to(retained_marker) self.assertEqual(self.agent.execute(operation), interrupted) self.assertTrue(partial.exists()) def test_rebuild_removal_keeps_cleanup_errors_and_changed_generation_unresolved(self): operation, interrupted = self.interrupted_rebuild_before_create() partial = self.agent.instance_dir(INSTANCE, 2) with patch.object(module.shutil, "rmtree", side_effect=PermissionError("cleanup denied")): self.assertEqual(self.agent.execute(operation), interrupted) self.assertTrue(partial.exists()) cleanup = module.shutil.rmtree def superseded(path): cleanup(path) module.atomic_json(partial.parent / "current.json", {"generation": 3}) with patch.object(module.shutil, "rmtree", side_effect=superseded): self.assertEqual(self.agent.execute(operation), interrupted) self.assertEqual(json.loads((partial.parent / "current.json").read_text()), {"generation": 3}) self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) def test_rebuild_removal_waits_for_durable_partial_file_cleanup(self): operation, interrupted = self.interrupted_rebuild_before_create() with patch.object(module, "sync_parent", side_effect=OSError("sync unavailable")): self.assertEqual(self.agent.execute(operation), interrupted) self.assertFalse(self.agent.instance_dir(INSTANCE, 2).exists()) self.assertEqual(self.agent.execute(operation), interrupted) self.assertEqual(self.agent.execute(operation)["state"], "failed") self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) def test_traversal_and_symlinks_are_rejected(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" (app / "escape").symlink_to(self.temp.name) for path in ("../spec.json", "/etc/passwd", "escape/secret", "configs/../../spec.json"): with self.assertRaises(AgentError): module.scoped(app, path) def manifest(self, content=b"complete-game-file"): return self.manifest_files({"zone/test.ff": content}) def manifest_files(self, contents): files = [{"path": path, "size": len(content), "sha256": module.digest(content)} for path, content in contents.items()] return {"id": "downloaded", "game": "t6", "files": files, "sizeBytes": sum(item["size"] for item in files), "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())} def test_downloads_use_bounded_parallel_transfers_and_aggregate_progress(self): contents = {f"zone/file-{index}.ff": bytes([index]) * 31 for index in range(8)} manifest = self.manifest_files(contents) barrier = threading.Barrier(4) lock = threading.Lock() active = maximum = 0 snapshots = [] agent = self.agent class ConcurrentResponse(Response): def __exit__(self, *args): nonlocal active with lock: active -= 1 snapshots.append(agent.library_progress["downloaded"]["bytesDownloaded"]) return super().__exit__(*args) def fetch(method, path, data=None, extra_headers=None): nonlocal active, maximum if path == "/manifest": return Response(json.dumps(manifest).encode()) with lock: active += 1 maximum = max(maximum, active) # A serial downloader cannot reach this barrier. Two batches also # prove that an unlimited worker pool cannot open all eight files. barrier.wait(timeout=5) return ConcurrentResponse(contents[path.removeprefix("/files/")]) self.agent.artifact_open = fetch result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(maximum, 4) self.assertEqual(snapshots, sorted(snapshots)) self.assertEqual(max(snapshots), manifest["sizeBytes"]) self.assertEqual(result["size"], manifest["sizeBytes"]) for path, content in contents.items(): self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content" / path).read_bytes(), content) def test_all_downloads_finish_before_any_file_is_hashed_and_verified_from_disk(self): contents = {"zone/first.ff": b"game-file" * 200000, "zone/second.ff": b"second-file"} manifest = self.manifest_files(contents) self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")]) original_hash, original_open = module.hashlib.sha256, Path.open reads = [] def hashing(*args, **kwargs): progress = self.agent.library_progress.get("downloaded") if progress: self.assertEqual(progress["status"], "verifying") self.assertEqual(progress["bytesDownloaded"], manifest["sizeBytes"]) return original_hash(*args, **kwargs) def opening(path, mode="r", *args, **kwargs): if mode == "rb" and path.name.endswith(".ff"): self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) reads.append(path.name) return original_open(path, mode, *args, **kwargs) with patch.object(module.hashlib, "sha256", side_effect=hashing), patch.object(Path, "open", opening): result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(result["state"], "ready") self.assertCountEqual(reads, ["first.ff", "second.ff"]) cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") self.assertEqual(cache["bytesVerified"], manifest["sizeBytes"]) self.assertEqual(cache["transfer"]["phase"], "verifying") def test_phase_boundaries_are_reported_even_between_heartbeats_and_status_errors_are_nonfatal(self): content = b"game-file" manifest = self.manifest(content) reports = [] def fetch(method, path, data=None, extra_headers=None, timeout=60): if path == "/status": self.assertEqual(timeout, 5) reports.append(json.loads(data)) raise AgentError("Controller temporarily unavailable") self.assertEqual(timeout, 60) return Response(json.dumps(manifest).encode() if path == "/manifest" else content) self.agent.artifact_open = fetch result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "statusUrl": "/status"}) self.assertEqual(result["state"], "ready") self.assertEqual([item["status"] for item in reports], ["downloading", "verifying", "ready"]) self.assertEqual([item["bytesVerified"] for item in reports], [0, 0, len(content)]) self.assertEqual(reports[1]["bytesDownloaded"], len(content)) def test_status_socket_timeout_is_short_without_reducing_download_timeout(self): with patch.object(module.urllib.request, "build_opener") as build_opener: self.agent.artifact_open("GET", "/file") self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 60) self.agent.artifact_open("POST", "/status", b"{}", timeout=5) self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 5) def test_ignored_range_restarts_file_and_checksum(self): content = b"complete-game-file" manifest = self.manifest(content) staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" staging.mkdir(parents=True) (staging / "test.ff").write_bytes(b"bad-prefix") def fetch(method, path, data=None, extra_headers=None): if path == "/manifest": return Response(json.dumps(manifest).encode()) self.assertEqual(extra_headers, {"Range": "bytes=10-"}) return Response(content) self.agent.artifact_open = fetch result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(result["state"], "ready") self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content) def test_resumed_download_verifies_saved_prefix(self): content = b"complete-game-file" manifest = self.manifest(content) staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" staging.mkdir(parents=True) (staging / "test.ff").write_bytes(b"corrupt!") self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode()) if path == "/manifest" else Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"}) with self.assertRaisesRegex(AgentError, "integrity"): self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) self.assertFalse((staging / "test.ff").exists()) self.assertEqual(self.agent.library_progress["downloaded"]["bytesDownloaded"], 0) requests = [] def fetch(method, path, data=None, extra_headers=None): if path != "/manifest": requests.append(extra_headers) return Response(json.dumps(manifest).encode() if path == "/manifest" else content) self.agent.artifact_open = fetch self.assertEqual(self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})["state"], "ready") self.assertEqual(requests, [{}]) def test_all_corrupt_complete_files_are_removed_so_retry_downloads_them_again(self): contents = {"zone/first.ff": b"first", "zone/second.ff": b"second", "zone/good.ff": b"good"} manifest = self.manifest_files(contents) staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" staging.mkdir(parents=True) for name, content in contents.items(): (staging / Path(name).name).write_bytes(content if name.endswith("good.ff") else b"x" * len(content)) requests = [] def fetch(method, path, data=None, extra_headers=None): requests.append(path) return Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")]) self.agent.artifact_open = fetch payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} with self.assertRaisesRegex(AgentError, "integrity"): self.agent.library_download(payload) self.assertEqual(requests, ["/manifest"]) self.assertEqual([path.name for path in staging.iterdir()], ["good.ff"]) self.assertEqual(self.agent.library_download(payload)["state"], "ready") self.assertCountEqual(requests, ["/manifest", "/manifest", "/files/zone/first.ff", "/files/zone/second.ff"]) def test_verification_keeps_download_phase_complete_until_failure_reports_retained_bytes(self): contents = {"zone/first.ff": b"first", "zone/second.ff": b"second"} for mode in ("staged-corrupt", "existing-missing", "existing-short"): with self.subTest(mode=mode): manifest = {**self.manifest_files(contents), "id": mode} directory = self.agent.library_dir("t6", mode) if mode.startswith("existing"): (directory / "content/zone").mkdir(parents=True) (directory / "content/zone/second.ff").write_bytes(contents["zone/second.ff"]) if mode == "existing-short": (directory / "content/zone/first.ff").write_bytes(b"f") module.atomic_json(directory / "manifest.json", {"libraryId": mode, "game": "t6", "size": manifest["sizeBytes"], "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]}) def fetch(method, path, **kwargs): if path == "/manifest": return Response(json.dumps(manifest).encode()) content = contents[path.removeprefix("/files/")] return Response(b"x" * len(content) if path.endswith("first.ff") else content) self.agent.artifact_open = fetch original_open = Path.open snapshots = [] def opening(path, open_mode="r", *args, **kwargs): open_mode = kwargs.pop("mode", open_mode) if open_mode == "rb" and path.name == "second.ff": cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode) snapshots.append(cache) self.assertEqual(cache["status"], "verifying") self.assertEqual(cache["bytesDownloaded"], cache["bytesTotal"]) self.assertEqual(cache["transfer"]["phase"], "verifying") return original_open(path, open_mode, *args, **kwargs) with patch.object(Path, "open", opening), self.assertRaisesRegex(AgentError, "integrity"): self.agent.library_download({"libraryVersionId": mode, "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(len(snapshots), 1) failed = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode) self.assertEqual(failed["status"], "failed") self.assertEqual(failed["bytesDownloaded"], len(contents["zone/second.ff"]) + (1 if mode == "existing-short" else 0)) def test_parallel_failure_retains_partial_files_for_verified_resume(self): contents = {"zone/first.ff": b"first-game-file", "zone/second.ff": b"second-game-file"} manifest = self.manifest_files(contents) payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} def failed_fetch(method, path, data=None, extra_headers=None): if path == "/manifest": return Response(json.dumps(manifest).encode()) return Response(contents[path.removeprefix("/files/")][:5]) self.agent.artifact_open = failed_fetch with self.assertRaises(AgentError): self.agent.library_download(payload) self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) self.assertEqual(self.agent.library_progress["downloaded"]["status"], "failed") resumed = [] def resume_fetch(method, path, data=None, extra_headers=None): if path == "/manifest": return Response(json.dumps(manifest).encode()) content = contents[path.removeprefix("/files/")] if extra_headers: self.assertEqual(extra_headers, {"Range": "bytes=5-"}) resumed.append(path) return Response(content[5:], 206, {"Content-Range": f"bytes 5-{len(content)-1}/{len(content)}"}) return Response(content) self.agent.artifact_open = resume_fetch self.assertEqual(self.agent.library_download(payload)["state"], "ready") self.assertTrue(resumed) self.assertEqual(self.agent.verify_library(self.agent.library_dir("t6", "downloaded"))["size"], manifest["sizeBytes"]) def test_overlapping_paths_rejected_before_concurrent_downloads(self): manifest = self.manifest_files({"zone": b"file", "zone/child.ff": b"child"}) calls = [] def fetch(method, path, data=None, extra_headers=None): calls.append(path) return Response(json.dumps(manifest).encode()) self.agent.artifact_open = fetch with self.assertRaisesRegex(AgentError, "overlap"): self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(calls, ["/manifest"]) def test_failed_transfer_restart_reports_retained_bytes_and_resumes_with_only_remaining_space(self): content = b"complete-game-file" manifest = self.manifest(content) payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content[:8]) with self.assertRaises(AgentError): self.agent.library_download(payload) self.agent.journal.close() self.agent = Agent(self.config, self.docker) cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") self.assertEqual(cache["status"], "failed") self.assertEqual(cache["bytesDownloaded"], 8) self.assertEqual(cache["bytesVerified"], 0) self.assertEqual(cache["bytesTotal"], len(content)) self.assertEqual(cache["manifestSha256"], manifest["manifestSha256"]) self.assertEqual(cache["transfer"]["phase"], "downloading") self.assertIsNone(cache["transfer"]["etaSeconds"]) def fetch(method, path, data=None, extra_headers=None): if path == "/manifest": return Response(json.dumps(manifest).encode()) self.assertEqual(extra_headers, {"Range": "bytes=8-"}) return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"}) self.agent.artifact_open = fetch usage = module.shutil.disk_usage(self.agent.root) with patch.object(module.shutil, "disk_usage", return_value=type(usage)(usage.total, usage.used, len(content) - 8)): self.assertEqual(self.agent.library_download(payload)["state"], "ready") def test_unpromoted_manifest_is_never_reported_as_ready_and_progress_is_not_duplicated(self): content = b"complete-game-file" manifest = self.manifest(content) staging = self.agent.root / "libraries/t6/.downloaded.partial" (staging / "content/zone").mkdir(parents=True) # Oversized retained files can be replaced during resume. Their progress # must still stay within the published file size. (staging / "content/zone/test.ff").write_bytes(content + b"extra") saved = {"libraryId": "downloaded", "game": "t6", "size": len(content), "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]} module.atomic_json(staging / "source.json", saved) module.atomic_json(staging / "manifest.json", saved) inventory = self.agent.inventory() caches = [item for item in inventory["libraryCaches"] if item["libraryVersionId"] == "downloaded"] self.assertEqual(len(caches), 1) self.assertEqual(caches[0]["status"], "missing") self.assertEqual(caches[0]["bytesDownloaded"], len(content)) self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"]) (staging / "content/zone/test.ff").write_bytes(content[:5]) original_read = Path.read_text def read(path, *args, **kwargs): if path == staging / "source.json": self.fail("Unchanged partial manifest should be cached") return original_read(path, *args, **kwargs) with patch.object(Path, "read_text", read): partial = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") self.assertEqual(partial["bytesDownloaded"], 5) self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3} caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"] self.assertEqual(len(caches), 1) self.assertEqual(caches[0]["status"], "downloading") self.assertEqual(caches[0]["bytesDownloaded"], 3) def test_interrupted_verification_restarts_from_disk_without_redownloading_and_keeps_elapsed(self): content = b"complete-game-file" manifest = self.manifest(content) payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content) def interrupted(files, targets, transfer, repair=False): transfer.advance("bytesVerified", 5, 5) transfer.persist(force=True) raise KeyboardInterrupt("Simulated process termination") with patch.object(self.agent, "verify_downloaded_files", side_effect=interrupted), self.assertRaises(KeyboardInterrupt): self.agent.library_download(payload) previous = self.agent.library_transfers["downloaded"].snapshot() self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) self.agent.journal.close() self.agent = Agent(self.config, self.docker) interrupted_cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") self.assertEqual(interrupted_cache["status"], "missing") self.assertEqual(interrupted_cache["bytesDownloaded"], len(content)) self.assertEqual(interrupted_cache["bytesVerified"], 5) requests = [] def fetch(method, path, **kwargs): requests.append(path) return Response(json.dumps(manifest).encode()) original_verify = self.agent.verify_downloaded_files def verifying(files, targets, transfer, repair=False): self.assertEqual(transfer.snapshot()["bytesVerified"], 0) self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) return original_verify(files, targets, transfer, repair) self.agent.artifact_open = fetch with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying): self.assertEqual(self.agent.library_download(payload)["state"], "ready") self.assertEqual(requests, ["/manifest"]) cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded") self.assertEqual(cache["bytesVerified"], len(content)) self.assertEqual(cache["transfer"]["startedAt"], previous["transfer"]["startedAt"]) self.assertGreaterEqual(cache["transfer"]["elapsedSeconds"], interrupted_cache["transfer"]["elapsedSeconds"]) self.assertGreaterEqual(cache["transfer"]["phaseElapsedSeconds"], interrupted_cache["transfer"]["phaseElapsedSeconds"]) self.agent.journal.close() self.agent = Agent(self.config, self.docker) self.assertEqual(next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"), cache) def test_crash_before_atomic_promotion_reverifies_read_only_staging_without_network_transfers(self): content = b"complete-game-file" manifest = self.manifest(content) payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"} self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content) original_replace = module.os.replace def replace(source, destination): if Path(destination) == self.agent.library_dir("t6", "downloaded"): raise KeyboardInterrupt("Simulated process termination before promotion") return original_replace(source, destination) with patch.object(module.os, "replace", side_effect=replace), self.assertRaises(KeyboardInterrupt): self.agent.library_download(payload) staging = self.agent.root / "libraries/t6/.downloaded.partial/content" self.assertEqual(staging.stat().st_mode & 0o777, 0o555) self.agent.journal.close() self.agent = Agent(self.config, self.docker) requests = [] def fetch(method, path, **kwargs): requests.append(path) return Response(json.dumps(manifest).encode()) self.agent.artifact_open = fetch self.assertEqual(self.agent.library_download(payload)["state"], "ready") self.assertEqual(requests, ["/manifest"]) self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content) def test_existing_cache_reverification_is_visible_and_failure_revokes_readiness_after_restart(self): saved = json.loads((self.library / "manifest.json").read_text()) manifest = {**saved, "id": self.library_id} self.agent.artifact_open = lambda *args, **kwargs: Response(json.dumps(manifest).encode()) original_verify = self.agent.verify_downloaded_files def verifying(files, targets, transfer, repair=False): inventory = self.agent.inventory() cache = next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id) self.assertEqual(cache["status"], "verifying") self.assertEqual(cache["bytesVerified"], 0) self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"]) return original_verify(files, targets, transfer, repair) (self.library / "content/game.ff").write_bytes(b"corrupt!!") with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying), self.assertRaisesRegex(AgentError, "integrity"): self.agent.library_download({"libraryVersionId": self.library_id, "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.agent.journal.close() self.agent = Agent(self.config, self.docker) inventory = self.agent.inventory() self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"]) self.assertEqual(next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id)["status"], "failed") def test_elapsed_and_phase_eta_use_new_work_only_and_freeze_at_completion_or_failure(self): for terminal_status in ("ready", "failed"): with self.subTest(terminal_status=terminal_status): now = [100.0] progress = {"libraryVersionId": "downloaded", "status": "downloading", "bytesDownloaded": 80, "bytesVerified": 0, "bytesTotal": 100} previous = {"transfer": {"phase": "downloading", "startedAt": "2026-10-04T00:00:00Z", "elapsedSeconds": 20, "phaseElapsedSeconds": 20}} with patch.object(module.time, "monotonic", side_effect=lambda: now[0]): transfer = module.LibraryTransfer(progress, self.agent.root / "agent/timing.json", previous) self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) now[0] += 0.5 transfer.advance("bytesDownloaded", 5, 5) self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) now[0] += 1.5 snapshot = transfer.snapshot()["transfer"] self.assertEqual(snapshot["bytesPerSecond"], 2.5) self.assertEqual(snapshot["etaSeconds"], 6) self.assertEqual(snapshot["elapsedSeconds"], 22) now[0] += 1 transfer.advance("bytesDownloaded", 15, 15) transfer.begin_verification() self.assertEqual(transfer.snapshot()["transfer"]["phaseElapsedSeconds"], 0) self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"]) now[0] += 2 transfer.advance("bytesVerified", 40, 40) snapshot = transfer.snapshot()["transfer"] self.assertEqual(snapshot["bytesPerSecond"], 20) self.assertEqual(snapshot["etaSeconds"], 3) self.assertEqual(snapshot["elapsedSeconds"], 25) transfer.finish(terminal_status) final = transfer.snapshot() now[0] += 100 self.assertEqual(transfer.snapshot(), final) self.assertEqual(final["transfer"]["etaSeconds"], 0 if terminal_status == "ready" else None) def test_partial_inventory_rejects_changed_manifest_and_symlinked_content(self): content = b"complete-game-file" manifest = self.manifest(content) staging = self.agent.root / "libraries/t6/.downloaded.partial" (staging / "content/zone").mkdir(parents=True) source = {"libraryId": "downloaded", "game": "t6", "files": manifest["files"], "manifestSha256": "0" * 64} module.atomic_json(staging / "source.json", source) (staging / "content/zone/test.ff").write_bytes(content) self.assertFalse(any(item["libraryVersionId"] == "downloaded" for item in self.agent.inventory()["libraryCaches"])) source["manifestSha256"] = manifest["manifestSha256"] module.atomic_json(staging / "source.json", source) (staging / "content/zone/test.ff").unlink() (staging / "content/zone/test.ff").symlink_to(self.library / "content/game.ff") self.assertFalse(any(item["libraryVersionId"] == "downloaded" for item in self.agent.inventory()["libraryCaches"])) def test_resume_verifies_and_atomically_publishes_library(self): content = b"complete-game-file" manifest = self.manifest(content) staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" staging.mkdir(parents=True) (staging / "test.ff").write_bytes(content[:8]) calls = [] def fetch(method, path, data=None, extra_headers=None): calls.append((path, extra_headers)) if path == "/manifest": return Response(json.dumps(manifest).encode()) return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"}) self.agent.artifact_open = fetch result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertEqual(result["state"], "ready") self.assertEqual(calls[1][1], {"Range": "bytes=8-"}) self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content) self.assertEqual(self.agent.inventory()["libraryCaches"][-1]["status"], "ready") def test_corrupt_or_incomplete_library_never_promoted(self): manifest = self.manifest() def fetch(method, path, data=None, extra_headers=None): return Response(json.dumps(manifest).encode() if path == "/manifest" else b"corrupt") self.agent.artifact_open = fetch with self.assertRaisesRegex(AgentError, "integrity"): self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) self.assertFalse(self.agent.library_dir("t6", "downloaded").exists()) self.assertEqual(self.agent.library_progress["downloaded"]["status"], "failed") def test_manifest_tampering_rejected_before_download(self): manifest = self.manifest() manifest["files"][0]["size"] = 999 self.agent.artifact_open = lambda *_: Response(json.dumps(manifest).encode()) with self.assertRaisesRegex(AgentError, "manifest checksum"): self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) def test_artifact_urls_never_leak_credentials_to_other_origins(self): for value in ("https://attacker.example/files", "//attacker.example/files", "http://controller.example/files", "https://user:pass@controller.example/files"): with self.assertRaises(AgentError): self.agent.artifact_url(value) def test_unsupported_operation_does_not_spawn_shell(self): result = self.agent.execute(self.operation("shell.execute", {"command": "touch /tmp/unsafe"})) self.assertEqual(result["state"], "failed") self.assertEqual(self.docker.calls, []) def test_restart_lost_acknowledgement_recovers_from_changed_start_timestamp(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.docker.lose_reply_after = "restart" operation = self.operation("instance.restart") self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.agent.journal.close() self.agent = Agent(self.config, self.docker) recovered = self.agent.execute(operation) self.assertEqual(recovered["state"], "completed") self.assertTrue(json.loads(recovered["result"])["reconciled"]) self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1) def test_created_container_lost_start_reply_reconciles_immutable_operation_label(self): payload = {"name": "t5_zombie_test_1", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []} self.docker.lose_reply_after = "start" operation = self.operation("instance.create", payload) self.assertEqual(self.agent.execute(operation)["state"], "unknown") result = self.agent.execute(operation) self.assertEqual(result["state"], "completed") self.assertEqual(sum(call[:2] == ("container", "create") for call in self.docker.calls), 1) def test_remove_lost_reply_reconciles_absence_without_duplicate_delete(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.docker.lose_reply_after = "rm" operation = self.operation("instance.remove") self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists()) self.assertEqual(sum(call[:2] == ("container", "rm") for call in self.docker.calls), 1) def test_rotation_recovers_lost_reply_using_durably_saved_pending_credential(self): self.agent.config_path = str(Path(self.temp.name) / "agent.json") accepted = {} def request(method, path, payload): if not accepted: accepted["credential"] = payload["credential"] persisted = json.loads(Path(self.agent.config_path).read_text()) self.assertEqual(persisted["pendingCredential"], payload["credential"]) raise AgentError("Controller request failed") self.assertEqual(self.agent.config["credential"], accepted["credential"]) return {"ok": True} self.agent.request = request result = self.agent.execute(self.operation("credentials.rotate")) self.assertEqual(result["state"], "completed") saved = json.loads(Path(self.agent.config_path).read_text()) self.assertNotIn("pendingCredential", saved) self.assertEqual(saved["credential"], accepted["credential"]) def test_status_parser_never_treats_partial_or_unfamiliar_rows_as_empty(self): header = "map: zombie_cosmodrome\nnum score ping guid name\n" self.assertIsNone(module.status_players(header)) self.assertEqual(module.status_players(header + "--- ----- ---- ---- ----\n"), 0) for row in ("0 0 CNCT abc player", "1 90 ZMBI abc player", "2 4 ???? abc player"): self.assertEqual(module.status_players(header + "--- ----- ---- ---- ----\n" + row), 1) self.assertIsNone(module.status_players(header + "--- ----- ---- ---- ----\nconnection in progress")) self.assertEqual(module.status_players('statusResponse\n\\mapname\\mp_kino\\sv_maxclients\\4\n10 45 "joe"'), 1) def test_udp_rcon_uses_host_local_endpoint_and_returns_structured_occupancy(self): self.create() class FakeSocket: def __enter__(self): return self def __exit__(self, *_): pass def settimeout(self, value): pass def connect(self, endpoint): self.endpoint = endpoint def send(self, packet): self.packet = packet def recv(self, _size): if getattr(self, "done", False): raise module.socket.timeout() self.done = True return b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n0 42 CNCT abc player\n" sock = FakeSocket() with patch.object(module.socket, "socket", return_value=sock): result = self.agent.rcon(INSTANCE, 1, "status") self.assertEqual(sock.endpoint, ("127.0.0.1", 28960)) self.assertTrue(sock.packet.startswith(b"\xff\xff\xff\xffrcon rcon-secret status")) self.assertEqual(result["players"], 1) self.assertTrue(result["ok"]) @staticmethod def rcon_socket(*replies): sock = MagicMock() sock.__enter__.return_value = sock sock.recv.side_effect = replies return sock def test_udp_status_recovers_after_transient_failures(self): self.create() for error in (module.socket.timeout(), OSError("unreachable")): with self.subTest(error=type(error).__name__): failed = self.rcon_socket(error) # Keep all response packets, including occupied player rows. recovered = self.rcon_socket( b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n", b"\xff\xff\xff\xffprint\n0 42 CNCT abc player\n", module.socket.timeout()) with patch.object(module.socket, "socket", side_effect=[failed, recovered]) as sockets, patch.object(module.time, "sleep") as sleep: result = self.agent.rcon(INSTANCE, 1, "status") self.assertTrue(result["ok"]) self.assertEqual(result["players"], 1) self.assertEqual(sockets.call_count, 2) sleep.assert_called_once_with(0.25) failed.__exit__.assert_called_once() recovered.__exit__.assert_called_once() def test_udp_status_retries_are_bounded(self): self.create() attempts = [self.rcon_socket(module.socket.timeout()) for _ in range(3)] with patch.object(module.socket, "socket", side_effect=attempts) as sockets, patch.object(module.time, "sleep") as sleep: result = self.agent.rcon(INSTANCE, 1, "status") self.assertFalse(result["ok"]) self.assertEqual(result["reason"], "timeout") self.assertEqual(sockets.call_count, 3) self.assertEqual(sleep.call_args_list, [call(0.25), call(0.5)]) def test_udp_status_does_not_retry_bad_replies_or_partial_tables(self): self.create() for packet in (b"wrong listener", b"\xff\xff\xff\xffprint\nBad rcon_password", b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n"): with self.subTest(packet=packet): sock = self.rcon_socket(packet, module.socket.timeout()) with patch.object(module.socket, "socket", return_value=sock) as sockets, patch.object(module.time, "sleep") as sleep: result = self.agent.rcon(INSTANCE, 1, "status") self.assertIsNone(result["players"]) sockets.assert_called_once() sleep.assert_not_called() def test_udp_commands_with_effects_are_never_replayed(self): self.create() for command in ("map_restart", "kick 0", "version"): with self.subTest(command=command): sock = self.rcon_socket(module.socket.timeout()) with patch.object(module.socket, "socket", return_value=sock) as sockets, patch.object(module.time, "sleep") as sleep: result = self.agent.rcon(INSTANCE, 1, command) self.assertEqual(result["reason"], "timeout") sockets.assert_called_once() sleep.assert_not_called() def test_t6_bot_column_allows_empty_restart_but_holds_players_and_partial_replies(self): self.create() header = "map: zm_nuked\nnum score bot ping guid name lastmsg address qport rate\n" table = header + "--- ----- --- ---- -------------------------------- ---------------- ------- --------------------- ------ -----\n" operation = self.operation("instance.restart") for body, players in ( (header, None), (table + "0 0 0 CNCT abc player 0 127.0.0.1:28960 123 25000\n", 1), (table + "0 0 1 0 abc bot 0 loopback 0 25000\n", 1), (table + "unrecognized trailing output", None), (table, 0), ): with self.subTest(body=body): self.assertEqual(module.status_players(body), players) self.agent.rcon = lambda *_: {"ok": True, "players": module.status_players(body)} result = self.agent.execute(operation) self.assertEqual(result["state"], "completed" if players == 0 else "accepted") if players != 0: self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) def test_invalid_rebuild_preflight_preserves_running_original(self): payload = self.create() result = self.agent.execute(self.operation("instance.rebuild", {**payload, "image": "--privileged", "nextGeneration": 2})) self.assertEqual(result["state"], "failed") self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"]) self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls)) def test_file_operation_resists_parent_symlink_swap_after_validation(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" outside = Path(self.temp.name) / "outside" outside.mkdir() (outside / "server.cfg").write_bytes(b"host-data") original = module.scoped def swap(root, path, **kwargs): result = original(root, path, **kwargs) (app / "configs").rename(app / "saved-configs") (app / "configs").symlink_to(outside) return result with patch.object(module, "scoped", side_effect=swap): with self.assertRaises(OSError): module.managed_write(app, "configs/server.cfg", b"escape-attempt") self.assertEqual((outside / "server.cfg").read_bytes(), b"host-data") def test_deleting_absent_file_with_absent_parent_is_idempotent(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" module.managed_write(app, "missing/subdirectory/file.gsc", None) self.assertFalse((app / "missing").exists()) def test_controller_contact_failure_holds_before_running_any_docker_command(self): self.create() count = len(self.docker.calls) self.agent.report = lambda *_: (_ for _ in ()).throw(AgentError("Controller request failed")) result = self.agent.execute(self.operation("instance.stop"), report_running=True) self.assertEqual(result["state"], "accepted") self.assertEqual(len(self.docker.calls), count) def test_async_import_publication_resumes_after_controller_restart(self): responses = iter([{"status": "verifying"}, {"status": "pending"}, {"status": "verifying"}, {"status": "published", "version": {"id": "library-one"}}]) methods = [] def request(method, path, *args): methods.append(method) return Response(json.dumps(next(responses)).encode()) self.agent.artifact_open = request with patch.object(self.agent.stopping, "wait", return_value=False): result = self.agent.complete_import({"completeUrl": "/complete"}, start=True) self.assertEqual(result["status"], "published") self.assertEqual(methods, ["POST", "GET", "POST", "GET"]) def test_digest_pinned_runtime_inventory_and_exact_aliases(self): image_digest = "sha256:" + "1" * 64 image_id = "sha256:" + "2" * 64 original = self.docker.call def call(*args, **kwargs): if args[:2] == ("image", "ls"): return f"ghcr.io/ayymoss/plutainer|latest|{image_digest}|{image_id}\n" return original(*args, **kwargs) self.docker.call = call prepared = ["ghcr.io/ayymoss/plutainer", "ghcr.io/ayymoss/plutainer@" + image_digest] module.atomic_json(self.agent.root / "agent/runtime-images.json", {**dict.fromkeys(prepared, image_id), "removed-image": "missing-id"}) inventory = self.agent.inventory() images = inventory["runtimeImages"] self.assertIn("ghcr.io/ayymoss/plutainer@" + image_digest, images) self.assertIn("ghcr.io/ayymoss/plutainer:latest", images) self.assertIn("ghcr.io/ayymoss/plutainer", images) self.assertNotIn("removed-image", images) self.assertEqual(inventory["preparedRuntimeImages"], prepared) def test_runtime_pull_prepares_only_the_requested_reference_from_the_docker_cache(self): image = "registry.example/custom-runtime:stable" image_id = "sha256:" + "1" * 64 original = self.docker.call def call(*args, **kwargs): if args[:2] == ("image", "ls"): return (f"registry.example/custom-runtime|stable||{image_id}\n" f"registry.example/custom-runtime|unprepared-alias||{image_id}\n" "unrelated-service|latest||sha256:unrelated\n") if args[:2] == ("image", "inspect"): self.assertEqual(args[-1], image) return image_id return original(*args, **kwargs) self.docker.call = call before = self.agent.inventory() self.assertIn(image, before["runtimeImages"]) self.assertIn("unrelated-service:latest", before["runtimeImages"]) self.assertEqual(before["preparedRuntimeImages"], []) result = self.agent.execute(self.operation("runtime.pull", {"image": image})) self.assertEqual(result["state"], "completed") inventory = self.agent.inventory() self.assertEqual(inventory["runtimeImages"], before["runtimeImages"]) self.assertEqual(inventory["preparedRuntimeImages"], [image]) def test_runtime_preparation_inventory_fails_closed_when_records_or_docker_are_unavailable(self): aliases = self.agent.root / "agent/runtime-images.json" original = self.docker.call def call(*args, **kwargs): if args[:2] == ("image", "ls"): return "custom-runtime|latest||sha256:prepared\n" return original(*args, **kwargs) self.docker.call = call for content in ("invalid json", "[]", '{"custom-runtime:latest": []}', '{"custom-runtime:latest": "sha256:removed"}'): with self.subTest(content=content): aliases.write_text(content) inventory = self.agent.inventory() self.assertEqual(inventory["preparedRuntimeImages"], []) self.assertEqual(inventory["runtimeImages"], ["custom-runtime:latest"]) module.atomic_json(aliases, {"custom-runtime:latest": "sha256:prepared"}) with patch.object(self.docker, "call", side_effect=AgentError("Docker unavailable")): self.assertEqual(self.agent.inventory()["preparedRuntimeImages"], []) def test_heartbeat_applies_configured_host_port_range_before_deployment(self): self.agent.request = lambda *_: {"host": {"portMin": 30000, "portMax": 30100}} self.agent.heartbeat() self.assertEqual(self.agent.config["portRange"], {"min": 30000, "max": 30100}) def test_heartbeat_requires_valid_retention_and_never_changes_the_journal(self): self.assertEqual(self.agent.observation_keep_days, 0) with patch.object(self.agent.journal, "prune_observations", side_effect=AssertionError("heartbeat must not use the journal")), patch.object(self.agent.journal, "confirm_completed", side_effect=AssertionError("heartbeat must not use the journal")): for value, expected in ((1, 1), (36500, 36500), (0, 0), (True, 0), (False, 0), (1.0, 0), ("1", 0), (-1, 0), (36501, 0), (None, 0)): with self.subTest(value=value): self.agent.observation_keep_days = 3 self.agent.request = lambda *_, value=value: {"observationKeepDays": value} self.agent.heartbeat() self.assertEqual(self.agent.observation_keep_days, expected) self.agent.observation_keep_days = 3 self.agent.request = lambda *_: {} self.agent.heartbeat() self.assertEqual(self.agent.observation_keep_days, 0) def test_lost_completed_receipt_is_retained_and_replay_reports_the_saved_result(self): operation = self.operation("files.stat", {"path": "configs/server.cfg"}) self.agent.journal.accept(operation) with patch.object(module.time, "time", return_value=module.time.time() - 172800): row = self.agent.journal.update(operation["id"], "completed", {"size": 8}) self.agent.request = MagicMock(side_effect=AgentError("receipt lost")) with self.assertRaisesRegex(AgentError, "receipt lost"): self.agent.report(row) self.assertIsNone(self.agent.journal.get(operation["id"])["reported_at"]) self.assertEqual(self.agent.journal.prune_observations(1), 0) with patch.object(self.agent, "dispatch", side_effect=AssertionError("completed read must not be executed again")): replay = self.agent.execute(operation) self.assertEqual(replay["result"], row["result"]) for response in ({}, None, {"operation": {"id": "different", "state": "completed"}}, {"operation": {"id": operation["id"], "state": "running"}}): self.agent.request = lambda *_, response=response: response self.agent.report(replay) self.assertIsNone(self.agent.journal.get(operation["id"])["reported_at"]) self.agent.request = lambda *_: {"operation": {"id": operation["id"], "state": "completed"}} self.agent.report(replay) self.assertIsNotNone(self.agent.journal.get(operation["id"])["reported_at"]) self.assertEqual(self.agent.journal.prune_observations(1), 1) def test_threaded_read_lane_confirms_and_prunes_through_its_own_connection(self): self.create() old_operation = self.operation("files.list", op_id="old-observation") self.agent.journal.accept(old_operation) with patch.object(module.time, "time", return_value=module.time.time() - 172800): old = self.agent.journal.update(old_operation["id"], "completed", {"entries": []}) self.agent.journal.confirm_completed(old) self.agent.observation_keep_days = 1 operation = self.operation("files.stat", {"path": "configs/server.cfg"}, "thread-observation") polls = 0 failures = [] def request(method, path, payload=None): nonlocal polls if method == "POST": return {"operation": {"id": operation["id"], "state": payload["state"]}} polls += 1 if polls == 1: return {"operations": [operation]} self.agent.stopping.set() return {"operations": []} def run(): try: self.agent.read_loop() except BaseException as error: failures.append(error) self.agent.request = request with patch.object(self.agent.journal, "confirm_completed", side_effect=AssertionError("main journal used by reader")), patch.object(self.agent.journal, "prune_observations", side_effect=AssertionError("main journal used by reader")): thread = threading.Thread(target=run) thread.start() thread.join(timeout=5) self.agent.stopping.set() thread.join(timeout=5) self.assertFalse(thread.is_alive()) self.assertEqual(failures, []) self.assertEqual(polls, 2) self.assertIsNotNone(self.agent.journal.get(operation["id"])["reported_at"]) with self.assertRaisesRegex(AgentError, "Unknown operation"): self.agent.journal.get(old_operation["id"]) self.assertEqual(self.agent.journal.get("create")["state"], "completed") def test_download_content_permissions_allow_container_read_access_with_private_umask(self): content = b"complete-game-file" manifest = self.manifest(content) self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content) previous = module.os.umask(0o077) try: self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}) finally: module.os.umask(previous) content_root = self.agent.library_dir("t6", "downloaded") / "content" self.assertEqual(content_root.stat().st_mode & 0o777, 0o555) self.assertEqual((content_root / "zone").stat().st_mode & 0o777, 0o555) self.assertEqual((content_root / "zone/test.ff").stat().st_mode & 0o777, 0o444) def test_agent_restart_resumes_same_immutable_library_operation(self): content = b"complete-game-file" manifest = self.manifest(content) payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "manifestSha256": manifest["manifestSha256"]} operation = self.operation("library.download", payload) self.agent.journal.accept(operation) self.agent.journal.update(operation["id"], "running") staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone" staging.mkdir(parents=True) (staging / "test.ff").write_bytes(content[:8]) self.agent.journal.close() self.agent = Agent(self.config, self.docker) def fetch(method, path, data=None, extra_headers=None): if path == "/manifest": return Response(json.dumps(manifest).encode()) self.assertEqual(extra_headers, {"Range": "bytes=8-"}) return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"}) self.agent.artifact_open = fetch result = self.agent.execute(operation) self.assertEqual(result["state"], "completed") self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content) def test_player_join_during_activation_holds_restart_without_reapplying_files(self): self.create() target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg" original_write = module.managed_write for index, kind in enumerate(("instance.restart", "files.activate", "instance.start")): with self.subTest(kind=kind): before = target.read_bytes() stage_id = "late-join-" + str(index) content = ("new-" + str(index)).encode() self.agent.execute(self.operation("files.stage", {"stageId": stage_id, "files": [{"id": index, "path": "configs/server.cfg", "contentBase64": base64.b64encode(content).decode()}]}, "stage-late-" + str(index))) joined, writes = [False], [] self.agent.rcon = lambda *_: {"ok": True, "players": 1 if joined[0] else 0} def write_then_join(*args, **kwargs): original_write(*args, **kwargs) writes.append(args) joined[0] = True operation = self.operation(kind, {"stageId": stage_id}, "action-late-" + str(index)) restarts_before = sum(call[:2] == ("container", "restart") for call in self.docker.calls) with patch.object(module, "managed_write", side_effect=write_then_join): held = self.agent.execute(operation) self.assertEqual(held["state"], "accepted") self.assertEqual(json.loads(held["result"])["blocked"], "occupied") self.assertEqual(target.read_bytes(), content) self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), restarts_before) joined[0] = False completed = self.agent.execute(operation) self.assertEqual(completed["state"], "completed") self.assertEqual(len(writes), 1) self.assertEqual(base64.b64decode(json.loads(completed["result"])["before"][0]["contentBase64"]), before) self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), restarts_before + 1) def test_player_join_during_rollback_holds_restart_without_repeating_restore(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.agent.execute(self.operation("files.stage", {"stageId": "rollback-join", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage-rollback-join")) self.agent.execute(self.operation("files.activate", {"stageId": "rollback-join", "restart": False}, "activate-rollback-join")) joined, writes = [False], [] original_write = module.managed_write self.agent.rcon = lambda *_: {"ok": True, "players": 1 if joined[0] else 0} def restore_then_join(*args, **kwargs): original_write(*args, **kwargs) writes.append(args) joined[0] = True operation = self.operation("files.rollback", {"stageId": "rollback-join"}, "rollback-join") with patch.object(module, "managed_write", side_effect=restore_then_join): self.assertEqual(self.agent.execute(operation)["state"], "accepted") self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) joined[0] = False self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertEqual(len(writes), 1) self.assertEqual((self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg").read_bytes(), b"original") def test_start_with_stage_does_not_reconcile_success_before_required_restart(self): self.create() self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.agent.execute(self.operation("files.stage", {"stageId": "start-stage", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage-start")) self.docker.fail_restart = True operation = self.operation("instance.start", {"stageId": "start-stage"}) self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertEqual(self.agent.execute(operation)["state"], "unknown") self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1) def test_running_unhealthy_container_uses_fresh_occupancy_before_recovery(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} operation = self.operation("instance.restart") self.agent.rcon = lambda *_: {"ok": True, "players": 2} self.assertEqual(self.agent.execute(operation)["state"], "accepted") self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.assertEqual(self.agent.execute(operation)["state"], "completed") def test_incremental_log_reads_append_bytes_without_rereading_unchanged_content(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" (app / "logs").mkdir() path = app / "logs/games_sp.log" path.write_bytes(b"first\n") payload = {"path": "logs/games_sp.log", "mode": "incremental"} result = self.agent.execute(self.operation("logs.read", payload)) self.assertEqual(result["state"], "completed") first = json.loads(result["result"]) self.assertEqual(base64.b64decode(first["contentBase64"]), b"first\n") self.assertTrue(first["reset"]) # An unchanged stat must not read even the cursor anchor. with path.open("rb") as stream, patch.object(stream, "read", side_effect=AssertionError("unchanged file was read")): unchanged = self.agent.read_log_chunk(stream, os.fstat(stream.fileno()), first["cursor"]) self.assertEqual(unchanged["contentBase64"], "") with path.open("ab") as stream: stream.write("café\n".encode()) appended = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": first["cursor"]}) self.assertEqual(base64.b64decode(appended["contentBase64"]), "café\n".encode()) self.assertFalse(appended["reset"]) self.assertEqual(appended["cursor"]["offset"], len("first\ncafé\n".encode())) self.assertIn("logs.read-offset", self.agent.inventory()["capabilities"]) def test_incremental_log_rotation_and_copy_truncate_reset_the_cursor(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" (app / "logs").mkdir() path = app / "logs/games_sp.log" header = b"same game header\n" * 30 path.write_bytes(header + b"original\n") payload = {"path": "logs/games_sp.log", "mode": "incremental"} result = self.agent.read_log(INSTANCE, 1, payload) path.rename(path.with_suffix(".old")) for content in (header + b"rotated!\n", header + b"rewritten and longer\n", b"short\n", b""): with self.subTest(content=content): path.write_bytes(content) result = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": result["cursor"]}) self.assertTrue(result["reset"]) self.assertEqual(base64.b64decode(result["contentBase64"]), content) def test_incremental_log_backlogs_are_bounded_and_preserve_split_utf8_bytes(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" (app / "logs").mkdir() path = app / "logs/games_sp.log" window = 256 * 1024 path.write_bytes(b"x" * (window + 10)) payload = {"path": "logs/games_sp.log", "mode": "incremental"} initial = self.agent.read_log(INSTANCE, 1, payload) self.assertTrue(initial["truncated"]) self.assertEqual(len(base64.b64decode(initial["contentBase64"])), window) data = b"y" * (window - 1) + "é\n".encode() with path.open("ab") as stream: stream.write(data) first = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": initial["cursor"]}) second = self.agent.read_log(INSTANCE, 1, {**payload, "cursor": first["cursor"]}) self.assertFalse(first["reset"]) self.assertEqual(len(base64.b64decode(first["contentBase64"])), window) self.assertEqual(base64.b64decode(first["contentBase64"]) + base64.b64decode(second["contentBase64"]), data) self.assertEqual(second["cursor"]["offset"], second["size"]) for cursor in ({}, {**second["cursor"], "offset": -1}, {**second["cursor"], "anchorLength": 257}, {**second["cursor"], "offset": True}): with self.subTest(cursor=cursor), self.assertRaisesRegex(AgentError, "Invalid log cursor"): self.agent.read_log(INSTANCE, 1, {**payload, "cursor": cursor}) def test_game_log_can_follow_internal_relative_plutainer_symlink(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" (app / "logs").mkdir() (app / "runtime").mkdir() (app / "runtime/game.log").write_text("first\nsecond\nthird\n") (app / "logs/games_sp.log").symlink_to("../runtime/game.log") result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log", "lines": 2})) self.assertEqual(result["state"], "completed") self.assertEqual(json.loads(result["result"])["lines"], ["second", "third"]) incremental = self.agent.read_log(INSTANCE, 1, {"path": "logs/games_sp.log", "mode": "incremental"}) self.assertEqual(base64.b64decode(incremental["contentBase64"]), b"first\nsecond\nthird\n") with self.assertRaises(AgentError): module.managed_write(app, "logs/games_sp.log", b"must-not-write-through-link") def test_game_log_rejects_symlink_escaping_instance(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" (app / "logs").mkdir() outside = Path(self.temp.name) / "host-secret" outside.write_text("secret") (app / "logs/games_sp.log").symlink_to(outside) result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log"})) self.assertEqual(result["state"], "failed") self.assertIn("outside", result["error"]) with self.assertRaisesRegex(AgentError, "outside"): self.agent.read_log(INSTANCE, 1, {"path": "logs/games_sp.log", "mode": "incremental"}) def test_console_log_falls_back_to_owned_docker_output_including_stderr(self): self.create() original = self.docker.call def call(*args, **kwargs): if args[:2] == ("container", "logs"): self.assertTrue(kwargs["merge_stderr"]) self.assertEqual(args[-1], self.agent.container_name(INSTANCE, 1)) return "stdout-line\nstderr-line\n" return original(*args, **kwargs) self.docker.call = call result = self.agent.execute(self.operation("logs.read", {"path": "logs/console.log"})) self.assertEqual(result["state"], "completed") self.assertEqual(json.loads(result["result"])["lines"], ["stdout-line", "stderr-line"]) if __name__ == "__main__": unittest.main()