"""저장소 계약의 회귀 검사. 실제 렌더링·승인·미디어 검수는 수행하지 않는다.""" from copy import deepcopy import json from pathlib import Path import re import tomllib import unittest from jsonschema import Draft202012Validator ROOT = Path(__file__).resolve().parents[1] PROFILE = dict(engine="BLENDER_EEVEE", device="GPU", fps=50, width=1080, height=1920, samples=32, seed=1) BLEND = dict(path="examples/blender-contract/preview.blend", verified=False, sha256=None, blender_version=None) IMAGE = dict(path="examples/blender-contract/preview.png", kind="image", verified=False, duration_ms=None, width=None, height=None) def read(path): return json.loads((ROOT / path).read_text(encoding="utf-8")) def artifact(stage, payload): return dict(schema_version="1.1", project_id="contract-example", stage_id=stage, revision=1, inputs=[], payload=payload) def samples(): frames = [dict(id="c1-" + pos, cut_id="c1", position=pos, prompt="상태 설명", reference_ids=["water"], scene_name="Water", frame_index=index) for pos, index in [("start", 0), ("end", 200)]] return { "08-scenes": artifact("08-scenes", dict(scenes=[], connections=[], preview=None)), "10-references": artifact("10-references", dict(request=dict(engine="Blender", style_reference="references/1.png", render_profile=deepcopy(PROFILE), items=[ dict(id="water", kind="object", entity_id=None, prompt="물방울")]), result=None)), "11-images": artifact("11-images", dict(request=dict(engine="Blender", scene_path=BLEND["path"], render_profile=deepcopy(PROFILE), frames=frames), result=None)), "12-motion": artifact("12-motion", dict(request=dict(engine="Blender", scene_path=BLEND["path"], render_profile=deepcopy(PROFILE), clips=[dict(cut_id="c1", start_frame_id="c1-start", end_frame_id="c1-end", duration_ms=4000, motion_prompt="둥글게", scene_name="Water", start_frame=0, end_frame=200)]), result=None)), } def semantic_errors(data, context, complete=False): """1.1 구조 검사 후 사용. context={stage_id: artifact}; 파일·승인은 별도 확인.""" if data["schema_version"] != "1.1": return [] errors = [] stage = data["stage_id"] payload = data["payload"] def require(condition, message): if not condition: errors.append(message) def indexed(items, key): ids = [item[key] for item in items] require(all(ids) and len(ids) == len(set(ids)), "빈 식별자 또는 중복: " + key) return {item[key]: item for item in items} def walk(value): if isinstance(value, dict): for key, child in value.items(): if key in ("path", "scene_path") and isinstance(child, str) and child: parts = child.replace("\\", "/").split("/") require(not child.startswith(("/", "\\")) and ":" not in child and ".." not in parts, "루트 상대 경로 필요: " + child) if "kind" in value: require(not child.lower().endswith(".blend"), ".blend는 media가 아니다") walk(child) elif isinstance(value, list): for child in value: walk(child) walk(payload) request = payload.get("request", {}) preview = payload.get("preview") profile = request.get("render_profile") or (preview or {}).get("render_profile") if profile: require(profile["width"] * 16 == profile["height"] * 9, "RenderProfile은 9:16 필요") cuts = context.get("07-cuts", {}).get("payload", {}).get("cuts", []) cut_ids = {cut["id"] for cut in cuts} if stage == "08-scenes" and preview: indexed(preview["samples"], "id") require(bool(preview["samples"]), "대표 샘플 필요") for sample in preview["samples"]: require(set(sample["cut_ids"]) <= cut_ids, "대표 샘플의 알 수 없는 컷") require(sample["media"]["verified"], "대표 미디어 미검증") require(preview["scene_file"]["verified"], "대표 .blend 미검증") if stage == "10-references": indexed(request["items"], "id") entities = context.get("09-entities", {}).get("payload", {}) for item in request["items"]: if item["kind"] in ("character", "background"): bucket = "characters" if item["kind"] == "character" else "backgrounds" require(item["entity_id"] in {x["id"] for x in entities.get(bucket, [])}, "알 수 없는 entity_id") if stage == "11-images": frames = indexed(request["frames"], "id") refs = context.get("10-references", {}).get("payload", {}).get("request", {}).get("items", []) reference_ids = {item["id"] for item in refs} pairs = {} for frame in frames.values(): require(frame["cut_id"] in cut_ids, "프레임의 알 수 없는 컷") require(set(frame["reference_ids"]) <= reference_ids, "알 수 없는 reference_id") pairs.setdefault(frame["cut_id"], []).append(frame) require(set(pairs) == cut_ids, "컷별 start/end 누락") for pair in pairs.values(): require(len(pair) == 2 and {x["position"] for x in pair} == {"start", "end"}, "start/end 중복 또는 누락") require(len({x["scene_name"] for x in pair}) == 1, "같은 컷은 같은 Scene 필요") if stage == "12-motion": clips = indexed(request["clips"], "cut_id") require(set(clips) == cut_ids, "클립 컷 누락 또는 추가") previous = context.get("11-images", {}).get("payload", {}) result = previous.get("result") require(bool(result) and result["scene_file"]["verified"], "11 실제 .blend 필요") if result: require(request["scene_path"] == result["scene_file"]["path"], "11 원본 경로 불일치") require(request["render_profile"] == previous.get("request", {}).get("render_profile"), "11 RenderProfile 불일치") frames = {x["id"]: x for x in previous.get("request", {}).get("frames", [])} for clip in clips.values(): require(clip["end_frame"] > clip["start_frame"], "프레임 범위는 양수 필요") for position in ("start", "end"): frame = frames.get(clip[position + "_frame_id"]) require(bool(frame), "알 수 없는 frame_id") if frame: require(frame["cut_id"] == clip["cut_id"] and frame["position"] == position and frame["scene_name"] == clip["scene_name"] and frame["frame_index"] == clip[position + "_frame"], "프레임 상태 참조 불일치") cut = next((x for x in cuts if x["id"] == clip["cut_id"]), None) if cut: require(clip["duration_ms"] == cut["end_ms"] - cut["start_ms"], "컷 실측 길이 불일치") require(clip["duration_ms"] is not None and (clip["end_frame"] - clip["start_frame"]) * 1000 == clip["duration_ms"] * request["render_profile"]["fps"], "프레임 수와 컷 길이 불일치") if complete and stage in ("10-references", "11-images", "12-motion"): result = payload["result"] require(result is not None, "result=null은 완료가 아니다") if result: key, requested_key, returned_key = { "10-references": ("items", "id", "reference_id"), "11-images": ("frames", "id", "frame_id"), "12-motion": ("clips", "cut_id", "cut_id"), }[stage] returned = indexed(result[key], returned_key) require(bool(request[key]) and set(returned) == {x[requested_key] for x in request[key]}, "요청/결과 누락 또는 추가") require(all(x["media"]["verified"] for x in result[key]), "결과 미디어 미검증") if stage != "12-motion": require(result["scene_file"]["verified"], "결과 .blend 미검증") return errors class ContractTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.validators = {} for name in ("project", "artifact"): schema = read("contracts/" + name + ".schema.json") Draft202012Validator.check_schema(schema) cls.validators[name] = Draft202012Validator(schema) def valid(self, data): self.validators["artifact" if "stage_id" in data else "project"].validate(data) def invalid(self, data): self.assertFalse(self.validators["artifact" if "stage_id" in data else "project"].is_valid(data)) def test_legacy_templates_examples_and_production(self): count = 0 for folder in ("templates", "examples", "projects"): for path in (ROOT / folder).rglob("*.json"): if "node_modules" in path.parts or ".cache" in path.parts: continue data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, dict) and "schema_version" in data and "project_id" in data: self.valid(data) count += 1 self.assertGreaterEqual(count, 30) def test_role_skill_and_document_links(self): agents = list((ROOT / ".codex/agents").glob("*.toml")) self.assertEqual(len(agents), 8) for path in agents: role = tomllib.loads(path.read_text(encoding="utf-8")) self.assertEqual(set(role), {"name", "description", "developer_instructions"}) self.assertEqual(role["name"], path.stem) skills = list((ROOT / ".agents/skills").glob("*/SKILL.md")) self.assertEqual(len(skills), 15) documents = [ROOT / "AGENTS.md", ROOT / "CLAUDE.md", *list((ROOT / "docs").glob("*.md")), ROOT / "examples/blender-contract/README.md", *skills] for path in documents: text = path.read_text(encoding="utf-8") for target in re.findall(r"\]\(([^)]+)\)", text): if "://" not in target and not target.startswith("#"): self.assertTrue((path.parent / target.split("#")[0]).resolve().exists(), (path, target)) def test_blender_contracts(self): project = read("templates/project.json") project["schema_version"] = "1.1" project["settings"].pop("image_model", None) project["settings"]["visual_engine"] = "Blender" self.valid(project) for data in samples().values(): self.valid(data) def test_wrong_version_engine_and_extra_fields(self): for data in samples().values(): for version in ("1.0", "2.0"): bad = deepcopy(data) bad["schema_version"] = version self.invalid(bad) if "request" in data["payload"]: bad = deepcopy(data) bad["payload"]["request"]["engine"] = "gpt-image-2.5-sunburst" self.invalid(bad) bad["payload"]["request"]["engine"] = "Blender" bad["payload"]["request"]["model"] = "Blender" self.invalid(bad) def test_verified_blend_requires_hash_and_version(self): data = samples()["08-scenes"] data["payload"]["preview"] = dict(scene_file=deepcopy(BLEND), render_profile=deepcopy(PROFILE), samples=[dict(id="p1", cut_ids=["c1"], media=deepcopy(IMAGE))]) self.valid(data) blend = data["payload"]["preview"]["scene_file"] blend["verified"] = True self.invalid(data) blend.update(sha256="a" * 64, blender_version="5.2.2 LTS") self.valid(data) for key, value in [("sha256", "A" * 64), ("blender_version", ""), ("path", "")]: bad = deepcopy(data) bad["payload"]["preview"]["scene_file"][key] = value self.invalid(bad) bad = deepcopy(data) bad["payload"]["preview"]["samples"][0]["media"]["kind"] = "blend" self.invalid(bad) def test_profile_and_duration_constraints(self): for key, value in [("engine", "UNKNOWN"), ("device", "CPU"), ("fps", 0), ("samples", 0), ("width", 0), ("height", 0), ("seed", -1)]: bad = samples()["11-images"] bad["payload"]["request"]["render_profile"][key] = value self.invalid(bad) for duration in (0, -1, 5001): bad = samples()["12-motion"] bad["payload"]["request"]["clips"][0]["duration_ms"] = duration self.invalid(bad) cycles = samples()["11-images"] cycles["payload"]["request"]["render_profile"].update(engine="CYCLES", device="CPU") self.valid(cycles) def test_result_shapes_and_missing_fields(self): fixtures = samples() fixtures["10-references"]["payload"]["result"] = dict(scene_file=deepcopy(BLEND), items=[dict(reference_id="water", object_name="Droplet", media=deepcopy(IMAGE))]) fixtures["11-images"]["payload"]["result"] = dict(scene_file=deepcopy(BLEND), frames=[dict(frame_id="c1-start", media=deepcopy(IMAGE))]) fixtures["12-motion"]["payload"]["result"] = dict(clips=[dict(cut_id="c1", media=dict(IMAGE, kind="video", path="examples/blender-contract/c1.mp4"))]) for data in fixtures.values(): self.valid(data) if "request" in data["payload"]: for key in data["payload"]["request"]: bad = deepcopy(data) del bad["payload"]["request"][key] self.invalid(bad) def test_semantic_references_and_completion(self): data = samples()["11-images"] context = {"07-cuts": {"payload": {"cuts": [dict(id="c1", start_ms=0, end_ms=4000)]}}, "10-references": samples()["10-references"]} self.assertEqual(semantic_errors(data, context), []) self.assertIn("result=null은 완료가 아니다", semantic_errors(data, context, complete=True)) for change in ("duplicate", "reference", "scene", "aspect", "absolute"): bad = deepcopy(data) request = bad["payload"]["request"] if change == "duplicate": request["frames"][1]["id"] = request["frames"][0]["id"] elif change == "reference": request["frames"][0]["reference_ids"] = ["missing"] elif change == "scene": request["frames"][1]["scene_name"] = "Other" elif change == "aspect": request["render_profile"]["width"] = 1920 else: request["scene_path"] = "C:/foreign.blend" self.assertTrue(semantic_errors(bad, context), change) def test_blend_media_impersonation(self): data = samples()["08-scenes"] data["payload"]["preview"] = dict(scene_file=dict(BLEND, verified=True, sha256="a" * 64, blender_version="5.2.2"), render_profile=deepcopy(PROFILE), samples=[dict(id="p1", cut_ids=["c1"], media=dict(IMAGE, path="examples/blender-contract/wrong.blend", verified=True, width=1080, height=1920))]) self.valid(data) self.assertIn(".blend는 media가 아니다", semantic_errors(data, {"07-cuts": {"payload": {"cuts": [{"id": "c1"}]}}})) if __name__ == "__main__": unittest.main(verbosity=2)