767 lines
44 KiB
Python
767 lines
44 KiB
Python
import base64
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import 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
|
|
|
|
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[: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 AgentTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.config = {"root": self.temp.name, "controller": "https://controller.example", "credential": "secret", "portRange": {"min": 28960, "max": 28980}}
|
|
self.docker = FakeDocker()
|
|
self.agent = Agent(self.config, self.docker)
|
|
self.library_id = "library-one"
|
|
self.library = self.agent.library_dir("t5", self.library_id)
|
|
(self.library / "content").mkdir(parents=True)
|
|
(self.library / "content" / "game.ff").write_bytes(b"game-data")
|
|
files = [{"path": "game.ff", "size": 9, "sha256": module.digest(b"game-data")}]
|
|
module.atomic_json(self.library / "manifest.json", {"libraryId": self.library_id, "game": "t5", "size": 9, "files": files, "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())})
|
|
|
|
def tearDown(self):
|
|
self.agent.journal.close()
|
|
self.temp.cleanup()
|
|
|
|
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_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_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 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 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_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_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 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_fresh_download_hashes_stream_without_rereading_whole_file(self):
|
|
content = b"game-file" * 200000
|
|
manifest = self.manifest(content)
|
|
self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
|
|
with patch.object(module, "file_digest", side_effect=AssertionError("download reread the file")):
|
|
result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
|
|
self.assertEqual(result["state"], "ready")
|
|
|
|
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())
|
|
|
|
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, {"libraryVersionId": "downloaded", "status": "missing", "bytesDownloaded": 8, "bytesTotal": len(content), "manifestSha256": manifest["manifestSha256"]})
|
|
|
|
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"])
|
|
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_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"])
|
|
|
|
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
|
|
module.atomic_json(self.agent.root / "agent/runtime-images.json", {"ghcr.io/ayymoss/plutainer": image_id, "removed-image": "missing-id"})
|
|
images = self.agent.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)
|
|
|
|
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_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_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"])
|
|
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"])
|
|
|
|
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()
|