#!/usr/bin/env python3 """Storage-routing tests using disposable repositories, without network access.""" from pathlib import Path import os import shlex import subprocess import sys import tempfile import unittest import storage_policy as policy POLICY = Path(policy.__file__).resolve() class ContentPolicyTests(unittest.TestCase): def test_code_and_readable_shader_source(self): for name in ["src/main.rs", "shaders/wall.wgsl", "docs/evidence/files.csv", "config.ron"]: policy.check_code(name, "// source and metadata ✓\n".encode()) def test_payload_paths_even_for_text_assets(self): for name in ["assets/mesh.json", "crates/app/assets/settings.ron", ".cache/foo.txt", "captures/frame.txt"]: with self.assertRaises(ValueError, msg=name): policy.check_code(name, b"{}") def test_payload_extensions_outside_asset_directory(self): for name in ["foo.PNG", "src/model.glb", "texture.texture", "docs/example.svg", "plugin.so.1", "game.pk3", "cgame.qvm", "map.bsp", "map.aas", "player.MD3", "original.shader", "library.dylib", "app.icns", "font.dat"]: with self.assertRaises(ValueError, msg=name): policy.check_code(name, b"innocent looking text") def test_binary_invalid_encoding_and_oversize(self): for data in [b"a\0b", b"\xff", b"x" * (policy.MAX_CODE_BYTES + 1)]: with self.assertRaises(ValueError): policy.check_code("src/data.rs", data) def test_asset_pointer_is_required(self): pointer = b"version https://git-lfs.github.com/spec/v1\noid sha256:" + b"a" * 64 + b"\nsize 10\n" policy.check_asset("mesh.json", pointer) policy.check_asset("assets/README.md", pointer) policy.check_asset("assets/empty.skin", b"") for data in [b"{}", pointer + b"hidden payload"]: with self.assertRaises(ValueError): policy.check_asset("mesh.json", data) with self.assertRaises(ValueError): policy.check_code("data.txt", pointer) def test_nested_asset_readme_is_not_control_metadata(self): with self.assertRaises(ValueError): policy.check_asset("original/README.md", b"original game text\n") class GitIntegrationTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.env = {**os.environ, "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.invalid", "GIT_COMMITTER_NAME": "Test", "GIT_COMMITTER_EMAIL": "test@example.invalid"} # Do not inherit a parent hook's repository/index environment. for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_COMMON_DIR"): self.env.pop(key, None) self.git("init", "-q") def tearDown(self): self.temp.cleanup() def run_command(self, *args, input=None, success=True): result = subprocess.run(args, cwd=self.root, env=self.env, input=input, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if success: self.assertEqual(result.returncode, 0, result.stderr.decode(errors="replace")) else: self.assertNotEqual(result.returncode, 0) return result def git(self, *args, **kwargs): return self.run_command("git", *args, **kwargs) def install_code_filter(self): (self.root / ".gitattributes").write_text((POLICY.parent.parent / ".gitattributes").read_text()) command = f"{shlex.quote(sys.executable)} {shlex.quote(str(POLICY))} clean -- %f" self.git("config", "filter.nanobrush-code.clean", command) self.git("config", "filter.nanobrush-code.smudge", "cat") self.git("config", "filter.nanobrush-code.required", "true") self.git("add", ".gitattributes") def test_git_add_preserves_source_and_quoted_filename(self): self.install_code_filter() name = "space ' and dollar $.rs" data = b"fn main() {}\n" (self.root / name).write_bytes(data) self.git("add", "--", name) self.assertEqual(self.git("show", ":" + name).stdout, data) self.run_command(sys.executable, str(POLICY), "check") def test_clean_filter_rejects_asset_and_missing_driver(self): self.install_code_filter() (self.root / "image.png").write_bytes(b"small") self.git("add", "image.png", success=False) (self.root / "main.rs").write_text("fn main() {}\n") self.git("config", "filter.nanobrush-code.clean", "missing-nanobrush-filter-command") self.git("add", "main.rs", success=False) def test_index_guard_catches_bypassed_filter_and_attribute_override(self): self.install_code_filter() (self.root / "data.txt").write_bytes(b"binary\0data") self.git("-c", "filter.nanobrush-code.clean=cat", "add", "data.txt") self.run_command(sys.executable, str(POLICY), "check", success=False) self.git("-c", "filter.nanobrush-code.clean=cat", "rm", "--cached", "data.txt") (self.root / ".gitattributes").write_text("* -filter\n") self.git("add", ".gitattributes") self.run_command(sys.executable, str(POLICY), "check", success=False) def test_oversize_bypassed_blob_does_not_block_batch_reader(self): self.install_code_filter() (self.root / "large.txt").write_bytes(b"x" * (policy.MAX_CODE_BYTES + 1)) self.git("-c", "filter.nanobrush-code.clean=cat", "add", "large.txt") self.run_command(sys.executable, str(POLICY), "check", success=False) def test_push_guard_catches_payload_deleted_in_later_commit(self): self.install_code_filter() (self.root / "data.txt").write_bytes(b"binary\0data") self.git("-c", "filter.nanobrush-code.clean=cat", "add", "data.txt") self.git("-c", "filter.nanobrush-code.clean=cat", "commit", "-qm", "payload") self.git("-c", "filter.nanobrush-code.clean=cat", "rm", "data.txt") self.git("commit", "-qm", "remove payload") self.run_command(sys.executable, str(POLICY), "check") oid = self.git("rev-parse", "HEAD").stdout.decode().strip() update = f"refs/heads/main {oid} refs/heads/main {'0' * 40}\n".encode() self.run_command(sys.executable, str(POLICY), "push", input=update, success=False) def install_mixed_filters(self): self.install_code_filter() self.git("lfs", "install", "--local", "--skip-repo") (self.root / "assets").mkdir() def bypass_lfs(self): self.git("config", "--unset", "filter.lfs.process") self.git("config", "filter.lfs.clean", "cat") def test_lfs_stores_text_asset_as_pointer_and_keeps_working_bytes(self): self.install_mixed_filters() data = b'{"positions": [1, 2, 3]}\n' (self.root / "assets/mesh.json").write_bytes(data) self.git("add", "assets/mesh.json") pointer = self.git("show", ":assets/mesh.json").stdout self.assertTrue(policy.POINTER.fullmatch(pointer)) self.assertEqual((self.root / "assets/mesh.json").read_bytes(), data) self.run_command(sys.executable, str(POLICY), "check") self.bypass_lfs() self.git("add", "--renormalize", "assets/mesh.json") self.run_command(sys.executable, str(POLICY), "check", success=False) def test_actual_attributes_route_all_assets_and_keep_code_literal(self): self.install_mixed_filters() (self.root / "README.md").write_text("# Code repository\n") (self.root / "assets/extracted").mkdir() payloads = { "assets/extracted/empty.skin": b"", "assets/README.md": b"asset documentation\n", "assets/extracted/README.md": b"original attribution\n", "assets/extracted/default.cfg": b"seta snaps 20\n", "assets/extracted/wall.shader": b"textures/wall { { map wall.tga } }\n", "assets/extracted/game.exe": b"MZ\0game payload", } for name, data in payloads.items(): (self.root / name).write_bytes(data) self.git("add", ".") self.assertEqual(self.git("show", ":README.md").stdout, b"# Code repository\n") for name, data in payloads.items(): pointer = self.git("show", ":" + name).stdout self.assertTrue(policy.POINTER.fullmatch(pointer) or (data == b"" and pointer == b""), name) self.assertEqual((self.root / name).read_bytes(), data) self.run_command(sys.executable, str(POLICY), "check") def test_asset_history_rejects_payload_even_after_deletion(self): self.install_mixed_filters() self.bypass_lfs() (self.root / "assets/config.txt").write_text("original game asset\n") self.git("add", ".") self.git("commit", "-qm", "raw asset") self.git("rm", "assets/config.txt") self.git("commit", "-qm", "remove asset") self.run_command(sys.executable, str(POLICY), "check") oid = self.git("rev-parse", "HEAD").stdout.decode().strip() update = f"refs/heads/main {oid} refs/heads/main {'0' * 40}\n".encode() self.run_command(sys.executable, str(POLICY), "push", input=update, success=False) def test_nested_attributes_cannot_exempt_asset_payload(self): self.install_mixed_filters() (self.root / "assets/.gitattributes").write_text("* -filter\n") (self.root / "assets/data.txt").write_text("game data\n") self.git("add", ".") self.run_command(sys.executable, str(POLICY), "check", success=False) def test_root_and_nested_references_are_filtered_but_tests_remain_text(self): self.install_mixed_filters() payloads = {"references/download.txt": b"original small text\r\n", "docs/references/old.exe": b"MZ\0binary"} for name, data in payloads.items(): path = self.root / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) (self.root / "tests").mkdir() (self.root / "tests/test_example.py").write_text("assert 1 + 1 == 2\n") self.git("add", ".") for name, data in payloads.items(): self.assertTrue(policy.POINTER.fullmatch(self.git("show", ":" + name).stdout)) self.assertEqual((self.root / name).read_bytes(), data) self.assertEqual(self.git("show", ":tests/test_example.py").stdout, b"assert 1 + 1 == 2\n") self.run_command(sys.executable, str(POLICY), "check") def test_symlink_is_rejected(self): self.install_mixed_filters() (self.root / "link").symlink_to("README.md") self.git("add", "link") self.run_command(sys.executable, str(POLICY), "check", success=False) def test_missing_local_setup_is_rejected(self): self.install_mixed_filters() self.run_command(sys.executable, str(POLICY), "check", "--require-setup", success=False) if __name__ == "__main__": unittest.main()