2172 lines
124 KiB
Python
2172 lines
124 KiB
Python
import base64
|
|
import importlib.util
|
|
import http.client
|
|
import io
|
|
import json
|
|
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
|
|
|
|
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 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<? ORDER BY updated,id LIMIT 500""", (self.now,)).fetchall()
|
|
self.assertIn("operations_observation_retention", " ".join(str(tuple(row)) for row in plan))
|
|
|
|
def test_migration_keeps_legacy_completions_until_their_receipt_is_reacknowledged(self):
|
|
self.journal.close()
|
|
self.path.unlink()
|
|
legacy = module.sqlite3.connect(self.path)
|
|
legacy.execute("CREATE TABLE operations(id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
|
|
request = {"id": "legacy-read", "kind": "files.list", "instanceId": None, "generation": None, "payload": {}}
|
|
canonical = json.dumps(request, sort_keys=True, separators=(",", ":"))
|
|
legacy.execute("INSERT INTO operations VALUES(?,?,?,'completed',?,NULL,?)", (request["id"], module.digest(canonical.encode()), canonical, '{"entries":[]}', self.now - 172800))
|
|
legacy.commit()
|
|
legacy.close()
|
|
self.journal = Journal(self.path)
|
|
row, accepted = self.journal.accept(request)
|
|
self.assertFalse(accepted)
|
|
self.assertIsNone(row["reported_at"])
|
|
self.assertEqual(self.prune(), 0)
|
|
self.journal.confirm_completed(row)
|
|
self.assertEqual(self.prune(), 1)
|
|
|
|
|
|
class ManifestTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name) / "app"
|
|
self.root.mkdir()
|
|
self.stopping = threading.Event()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def write(self, path, content=b"content"):
|
|
target = self.root / path
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(content)
|
|
return target
|
|
|
|
def manifest(self, roots=None, ignore_names=None, paths=None):
|
|
return module.managed_manifest(self.root, {
|
|
"roots": ["scripts", "configs", "plugins"] if roots is None else roots,
|
|
"ignoreNames": [] if ignore_names is None else ignore_names,
|
|
"paths": [] if paths is None else paths,
|
|
}, self.stopping)
|
|
|
|
def test_hashes_sorted_unique_files_across_overlapping_roots_and_skips_ignored_names(self):
|
|
self.write("scripts/nested/z.gsc", b"script")
|
|
self.write("configs/server.cfg", b"config")
|
|
self.write("scripts/cache/ignored.gsc")
|
|
self.write("scripts/nested/ignore.gsc")
|
|
expected = [{"path": path, "size": len(content), "sha256": module.digest(content)} for path, content in (
|
|
("configs/server.cfg", b"config"), ("scripts/nested/z.gsc", b"script"),
|
|
)]
|
|
self.assertEqual(self.manifest(["scripts/nested", "configs", "scripts", "scripts", "plugins"], ["cache", "ignore.gsc"]), {"files": expected})
|
|
|
|
def test_nested_symlinks_are_skipped_but_requested_symlink_roots_fail(self):
|
|
self.write("scripts/good.gsc")
|
|
outside = Path(self.temp.name) / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret").write_bytes(b"secret")
|
|
(self.root / "scripts/file-link").symlink_to(outside / "secret")
|
|
(self.root / "scripts/dir-link").symlink_to(outside, target_is_directory=True)
|
|
(self.root / "scripts/dangling").symlink_to(outside / "missing")
|
|
self.assertEqual([item["path"] for item in self.manifest()["files"]], ["scripts/good.gsc"])
|
|
for roots in (["scripts/dir-link"], ["scripts", "scripts/dir-link"]):
|
|
with self.subTest(roots=roots), self.assertRaisesRegex(AgentError, "symlink|Symlink"):
|
|
self.manifest(roots)
|
|
|
|
def test_validates_roots_and_ignored_names(self):
|
|
for roots in ([], "scripts", [""], ["."], [".."], ["../scripts"], ["/scripts"], ["scripts//nested"], ["scripts/../configs"], [None], ["scripts"] * 65):
|
|
with self.subTest(roots=roots), self.assertRaises(AgentError):
|
|
self.manifest(roots)
|
|
for ignored in ("cache", [None], [""], ["."], [".."], ["scripts/cache"], ["a\\b"], ["a\x00b"], ["cache"] * 257):
|
|
with self.subTest(ignored=ignored), self.assertRaises(AgentError):
|
|
self.manifest(ignore_names=ignored)
|
|
for paths in ("scripts/a", [None], ["../outside"], ["."], ["scripts/a"] * 10001):
|
|
with self.subTest(paths=paths), self.assertRaises(AgentError):
|
|
self.manifest(paths=paths)
|
|
|
|
def test_explicit_managed_paths_cannot_be_symlinks_or_directories(self):
|
|
self.write("scripts/good.gsc")
|
|
(self.root / "scripts/link.gsc").symlink_to(self.root / "scripts/good.gsc")
|
|
(self.root / "scripts/dangling.gsc").symlink_to(self.root / "missing")
|
|
for path in ("scripts", "scripts/link.gsc", "scripts/dangling.gsc"):
|
|
with self.subTest(path=path), self.assertRaises(AgentError):
|
|
self.manifest(paths=[path])
|
|
self.assertEqual([item["path"] for item in self.manifest(paths=["scripts/good.gsc", "scripts/missing.gsc", "missing/future.gsc"])["files"]], ["scripts/good.gsc"])
|
|
|
|
def test_explicit_managed_paths_are_validated_again_after_the_scan(self):
|
|
self.write("scripts/good.gsc")
|
|
original = module.managed_open
|
|
opened = 0
|
|
|
|
@module.contextlib.contextmanager
|
|
def swap(root, value):
|
|
nonlocal opened
|
|
opened += 1
|
|
if opened == 2:
|
|
(self.root / "scripts/good.gsc").unlink()
|
|
(self.root / "scripts/good.gsc").symlink_to(self.root / "missing")
|
|
with original(root, value) as stream:
|
|
yield stream
|
|
|
|
with patch.object(module, "managed_open", side_effect=swap), self.assertRaisesRegex(AgentError, "Symlinks"):
|
|
self.manifest(paths=["scripts/good.gsc"])
|
|
|
|
def test_missing_destinations_are_empty_but_unavailable_app_and_regular_roots_fail(self):
|
|
self.assertEqual(self.manifest(), {"files": []})
|
|
self.write("scripts", b"not a directory")
|
|
with self.assertRaisesRegex(AgentError, "complete managed tree"):
|
|
self.manifest()
|
|
(self.root / "scripts").unlink()
|
|
self.root.rmdir()
|
|
with self.assertRaisesRegex(AgentError, "complete managed tree"):
|
|
self.manifest()
|
|
replacement = Path(self.temp.name) / "replacement"
|
|
replacement.mkdir()
|
|
self.root.symlink_to(replacement, target_is_directory=True)
|
|
with self.assertRaisesRegex(AgentError, "symlink"):
|
|
self.manifest()
|
|
|
|
def test_errors_and_special_files_never_return_a_partial_manifest(self):
|
|
self.write("scripts/good.gsc")
|
|
self.write("scripts/unreadable.gsc")
|
|
original = module.os.stat
|
|
|
|
def denied(path, *args, **kwargs):
|
|
if path == "unreadable.gsc":
|
|
raise PermissionError("unreadable")
|
|
return original(path, *args, **kwargs)
|
|
|
|
with patch.object(module.os, "stat", side_effect=denied), self.assertRaisesRegex(AgentError, "complete managed tree"):
|
|
self.manifest()
|
|
module.os.mkfifo(self.root / "scripts/fifo")
|
|
with self.assertRaisesRegex(AgentError, "non-regular"):
|
|
self.manifest()
|
|
|
|
def test_scan_limits_fail_instead_of_truncating_results(self):
|
|
self.write("scripts/a.gsc")
|
|
self.write("scripts/b.gsc")
|
|
for setting, limit, message in (
|
|
("MANIFEST_MAX_ENTRIES", 1, "entry limit"),
|
|
("MANIFEST_MAX_BYTES", 1, "result size limit"),
|
|
("MANIFEST_MAX_SECONDS", 0, "time limit"),
|
|
("MANIFEST_MAX_DEPTH", 0, "depth limit"),
|
|
):
|
|
with self.subTest(setting=setting), patch.object(module, setting, limit), self.assertRaisesRegex(AgentError, message):
|
|
self.manifest()
|
|
self.stopping.set()
|
|
with self.assertRaisesRegex(AgentError, "interrupted"):
|
|
self.manifest()
|
|
|
|
def test_files_are_streamed_and_not_limited_by_the_content_read_api(self):
|
|
content = b"x" * (2 * 1024 * 1024 + 13)
|
|
self.write("plugins/large.dll", content)
|
|
original = module.os.fdopen
|
|
reads = []
|
|
|
|
@module.contextlib.contextmanager
|
|
def tracked(*args, **kwargs):
|
|
with original(*args, **kwargs) as stream:
|
|
proxy = MagicMock(wraps=stream)
|
|
def read(size):
|
|
reads.append(size)
|
|
return stream.read(size)
|
|
proxy.read.side_effect = read
|
|
yield proxy
|
|
|
|
with patch.object(module, "MAX_FILE", 1), patch.object(module.os, "fdopen", side_effect=tracked):
|
|
result = self.manifest()
|
|
self.assertEqual(result, {"files": [{"path": "plugins/large.dll", "size": len(content), "sha256": module.digest(content)}]})
|
|
self.assertGreater(len(reads), 2)
|
|
self.assertEqual(set(reads), {1024 * 1024})
|
|
|
|
def test_file_changes_while_hashing_invalidate_the_scan(self):
|
|
target = self.write("scripts/changing.gsc", b"before")
|
|
original = module.hashlib.sha256
|
|
|
|
def changing_hash():
|
|
digest = original()
|
|
proxy = MagicMock(wraps=digest)
|
|
def update(block):
|
|
digest.update(block)
|
|
prior = target.stat()
|
|
target.write_bytes(b"after!")
|
|
module.os.utime(target, ns=(prior.st_atime_ns, prior.st_mtime_ns + 1))
|
|
proxy.update.side_effect = update
|
|
return proxy
|
|
|
|
with patch.object(module.hashlib, "sha256", side_effect=changing_hash), self.assertRaisesRegex(AgentError, "file changed"):
|
|
self.manifest()
|
|
|
|
def test_symlink_swap_after_validation_cannot_hash_outside_the_app(self):
|
|
self.write("scripts/good.gsc")
|
|
outside = Path(self.temp.name) / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret").write_bytes(b"secret")
|
|
original = module.scoped
|
|
|
|
def swap(root, relative, **kwargs):
|
|
result = original(root, relative, **kwargs)
|
|
(self.root / "scripts").rename(self.root / "saved")
|
|
(self.root / "scripts").symlink_to(outside, target_is_directory=True)
|
|
return result
|
|
|
|
with patch.object(module, "scoped", side_effect=swap), self.assertRaisesRegex(AgentError, "complete managed tree"):
|
|
self.manifest(["scripts"])
|
|
|
|
def test_disappearing_app_after_hashing_is_not_successful_evidence(self):
|
|
self.write("scripts/good.gsc")
|
|
original = module.hashlib.sha256
|
|
|
|
def moving_hash():
|
|
digest = original()
|
|
proxy = MagicMock(wraps=digest)
|
|
def update(block):
|
|
digest.update(block)
|
|
self.root.rename(self.root.with_name("moved-app"))
|
|
proxy.update.side_effect = update
|
|
return proxy
|
|
|
|
with patch.object(module.hashlib, "sha256", side_effect=moving_hash), self.assertRaisesRegex(AgentError, "complete managed tree"):
|
|
self.manifest()
|
|
|
|
|
|
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 test_controller_truncated_response_is_a_sanitized_retryable_error(self):
|
|
# Exercise Python's actual HTTP body reader against a truncated stream.
|
|
sock = MagicMock()
|
|
sock.makefile.return_value = io.BytesIO(b'HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n{"secret":')
|
|
response = http.client.HTTPResponse(sock)
|
|
response.begin()
|
|
with patch.object(module.urllib.request, "build_opener") as opener:
|
|
opener.return_value.open.return_value = response
|
|
with self.assertRaisesRegex(AgentError, "^Controller request failed$"):
|
|
self.agent.request("GET", "/api/agent/operations")
|
|
|
|
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], 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_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 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_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_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_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
|
|
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_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_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()
|