"""Safety and reproducibility tests for AirDef's asset generation pipeline.""" import hashlib import importlib.util import json import subprocess import os import sys import tempfile import unittest from pathlib import Path from PIL import Image TOOLS = Path(__file__).resolve().parent sys.path.insert(0, str(TOOLS)) from airdef_assets import catalog from airdef_assets import pipeline from airdef_assets import icons import gen_tex SPEC = importlib.util.spec_from_file_location("gen_models", TOOLS / "gen_models.py") gen_models = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(gen_models) class ContentCatalogTests(unittest.TestCase): def test_catalog_rust_entries_are_current_and_registered_items_match_assets(self): root = TOOLS.parent subprocess.run([sys.executable, str(TOOLS / "gen_catalog.py"), "--check"], check=True) language = json.loads((root / "airdef-mod/assets/airdef/lang/ru_ru.json").read_text(encoding="utf-8")) for name, title, _ in icons.ITEMS: self.assertEqual(language[f"item.airdef.{name}"], title) self.assertTrue((root / "airdef-mod/data/airdef/horizon/item" / (name + ".json")).is_file()) def test_a_new_unit_kind_needs_only_catalog_entries(self): # The validator must not carry its own list of kind names: a hypothetical unit is # accepted from a specs row plus its content entry, with no code change. data = json.loads(json.dumps(catalog.load_specs())) content = json.loads(json.dumps(catalog.load())) template = next(entry for entry in data["units"] if entry["gun"] is not None) added = dict(template, kind="TestGun", key="testgun", name="Test gun") data["units"].append(added) content["units"].append({"kind": "TestGun", "item": None}) events = {entry["event"] for entry in catalog.load_sounds()} catalog.validate_specs(data, content, events) class ManifestTests(unittest.TestCase): def test_manifest_rejects_traversal_and_duplicate_paths(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "manifest.txt" path.write_text("0" * 64 + " ../outside\n", encoding="utf-8") with self.assertRaises(ValueError): pipeline.read_manifest(path) path.write_text("0" * 64 + " safe.txt\n" + "1" * 64 + " safe.txt\n", encoding="utf-8") with self.assertRaises(ValueError): pipeline.read_manifest(path) def test_plan_protects_untracked_and_modified_files_and_keeps_edited_obsolete_file(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "tracked.txt").write_text("edited", encoding="utf-8") (root / "obsolete.txt").write_text("artist edit", encoding="utf-8") (root / "untracked.txt").write_text("user", encoding="utf-8") old = { "tracked.txt": hashlib.sha256(b"old generated").hexdigest(), "obsolete.txt": hashlib.sha256(b"old generated").hexdigest(), } plan = pipeline.make_plan(root, old, {"tracked.txt": hashlib.sha256(b"new").hexdigest(), "untracked.txt": hashlib.sha256(b"output").hexdigest()}) self.assertEqual(plan.conflict, ["tracked.txt", "untracked.txt"]) self.assertEqual(plan.keep, ["obsolete.txt"]) self.assertEqual(plan.remove, []) def test_install_rolls_back_files_and_manifest_after_replace_error(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) / "runtime" stage = Path(tmp) / "stage" root.mkdir() stage.mkdir() manifest = root / "generated-manifest.txt" manifest.write_text("old manifest\n", encoding="utf-8") for name in ("a.txt", "b.txt"): (root / name).write_text("old " + name, encoding="utf-8") (stage / name).write_text("new " + name, encoding="utf-8") old = {name: pipeline.sha256(root / name) for name in ("a.txt", "b.txt")} new = {name: pipeline.sha256(stage / name) for name in ("a.txt", "b.txt")} plan = pipeline.make_plan(root, old, new) real_replace = os.replace calls = 0 def fail_once(src, dst): nonlocal calls calls += 1 if calls == 3: raise OSError("simulated write failure") real_replace(src, dst) with self.assertRaisesRegex(OSError, "simulated"): pipeline.install(root, stage, manifest, plan, new, replace=fail_once) self.assertEqual((root / "a.txt").read_text(), "old a.txt") self.assertEqual((root / "b.txt").read_text(), "old b.txt") self.assertEqual(manifest.read_text(), "old manifest\n") class ReproducibilityTests(unittest.TestCase): def test_texture_renders_are_seed_deterministic(self): first = gen_tex.camo("nato_green", 1234, 32) second = gen_tex.camo("nato_green", 1234, 32) self.assertEqual(first.tobytes(), second.tobytes()) def test_authored_override_is_merged_without_changing_source(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) source = root / "source-assets" stage = root / "stage" authored = source / "assets/airdef/textures/item/custom.png" authored.parent.mkdir(parents=True) authored.write_bytes(b"authored") stage.mkdir() pipeline.overlay_sources(stage, source) self.assertEqual((stage / "assets/airdef/textures/item/custom.png").read_bytes(), b"authored") self.assertEqual(authored.read_bytes(), b"authored") def test_full_pipeline_installs_valid_icon_override_and_rejects_bad_one_without_mutation(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) / "mod" root.mkdir() source = root / "source-assets" manifest = root / "generated-manifest.txt" icon_source = source / "assets/airdef/textures/item/deploy_gepard.png" icon_source.parent.mkdir(parents=True) self.assertTrue(gen_models.regenerate(root, source, manifest)) previous_icon = (root / "assets/airdef/textures/item/deploy_gepard.png").read_bytes() previous_manifest = manifest.read_bytes() self.assertFalse(gen_models.regenerate(root, source, manifest)) Image.new("RGBA", (16, 16), (12, 34, 56, 255)).save(icon_source) authored_bytes = icon_source.read_bytes() self.assertTrue(gen_models.regenerate(root, source, manifest)) installed = root / "assets/airdef/textures/item/deploy_gepard.png" self.assertEqual(installed.read_bytes(), authored_bytes) self.assertEqual(icon_source.read_bytes(), authored_bytes) self.assertNotEqual(installed.read_bytes(), previous_icon) installed_before_bad_override = installed.read_bytes() manifest_before_bad_override = manifest.read_bytes() Image.new("RGBA", (20, 16), (255, 0, 0, 255)).save(icon_source) invalid_source = icon_source.read_bytes() with self.assertRaisesRegex(ValueError, "не квадратная"): gen_models.regenerate(root, source, manifest) self.assertEqual(installed.read_bytes(), installed_before_bad_override) self.assertEqual(manifest.read_bytes(), manifest_before_bad_override) self.assertEqual(icon_source.read_bytes(), invalid_source) self.assertNotEqual(previous_manifest, manifest_before_bad_override) def test_icon_catalog_has_stable_explicit_renderer_api(self): entry = next(item for item in icons.ITEMS if item[0] == "deploy_gepard") first = icons.render_icon(entry[0], entry[2], 1234) second = icons.render_icon(entry[0], entry[2], 1234) self.assertEqual(first.tobytes(), second.tobytes()) def test_full_generated_tree_is_reproducible(self): with tempfile.TemporaryDirectory() as tmp: outputs = [] for name in ("first", "second"): root = Path(tmp) / name root.mkdir() gen_models.generate(str(root)) outputs.append({ str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest() for path in sorted(root.rglob("*")) if path.is_file() }) self.assertEqual(outputs[0], outputs[1]) if __name__ == "__main__": unittest.main()