NINI
350 lines
18 KiB
Python
350 lines
18 KiB
Python
"""저장소 계약의 회귀 검사. 실제 렌더링·승인·미디어 검수는 수행하지 않는다."""
|
|
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 complete:
|
|
require(preview is not None, "대표 품질 검토 자료 필요")
|
|
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"], "대표 미디어 미검증")
|
|
width, height = sample["media"]["width"], sample["media"]["height"]
|
|
require(isinstance(width, int) and isinstance(height, int)
|
|
and width * 16 == height * 9, "대표 실제 미디어는 9:16 필요")
|
|
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 stage in ("10-references", "11-images", "12-motion"):
|
|
profile = read("templates/stages/" + stage + ".json")["payload"]["request"]["render_profile"]
|
|
self.assertEqual((profile["samples"], profile["seed"]), (64, 0))
|
|
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"}]}}}))
|
|
|
|
def test_preview_video_duration_and_actual_aspect(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, kind="video",
|
|
path="examples/blender-contract/preview.mp4", verified=True,
|
|
duration_ms=5000, width=1080, height=1920))])
|
|
context = {"07-cuts": {"payload": {"cuts": [{"id": "c1"}]}}}
|
|
for duration in (1, 5000):
|
|
with self.subTest(valid_duration=duration):
|
|
data["payload"]["preview"]["samples"][0]["media"]["duration_ms"] = duration
|
|
self.valid(data)
|
|
self.assertEqual(semantic_errors(data, context, complete=True), [])
|
|
for duration in (0, 5001, 6000):
|
|
with self.subTest(invalid_duration=duration):
|
|
bad = deepcopy(data)
|
|
bad["payload"]["preview"]["samples"][0]["media"]["duration_ms"] = duration
|
|
self.invalid(bad)
|
|
for width, height in ((1920, 1080), (1080, 1800)):
|
|
with self.subTest(invalid_aspect=(width, height)):
|
|
bad = deepcopy(data)
|
|
bad["payload"]["preview"]["samples"][0]["media"].update(width=width, height=height)
|
|
self.valid(bad)
|
|
self.assertTrue(semantic_errors(bad, context, complete=True))
|
|
for field in ("duration_ms", "width", "height"):
|
|
with self.subTest(verified_missing_metadata=field):
|
|
bad = deepcopy(data)
|
|
bad["payload"]["preview"]["samples"][0]["media"][field] = None
|
|
self.invalid(bad)
|
|
bad = deepcopy(data)
|
|
bad["payload"]["preview"]["samples"][0]["media"].update(verified=False,
|
|
duration_ms=None, width=None, height=None)
|
|
self.valid(bad)
|
|
self.assertTrue(semantic_errors(bad, context, complete=True))
|
|
bad = samples()["08-scenes"]
|
|
self.valid(bad)
|
|
self.assertTrue(semantic_errors(bad, context, complete=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|