| |
| """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"} |
| |
| 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() |
|
|
