507 lines
27 KiB
Python
507 lines
27 KiB
Python
import base64
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import urllib.error
|
|
|
|
import test_agent as fixtures
|
|
|
|
|
|
module = fixtures.module
|
|
AgentError = fixtures.AgentError
|
|
IMAGE = "ghcr.io/ayymoss/plutainer:latest"
|
|
BOOTSTRAPPER = "bin/plutonium-bootstrapper-win32.exe"
|
|
BASE_URL = "https://cdn.plutoniummod.com/updater/prod/files/"
|
|
|
|
|
|
class PlutoniumCacheTests(unittest.TestCase):
|
|
tearDown = fixtures.AgentTests.tearDown
|
|
operation = fixtures.AgentTests.operation
|
|
create = fixtures.AgentTests.create
|
|
|
|
def setUp(self):
|
|
fixtures.AgentTests.setUp(self)
|
|
network = patch.object(self.agent, "plutonium_open", side_effect=AssertionError("Unexpected vendor request"), create=True)
|
|
network.start()
|
|
self.addCleanup(network.stop)
|
|
|
|
def release(self, revision=5354, contents=None):
|
|
contents = contents or {
|
|
BOOTSTRAPPER: b"bootstrapper",
|
|
"bin/plutonium.dll": b"runtime",
|
|
"storage/t5/zone/plutonium.ff": b"zone-data",
|
|
"storage/iw5/images/~weapon_ak47_iw5_spc-rgb&weap~fa48e615.iwi": b"texture-data",
|
|
"launcher/plutonium.exe": b"launcher-not-needed",
|
|
}
|
|
manifest = {
|
|
"product": "plutonium-core-prod",
|
|
"revision": revision,
|
|
"baseUrl": BASE_URL,
|
|
"files": [
|
|
{"name": name, "size": len(content), "hash": hashlib.sha1(content).hexdigest()}
|
|
for name, content in contents.items()
|
|
],
|
|
}
|
|
return manifest, contents
|
|
|
|
def opener(self, manifest, contents, calls=None):
|
|
calls = calls if calls is not None else []
|
|
|
|
def open_url(url):
|
|
calls.append(url)
|
|
if url == module.PLUTONIUM_MANIFEST_URL:
|
|
return fixtures.Response(json.dumps(manifest).encode())
|
|
self.assertTrue(url.startswith(BASE_URL), url)
|
|
name = next(entry["name"] for entry in manifest["files"] if url == BASE_URL + entry["hash"])
|
|
return fixtures.Response(contents[name])
|
|
|
|
return open_url
|
|
|
|
def file_url(self, content):
|
|
return BASE_URL + hashlib.sha1(content).hexdigest()
|
|
|
|
def prepare(self, revision=5354, contents=None):
|
|
manifest, contents = self.release(revision, contents)
|
|
calls = []
|
|
with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
result = self.agent.prepare_plutonium({"revision": revision})
|
|
return manifest, contents, calls, result
|
|
|
|
def cache_content(self):
|
|
root = self.agent.root / "agent" / "plutonium-cache"
|
|
pointer = json.loads((root / "current.json").read_text())
|
|
return root / pointer["key"] / "content"
|
|
|
|
def app(self, name="app"):
|
|
app = Path(self.temp.name) / name
|
|
app.mkdir(exist_ok=True)
|
|
return app
|
|
|
|
def assert_runtime(self, app, manifest, contents):
|
|
runtime = app / "runtime" / "plutonium"
|
|
self.assertEqual(json.loads((runtime / "cdn_info.json").read_text()), manifest)
|
|
for name, content in contents.items():
|
|
if name.startswith("launcher"):
|
|
self.assertFalse((runtime / name).exists())
|
|
else:
|
|
self.assertEqual((runtime / name).read_bytes(), content)
|
|
|
|
def test_cold_preparation_verifies_downloads_and_reports_inventory(self):
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
manifest, contents, calls, result = self.prepare()
|
|
expected_bytes = sum(len(content) for name, content in contents.items() if not name.startswith("launcher"))
|
|
self.assertEqual(result, {"revision": 5354, "bytes": expected_bytes})
|
|
self.assertEqual(self.agent.plutonium_cache_inventory(), result)
|
|
self.assertEqual(calls[0], module.PLUTONIUM_MANIFEST_URL)
|
|
self.assertEqual(set(calls[1:]), {self.file_url(content) for name, content in contents.items() if not name.startswith("launcher")})
|
|
self.assertEqual(json.loads((self.cache_content() / "cdn_info.json").read_text()), manifest)
|
|
self.assertEqual((self.agent.root / "agent" / "plutonium-cache").stat().st_mode & 0o077, 0)
|
|
|
|
def test_warm_preparation_only_fetches_manifest_and_dispatches(self):
|
|
manifest, contents, _, first = self.prepare()
|
|
calls = []
|
|
with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
result = self.agent.dispatch({"kind": "plutonium.prepare", "payload": {"revision": 5354}})
|
|
self.assertEqual(result, first)
|
|
self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL])
|
|
|
|
def test_refresh_downloads_changed_files_and_reuses_verified_unchanged_files(self):
|
|
_, contents, _, _ = self.prepare()
|
|
changed = {**contents, "bin/plutonium.dll": b"new-runtime", "storage/t6/new.ff": b"new-zone"}
|
|
manifest, changed, calls, result = self.prepare(5355, changed)
|
|
self.assertEqual(result["revision"], 5355)
|
|
self.assertEqual(set(calls[1:]), {self.file_url(changed["bin/plutonium.dll"]), self.file_url(changed["storage/t6/new.ff"])})
|
|
app = self.app()
|
|
self.assertTrue(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assert_runtime(app, manifest, changed)
|
|
|
|
def test_changed_manifest_is_refreshed_even_when_revision_number_is_unchanged(self):
|
|
_, contents, _, _ = self.prepare()
|
|
previous = self.cache_content()
|
|
manifest, changed, calls, result = self.prepare(5354, {**contents, "bin/plutonium.dll": b"corrected-runtime"})
|
|
self.assertEqual(result["revision"], 5354)
|
|
self.assertNotEqual(self.cache_content(), previous)
|
|
self.assertEqual(calls[1:], [self.file_url(changed["bin/plutonium.dll"])])
|
|
app = self.app()
|
|
self.assertTrue(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assert_runtime(app, manifest, changed)
|
|
|
|
def test_failed_refresh_preserves_the_older_complete_cache(self):
|
|
manifest, contents, _, result = self.prepare()
|
|
previous = self.cache_content()
|
|
next_manifest, next_contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"})
|
|
next_contents["bin/plutonium.dll"] = b"bad-runtime"
|
|
with patch.object(self.agent, "plutonium_open", side_effect=self.opener(next_manifest, next_contents)):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({"revision": 5355})
|
|
self.assertEqual(self.cache_content(), previous)
|
|
self.assertEqual(self.agent.plutonium_cache_inventory(), result)
|
|
app = self.app()
|
|
self.assertTrue(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assert_runtime(app, manifest, contents)
|
|
|
|
def test_interrupted_preparation_resumes_without_advertising_a_partial_cache(self):
|
|
manifest, contents = self.release()
|
|
open_url = self.opener(manifest, contents)
|
|
|
|
def interrupted_open(url):
|
|
if url == self.file_url(contents["bin/plutonium.dll"]):
|
|
raise urllib.error.URLError("simulated network interruption")
|
|
return open_url(url)
|
|
|
|
with patch.object(self.agent, "plutonium_open", side_effect=interrupted_open):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({})
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
self.assertFalse((self.agent.root / "agent/plutonium-cache/current.json").exists())
|
|
_, _, calls, result = self.prepare()
|
|
self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls)
|
|
self.assertEqual(result["revision"], manifest["revision"])
|
|
|
|
def test_agent_restart_reconciles_preparation_instead_of_accepting_an_older_cache(self):
|
|
_, contents, _, _ = self.prepare()
|
|
operation = {"id": "prepare-new-release", "kind": "plutonium.prepare", "payload": {"revision": 5355}}
|
|
self.agent.journal.accept(operation)
|
|
self.agent.journal.update(operation["id"], "running")
|
|
self.agent.journal.close()
|
|
self.agent = fixtures.Agent(self.config, self.docker)
|
|
self.assertEqual(self.agent.journal.get(operation["id"])["state"], "unknown")
|
|
manifest, contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"})
|
|
calls = []
|
|
with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
result = self.agent.execute(operation)
|
|
self.assertEqual(result["state"], "completed", result)
|
|
self.assertEqual(json.loads(result["result"])["revision"], 5355)
|
|
self.assertTrue(json.loads(result["result"])["reconciled"])
|
|
self.assertEqual(self.agent.plutonium_cache_inventory()["revision"], 5355)
|
|
self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL, self.file_url(contents["bin/plutonium.dll"])])
|
|
with patch.object(self.agent, "plutonium_open", side_effect=AssertionError("Completed operation repeated")):
|
|
self.assertEqual(self.agent.execute(operation)["result"], result["result"])
|
|
|
|
def test_heartbeat_defers_preparation_only_for_explicit_true(self):
|
|
self.assertFalse(self.agent.defer_plutonium_preparation)
|
|
for response, expected in (
|
|
({"deferPlutoniumPreparation": True}, True),
|
|
({}, False),
|
|
({"deferPlutoniumPreparation": False}, False),
|
|
({"deferPlutoniumPreparation": 1}, False),
|
|
({"deferPlutoniumPreparation": "true"}, False),
|
|
({"deferPlutoniumPreparation": None}, False),
|
|
):
|
|
self.agent.defer_plutonium_preparation = not expected
|
|
with self.subTest(response=response), patch.object(self.agent, "request", return_value=response), patch.object(self.agent, "inventory", return_value={}), patch.object(self.agent, "record_health"):
|
|
self.agent.heartbeat()
|
|
self.assertIs(self.agent.defer_plutonium_preparation, expected)
|
|
|
|
def assert_preparation_yields_and_resumes(self, *, restarted):
|
|
manifest, contents = self.release()
|
|
operation = {"id": "cooperative-prepare", "kind": "plutonium.prepare", "payload": {"revision": 5354}}
|
|
if restarted:
|
|
self.agent.journal.accept(operation)
|
|
self.agent.journal.update(operation["id"], "running")
|
|
self.agent.journal.close()
|
|
self.agent = fixtures.Agent(self.config, self.docker)
|
|
self.assertEqual(self.agent.journal.get(operation["id"])["state"], "unknown")
|
|
agent = self.agent
|
|
open_url = self.opener(manifest, contents)
|
|
|
|
class BusyResponse(fixtures.Response):
|
|
def read(self, size=-1):
|
|
chunk = super().read(size)
|
|
if chunk:
|
|
agent.defer_plutonium_preparation = True
|
|
return chunk
|
|
|
|
def become_busy_during_download(url):
|
|
if url == self.file_url(contents["bin/plutonium.dll"]):
|
|
return BusyResponse(contents["bin/plutonium.dll"])
|
|
return open_url(url)
|
|
|
|
with patch.object(self.agent, "plutonium_open", side_effect=become_busy_during_download):
|
|
result = self.agent.execute(operation)
|
|
self.assertEqual(result["state"], "accepted", result)
|
|
self.assertIn("waiting for other host operations", json.loads(result["result"])["message"])
|
|
self.assertIsNone(self.agent.executing_id)
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
self.assertTrue(self.agent.defer_plutonium_preparation)
|
|
# A yielded preparation must let other mutations finish before retrying.
|
|
self.create()
|
|
self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 1)["running"])
|
|
self.agent.defer_plutonium_preparation = False
|
|
calls = []
|
|
with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
result = self.agent.execute(operation)
|
|
self.assertEqual(result["state"], "completed", result)
|
|
self.assertEqual(result["id"], operation["id"])
|
|
self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls)
|
|
self.assertIn(self.file_url(contents["bin/plutonium.dll"]), calls)
|
|
self.assertEqual(self.agent.plutonium_cache_inventory()["revision"], 5354)
|
|
|
|
def test_busy_host_yields_mid_download_and_resumes_completed_files(self):
|
|
self.assert_preparation_yields_and_resumes(restarted=False)
|
|
|
|
def test_reconciled_preparation_yields_mid_download_and_resumes_same_operation(self):
|
|
self.assert_preparation_yields_and_resumes(restarted=True)
|
|
|
|
def test_resumption_removes_orphans_before_storage_check_and_publication(self):
|
|
manifest, contents = self.release()
|
|
raw = json.dumps(manifest).encode()
|
|
staging = self.agent.root / "agent/plutonium-cache" / (module.digest(raw) + ".partial/content")
|
|
bootstrapper = staging / BOOTSTRAPPER
|
|
bootstrapper.parent.mkdir(parents=True)
|
|
bootstrapper.write_bytes(contents[BOOTSTRAPPER])
|
|
orphans = [staging / ".download-abandoned", staging / "bin/.ec-abandoned", staging / "unlisted/extra.dat", staging / "cdn_info.json"]
|
|
for orphan in orphans:
|
|
orphan.parent.mkdir(parents=True, exist_ok=True)
|
|
orphan.write_bytes(b"orphaned bytes")
|
|
outside = Path(self.temp.name) / "outside-cache"
|
|
outside.mkdir()
|
|
(outside / "untouched").write_bytes(b"private outside bytes")
|
|
link = staging / "outside-link"
|
|
link.symlink_to(outside, target_is_directory=True)
|
|
original_disk_usage = module.shutil.disk_usage
|
|
|
|
def checked_storage(path):
|
|
for orphan in orphans:
|
|
self.assertFalse(orphan.exists(), "orphan must be removed before storage accounting")
|
|
self.assertFalse(link.is_symlink())
|
|
return original_disk_usage(path)
|
|
|
|
calls = []
|
|
with patch.object(module.shutil, "disk_usage", side_effect=checked_storage), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
result = self.agent.prepare_plutonium({})
|
|
content = self.cache_content()
|
|
expected = {name for name in contents if not name.startswith("launcher")}
|
|
actual = {path.relative_to(content).as_posix() for path in content.rglob("*") if path.is_file()}
|
|
self.assertEqual(actual, expected | {"cdn_info.json"})
|
|
self.assertEqual(result["bytes"], sum(len(contents[name]) for name in expected))
|
|
self.assertEqual(sum(path.stat().st_size for path in content.rglob("*") if path.is_file()), result["bytes"] + len(raw))
|
|
self.assertNotIn(self.file_url(contents[BOOTSTRAPPER]), calls)
|
|
self.assertEqual((outside / "untouched").read_bytes(), b"private outside bytes")
|
|
app = self.app()
|
|
self.assertTrue(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assert_runtime(app, manifest, contents)
|
|
self.assertFalse((app / "runtime/plutonium/unlisted/extra.dat").exists())
|
|
|
|
def test_public_downloads_reject_other_origins_and_never_send_host_credentials(self):
|
|
# Call the class method to bypass the fixture's default network blocker;
|
|
# the underlying opener is mocked, so no request reaches the network.
|
|
with patch.object(module.urllib.request, "build_opener") as build:
|
|
for url in (
|
|
"http://cdn.plutoniummod.com/updater/prod/info.json",
|
|
"https://attacker.example/updater/prod/info.json",
|
|
"https://cdn.plutoniummod.com.attacker.example/updater/prod/info.json",
|
|
"https://secret@cdn.plutoniummod.com/updater/prod/info.json",
|
|
"https://cdn.plutoniummod.com/updater/dev/info.json",
|
|
):
|
|
with self.subTest(url=url), self.assertRaises(AgentError):
|
|
fixtures.Agent.plutonium_open(self.agent, url)
|
|
build.assert_not_called()
|
|
fixtures.Agent.plutonium_open(self.agent, module.PLUTONIUM_MANIFEST_URL)
|
|
build.return_value.open.assert_called_once_with(module.PLUTONIUM_MANIFEST_URL, timeout=30)
|
|
handler = build.call_args.args[0]
|
|
for redirect in (module.PLUTONIUM_MANIFEST_URL, "https://attacker.example/collect"):
|
|
with self.subTest(redirect=redirect), self.assertRaises(AgentError):
|
|
handler.redirect_request(None, None, 302, "Found", {}, redirect)
|
|
|
|
def test_rejects_revision_downgrade_and_invalid_preparation_arguments(self):
|
|
self.prepare()
|
|
manifest, contents = self.release(5353)
|
|
with patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents)):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({"revision": 5354})
|
|
for payload in ({"revision": -1}, {"revision": True}, {"revision": "5354"},
|
|
{"minimumFreeBytes": -1}, {"minimumFreeBytes": True}, {"minimumFreeBytes": "10"}):
|
|
with self.subTest(payload=payload), self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium(payload)
|
|
|
|
def test_disk_reserve_failure_leaves_existing_cache_usable(self):
|
|
_, contents, _, result = self.prepare()
|
|
manifest, contents = self.release(5355, {**contents, "bin/plutonium.dll": b"new-runtime"})
|
|
calls = []
|
|
usage = type(module.shutil.disk_usage(self.agent.root))(10000, 9900, 100)
|
|
with patch.object(module.shutil, "disk_usage", return_value=usage), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({"minimumFreeBytes": 1000})
|
|
self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL])
|
|
self.assertEqual(self.agent.plutonium_cache_inventory(), result)
|
|
|
|
def test_malformed_manifest_never_publishes_a_cache(self):
|
|
good, contents = self.release()
|
|
malformed = [
|
|
None, [], {},
|
|
{**good, "product": "other-product"},
|
|
{**good, "revision": 0},
|
|
{**good, "revision": True},
|
|
{**good, "revision": "5354"},
|
|
{**good, "baseUrl": "http://cdn.plutoniummod.com/updater/prod/"},
|
|
{**good, "baseUrl": "https://attacker.example/"},
|
|
{**good, "files": []},
|
|
{**good, "files": {}},
|
|
{**good, "files": good["files"][1:]},
|
|
{**good, "files": good["files"] + [good["files"][0]]},
|
|
]
|
|
for manifest in malformed:
|
|
with self.subTest(manifest=manifest), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents)):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({})
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
|
|
def test_manifest_rejects_unsafe_paths_sizes_and_hashes_before_downloading(self):
|
|
good, contents = self.release()
|
|
bad_entries = [
|
|
{"name": name} for name in (
|
|
".", "../escape", "/absolute", "bin/../../escape", "bin//file", "bin/./file",
|
|
"bin\\file", "C:/escape", "bin/file\x00", "cdn_info.json",
|
|
)
|
|
] + [
|
|
{"size": size} for size in (-1, True, "5", None)
|
|
] + [
|
|
{"hash": value} for value in ("", "0" * 39, "z" * 40, None)
|
|
]
|
|
for override in bad_entries:
|
|
manifest = copy.deepcopy(good)
|
|
manifest["files"][1].update(override)
|
|
calls = []
|
|
with self.subTest(override=override), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, contents, calls)):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({})
|
|
self.assertEqual(calls, [module.PLUTONIUM_MANIFEST_URL])
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
|
|
def test_short_oversize_and_wrong_hash_downloads_are_not_published(self):
|
|
manifest, contents = self.release()
|
|
for body in (b"short", contents[BOOTSTRAPPER] + b"extra", b"X" * len(contents[BOOTSTRAPPER])):
|
|
with self.subTest(body=body), patch.object(self.agent, "plutonium_open", side_effect=self.opener(manifest, {**contents, BOOTSTRAPPER: body})):
|
|
with self.assertRaises(AgentError):
|
|
self.agent.prepare_plutonium({})
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
|
|
def test_seeds_are_private_and_preserve_existing_lobby_state(self):
|
|
manifest, contents, _, _ = self.prepare()
|
|
first, second = self.app("first"), self.app("second")
|
|
private = {
|
|
"configs/server.cfg": b"secret config",
|
|
"runtime/plutonium/plugins/private.dll": b"private plugin",
|
|
"runtime/plutonium/storage/t5/logs/games_mp.log": b"lobby log",
|
|
}
|
|
for name, body in private.items():
|
|
target = first / name
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(body)
|
|
self.assertTrue(self.agent.seed_plutonium(first, IMAGE))
|
|
self.assertTrue(self.agent.seed_plutonium(second, IMAGE))
|
|
self.assert_runtime(first, manifest, contents)
|
|
self.assert_runtime(second, manifest, contents)
|
|
for name, body in private.items():
|
|
self.assertEqual((first / name).read_bytes(), body)
|
|
self.assertFalse((second / name).exists())
|
|
first_file = first / "runtime/plutonium" / BOOTSTRAPPER
|
|
second_file = second / "runtime/plutonium" / BOOTSTRAPPER
|
|
cached_file = self.cache_content() / BOOTSTRAPPER
|
|
self.assertNotEqual(first_file.stat().st_ino, second_file.stat().st_ino)
|
|
self.assertNotEqual(first_file.stat().st_ino, cached_file.stat().st_ino)
|
|
first_file.write_bytes(b"container-updated")
|
|
self.assertEqual(second_file.read_bytes(), contents[BOOTSTRAPPER])
|
|
self.assertEqual(cached_file.read_bytes(), contents[BOOTSTRAPPER])
|
|
|
|
def test_absent_or_corrupt_cache_and_unsupported_image_fall_back(self):
|
|
app = self.app()
|
|
self.assertFalse(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists())
|
|
self.prepare()
|
|
for image in ("example.invalid/plutainer:latest", "ghcr.io/ayymoss/plutainer-other:latest", "ghcr.io/attacker/plutainer:latest"):
|
|
with self.subTest(image=image):
|
|
self.assertFalse(self.agent.seed_plutonium(app, image))
|
|
(self.cache_content() / BOOTSTRAPPER).write_bytes(b"broken-cache")
|
|
self.assertFalse(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists())
|
|
self.assertIsNone(self.agent.plutonium_cache_inventory())
|
|
|
|
def test_interrupted_seed_does_not_install_revision_marker_and_can_retry(self):
|
|
manifest, contents, _, _ = self.prepare()
|
|
app = self.app()
|
|
marker = app / "runtime/plutonium/cdn_info.json"
|
|
marker.parent.mkdir(parents=True)
|
|
marker.write_text(json.dumps({"revision": 5353}))
|
|
copy_file = module.managed_copy
|
|
copied = []
|
|
|
|
def interrupted_copy(root, name, source, *args, **kwargs):
|
|
self.assertFalse(marker.exists(), "revision marker must be installed only after all files")
|
|
if copied:
|
|
raise OSError("simulated full disk")
|
|
copied.append(name)
|
|
return copy_file(root, name, source, *args, **kwargs)
|
|
|
|
with patch.object(module, "managed_copy", side_effect=interrupted_copy):
|
|
with self.assertRaises((AgentError, OSError)):
|
|
self.agent.seed_plutonium(app, IMAGE)
|
|
self.assertFalse(marker.exists())
|
|
self.assertTrue(copied)
|
|
self.assertTrue(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assert_runtime(app, manifest, contents)
|
|
|
|
def test_failed_seed_prevents_container_start(self):
|
|
self.prepare()
|
|
with patch.object(module, "managed_copy", side_effect=OSError("simulated full disk")):
|
|
payload = {"game": "t5", "libraryId": self.library_id, "image": IMAGE, "port": 28960, "env": {}}
|
|
result = self.agent.execute(self.operation("instance.create", payload, "failed-seed"))
|
|
self.assertEqual(result["state"], "failed", result)
|
|
self.assertFalse(any(call[:2] == ("container", "start") for call in self.docker.calls))
|
|
self.assertFalse((self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium/cdn_info.json").exists())
|
|
|
|
def test_symlink_in_cache_cannot_seed_outside_data(self):
|
|
_, contents, _, _ = self.prepare()
|
|
cached = self.cache_content() / BOOTSTRAPPER
|
|
outside = Path(self.temp.name) / "outside"
|
|
outside.write_bytes(contents[BOOTSTRAPPER])
|
|
cached.unlink()
|
|
cached.symlink_to(outside)
|
|
app = self.app()
|
|
self.assertFalse(self.agent.seed_plutonium(app, IMAGE))
|
|
self.assertFalse((app / "runtime/plutonium/cdn_info.json").exists())
|
|
|
|
def test_create_applies_initial_overlay_after_seed_and_starts_without_cache(self):
|
|
self.create()
|
|
self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 1)["running"])
|
|
self.assertFalse((self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium/cdn_info.json").exists())
|
|
manifest, contents, _, _ = self.prepare()
|
|
payload = self.agent.spec(fixtures.INSTANCE, 1)
|
|
payload["port"] += 1
|
|
payload["files"].append({"path": "runtime/plutonium/" + BOOTSTRAPPER, "contentBase64": base64.b64encode(b"custom-overlay").decode()})
|
|
other_id = "00000000-0000-4000-8000-000000000002"
|
|
result = self.agent.execute({**self.operation("instance.create", payload, "second-create"), "instanceId": other_id})
|
|
self.assertEqual(result["state"], "completed", result)
|
|
app = self.agent.instance_dir(other_id, 1) / "app"
|
|
self.assertEqual((app / "runtime/plutonium" / BOOTSTRAPPER).read_bytes(), b"custom-overlay")
|
|
self.assertEqual(json.loads((app / "runtime/plutonium/cdn_info.json").read_text()), manifest)
|
|
self.assertEqual((app / "configs/server.cfg").read_bytes(), b"original")
|
|
|
|
def test_rebuild_seeds_the_new_generation_from_current_cache(self):
|
|
payload = self.create()
|
|
manifest, contents, _, _ = self.prepare()
|
|
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
|
result = self.agent.execute(self.operation("instance.rebuild", {**payload, "nextGeneration": 2}, "rebuild"))
|
|
self.assertEqual(result["state"], "completed", result)
|
|
self.assertTrue(self.agent.inspect(fixtures.INSTANCE, 2)["running"])
|
|
self.assert_runtime(self.agent.instance_dir(fixtures.INSTANCE, 2) / "app", manifest, contents)
|
|
|
|
def test_restart_preserves_the_instances_downloaded_runtime(self):
|
|
self.prepare()
|
|
self.create()
|
|
runtime_file = self.agent.instance_dir(fixtures.INSTANCE, 1) / "app/runtime/plutonium" / BOOTSTRAPPER
|
|
runtime_file.write_bytes(b"container-updated-runtime")
|
|
self.agent.rcon = lambda *_: {"ok": True, "players": 0}
|
|
with patch.object(self.agent, "seed_plutonium", side_effect=AssertionError("Restart must retain its runtime")):
|
|
result = self.agent.execute(self.operation("instance.restart", op_id="restart"))
|
|
self.assertEqual(result["state"], "completed", result)
|
|
self.assertEqual(runtime_file.read_bytes(), b"container-updated-runtime")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|