import json import hashlib import os from pathlib import Path import subprocess import tempfile import textwrap import unittest ROOT = Path(__file__).resolve().parent UPLOADER = ROOT / "w3cs-bundle-upload.sh" FAKE_RCLONE = r'''#!/usr/bin/env python3 import fnmatch import json import os from pathlib import Path import shutil import sys args = sys.argv[1:] with open(os.environ["FAKE_RCLONE_LOG"], "a", encoding="utf-8") as log: log.write(json.dumps(args) + "\n") for index, value in enumerate(args): if value in ("lsf", "copy", "sync"): command_index = index break else: raise SystemExit("fake rclone: no command") command = args[command_index] operands = args[command_index + 1:command_index + 3] loose_name = os.environ["FAKE_LOOSE_REMOTE_NAME"] pack_name = os.environ["FAKE_PACK_REMOTE_NAME"] loose_dir = Path(os.environ["FAKE_LOOSE_REMOTE_DIR"]) pack_dir = Path(os.environ["FAKE_PACK_REMOTE_DIR"]) loose_dir.mkdir(parents=True, exist_ok=True) pack_dir.mkdir(parents=True, exist_ok=True) def include_patterns(): return [args[i + 1] for i, value in enumerate(args[:-1]) if value == "--include"] def allowed(name): patterns = include_patterns() return not patterns or any(fnmatch.fnmatch(name, pattern) for pattern in patterns) if command == "lsf": if operands[0] != loose_name: raise SystemExit("fake rclone: unexpected lsf remote") for entry in sorted(loose_dir.iterdir()): if entry.is_file() and allowed(entry.name): print(f"{entry.name}\t{entry.stat().st_size}") raise SystemExit(0) source, destination = operands if command == "sync" and not destination.startswith( pack_name.rstrip("/") + "/by-worker/"): raise SystemExit("fake rclone: unsafe sync was invoked") if os.environ.get("FAKE_FAIL_LOOSE_COPY") == "1" and destination == loose_name: raise SystemExit(19) if os.environ.get("FAKE_FAIL_PACK_COPY") == "1" and destination == pack_name: raise SystemExit(20) if source == pack_name: source_dir = pack_dir destination_dir = Path(destination) elif destination == loose_name: source_dir = Path(source) destination_dir = loose_dir elif destination == pack_name: source_dir = Path(source) destination_dir = pack_dir elif destination.startswith(pack_name.rstrip("/") + "/"): source_dir = Path(source) relative = destination[len(pack_name.rstrip("/")):].lstrip("/") destination_dir = pack_dir / relative else: raise SystemExit(f"fake rclone: unexpected copy {source!r} -> {destination!r}") destination_dir.mkdir(parents=True, exist_ok=True) if command == "sync": source_names = {entry.name for entry in source_dir.iterdir() if entry.is_file() and allowed(entry.name)} for entry in destination_dir.iterdir(): if entry.is_file() and allowed(entry.name) and entry.name not in source_names: entry.unlink() for entry in source_dir.iterdir(): if entry.is_file() and allowed(entry.name): shutil.copy2(entry, destination_dir / entry.name) ''' class BundleUploaderTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.spool = self.root / "spool" self.markers = self.spool / "uploaded" self.loose_remote = self.root / "remote-loose" self.pack_remote = self.root / "remote-packs" self.tmp_root = self.root / "tmp" for path in (self.spool, self.markers, self.loose_remote, self.pack_remote, self.tmp_root): path.mkdir() self.rclone_log = self.root / "rclone.log" self.rclone_log.touch() self.fake_rclone = self.root / "rclone" self.fake_rclone.write_text(FAKE_RCLONE, encoding="utf-8") self.fake_rclone.chmod(0o755) self.builder_log = self.root / "builder.log" self.fake_builder = self.root / "pack-builder" self.fake_builder.write_text(textwrap.dedent(f"""\ #!/bin/sh printf 'called\\n' >> '{self.builder_log}' """), encoding="utf-8") self.fake_builder.chmod(0o755) self.env = { **os.environ, "W3CS_BUNDLE_SPOOL": str(self.spool), "W3CS_BUNDLE_REMOTE": "loose:", "W3CS_BUNDLE_PACK_REMOTE": "packs:", "W3CS_RCLONE": str(self.fake_rclone), "W3CS_RCLONE_CONFIG": str(self.root / "unused-rclone.conf"), "W3CS_PACK_BUILDER": str(self.fake_builder), "W3CS_PACK_PUBLISHER_ID": "worker-a.example", "W3CS_BUNDLE_TMP_ROOT": str(self.tmp_root), "FAKE_RCLONE_LOG": str(self.rclone_log), "FAKE_LOOSE_REMOTE_NAME": "loose:", "FAKE_PACK_REMOTE_NAME": "packs:", "FAKE_LOOSE_REMOTE_DIR": str(self.loose_remote), "FAKE_PACK_REMOTE_DIR": str(self.pack_remote), } def tearDown(self): self.temp.cleanup() def digest(self, character): return hashlib.sha256(character.encode()).hexdigest() def add_bundle(self, body=b"bundle"): digest = hashlib.sha256(body).hexdigest() path = self.spool / f"{digest}.bundle" path.write_bytes(body) return digest def mark(self, digest): (self.markers / digest).touch() def add_pack_manifest(self, replay="one", body=b"pack", *, entries=7, replay_entries=None, packed_bytes=None): pack_out = self.spool / "pack-out" pack_out.mkdir(exist_ok=True) object_name = hashlib.sha256(body).hexdigest() + ".zpack" (pack_out / object_name).write_bytes(body) manifest = { "version": 1, "key": "profile__map", "replayEntries": (entries if replay_entries is None else replay_entries), "packs": [{ "object": object_name, "plainBytes": 100, "packedBytes": len(body) if packed_bytes is None else packed_bytes, "entries": entries, }], } (pack_out / f"replay-{replay}.json").write_text( json.dumps(manifest), encoding="utf-8") return object_name, manifest def shard_manifest(self, publisher="worker-a.example", replay="one"): return (self.pack_remote / "by-worker" / publisher / f"replay-{replay}.json") def run_uploader(self, *, check=True, extra_env=None): env = {**self.env, **(extra_env or {})} return subprocess.run( ["bash", str(UPLOADER)], env=env, check=check, text=True, capture_output=True) def rclone_calls(self): return [json.loads(line) for line in self.rclone_log.read_text().splitlines()] def test_zero_pending_still_builds_and_uses_non_deleting_pack_copy(self): digest = self.add_bundle() self.mark(digest) stamp = self.markers / ".last-r2-reconcile" stamp.touch() os.utime(stamp, (1_000_000, 1_000_000)) object_name, _ = self.add_pack_manifest() self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "1000001", "W3CS_BUNDLE_RECONCILE_SECONDS": "86400", }) self.assertEqual("called\n", self.builder_log.read_text()) self.assertEqual(b"pack", (self.pack_remote / object_name).read_bytes()) self.assertTrue(self.shard_manifest().is_file()) self.assertFalse((self.pack_remote / "replay-one.json").exists()) shard_syncs = [call for call in self.rclone_calls() if "sync" in call and "by-worker" in " ".join(call)] self.assertEqual(1, len(shard_syncs)) self.assertIn("--delete-after", shard_syncs[0]) zpack_copies = [call for call in self.rclone_calls() if "copy" in call and "packs:" in call and "*.zpack" in call] self.assertEqual(1, len(zpack_copies)) self.assertIn("--size-only", zpack_copies[0]) self.assertFalse(any("copy" in call and "loose:" in call for call in self.rclone_calls())) def test_full_reconcile_reuploads_stale_marker(self): body = b"missing from R2" digest = self.add_bundle(body) self.mark(digest) result = self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "2000000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertIn("stale_markers=1", result.stdout) self.assertEqual(body, (self.loose_remote / f"{digest}.bin").read_bytes()) self.assertTrue((self.markers / digest).is_file()) loose_copy = next(call for call in self.rclone_calls() if "copy" in call and "loose:" in call) self.assertIn("--no-traverse", loose_copy) self.assertIn("--size-only", loose_copy) self.assertEqual(2_000_000, int((self.markers / ".last-r2-reconcile").stat().st_mtime)) def test_failed_repair_leaves_no_stale_marker_or_success_stamp(self): digest = self.add_bundle() self.mark(digest) result = self.run_uploader(check=False, extra_env={ "W3CS_NOW_EPOCH": "3000000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", "FAKE_FAIL_LOOSE_COPY": "1", }) self.assertNotEqual(0, result.returncode) self.assertFalse((self.markers / digest).exists()) self.assertFalse((self.markers / ".last-r2-reconcile").exists()) self.assertFalse(self.builder_log.exists()) def test_reconcile_confirms_remote_object_without_reupload(self): body = b"already remote" digest = self.add_bundle(body) (self.loose_remote / f"{digest}.bin").write_bytes(body) self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "4000000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertTrue((self.markers / digest).is_file()) loose_copies = [call for call in self.rclone_calls() if "copy" in call and "loose:" in call] self.assertEqual([], loose_copies) def test_reconcile_replaces_wrong_size_remote_object(self): body = b"correct immutable body" digest = self.add_bundle(body) self.mark(digest) (self.loose_remote / f"{digest}.bin").write_bytes(b"wrong") result = self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "4250000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertIn("stale_markers=1", result.stdout) self.assertEqual( body, (self.loose_remote / f"{digest}.bin").read_bytes()) self.assertTrue((self.markers / digest).is_file()) def test_reconcile_removes_stale_marker_without_local_body(self): digest = self.digest("e") self.mark(digest) result = self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "4500000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertIn("stale_markers=1", result.stdout) self.assertFalse((self.markers / digest).exists()) def test_two_publishers_write_independent_manifest_shards(self): _, first_manifest = self.add_pack_manifest("game", b"first", entries=10) self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "5000000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) first_path = self.shard_manifest("worker-a.example", "game") self.assertEqual(first_manifest, json.loads(first_path.read_text())) _, second_manifest = self.add_pack_manifest("game", b"second", entries=20) self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "5000001", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", "W3CS_PACK_PUBLISHER_ID": "worker-b.example", }) self.assertEqual(first_manifest, json.loads(first_path.read_text())) self.assertEqual(second_manifest, json.loads( self.shard_manifest("worker-b.example", "game").read_text())) self.assertFalse((self.pack_remote / "replay-game.json").exists()) def test_missing_publisher_refuses_manifest_publication(self): self.add_pack_manifest() result = self.run_uploader(check=False, extra_env={ "W3CS_NOW_EPOCH": "5500000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", "W3CS_PACK_PUBLISHER_ID": "", }) self.assertNotEqual(0, result.returncode) self.assertIn("invalid or missing W3CS_PACK_PUBLISHER_ID", result.stderr) self.assertFalse((self.pack_remote / "by-worker").exists()) def test_manifest_with_missing_or_wrong_size_pack_is_rejected(self): object_name, _ = self.add_pack_manifest("missing", b"gone") (self.spool / "pack-out" / object_name).unlink() self.add_pack_manifest("wrong", b"wrong-size", packed_bytes=999) result = self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "6000000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertIn("rejected=2", result.stdout) self.assertFalse(self.shard_manifest(replay="missing").exists()) self.assertFalse(self.shard_manifest(replay="wrong").exists()) def test_manifest_cannot_claim_more_replay_entries_than_its_packs(self): self.add_pack_manifest("impossible", entries=7, replay_entries=8) result = self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "6250000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertIn("rejected=1", result.stdout) self.assertFalse(self.shard_manifest(replay="impossible").exists()) def test_failed_zpack_upload_publishes_no_manifest(self): self.add_pack_manifest() result = self.run_uploader(check=False, extra_env={ "W3CS_NOW_EPOCH": "6500000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", "FAKE_FAIL_PACK_COPY": "1", }) self.assertNotEqual(0, result.returncode) self.assertFalse(self.shard_manifest().exists()) def test_zero_pending_run_restores_deleted_remote_pack_and_shard(self): object_name, manifest = self.add_pack_manifest() self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "7000000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) (self.pack_remote / object_name).unlink() self.shard_manifest().unlink() self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "7000001", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertEqual(b"pack", (self.pack_remote / object_name).read_bytes()) self.assertEqual(manifest, json.loads(self.shard_manifest().read_text())) def test_worker_shard_sync_removes_only_its_retired_manifest(self): self.add_pack_manifest("current") own_dir = self.pack_remote / "by-worker" / "worker-a.example" other_dir = self.pack_remote / "by-worker" / "worker-b.example" own_dir.mkdir(parents=True) other_dir.mkdir(parents=True) (own_dir / "replay-retired.json").write_text("{}") (other_dir / "replay-retired.json").write_text("{}") self.run_uploader(extra_env={ "W3CS_NOW_EPOCH": "7500000", "W3CS_BUNDLE_RECONCILE_SECONDS": "0", }) self.assertFalse((own_dir / "replay-retired.json").exists()) self.assertTrue((other_dir / "replay-retired.json").exists()) self.assertTrue((own_dir / "replay-current.json").exists()) if __name__ == "__main__": unittest.main()