NINI
435 lines
25 KiB
Python
435 lines
25 KiB
Python
"""승인 경계·시간·원본 참조 회귀. 생성한 시도는 진단을 위해 남긴다."""
|
|
from copy import deepcopy
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import uuid
|
|
import wave
|
|
import zlib
|
|
import struct
|
|
import shutil
|
|
|
|
import run
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
WATER = ROOT / "projects/water-hydrogen-bond-surface-tension"
|
|
PROFILE = dict(engine="CYCLES", device="CPU", fps=10, width=90, height=160, samples=4, seed=0)
|
|
|
|
|
|
def write(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def fixture():
|
|
folder = ROOT / "tools/blender/.cache" / uuid.uuid4().hex / WATER.name
|
|
project = json.loads((WATER / "project.json").read_text(encoding="utf-8"))
|
|
project["schema_version"] = "1.1"
|
|
project["settings"].pop("image_model", None)
|
|
project["settings"]["visual_engine"] = "Blender"
|
|
artifacts = {}
|
|
for stage in project["stages"]:
|
|
if stage["revision"] is None:
|
|
continue
|
|
data = json.loads((ROOT / stage["artifact_path"]).read_text(encoding="utf-8"))
|
|
if stage["stage_id"] in ("08-scenes", "09-entities"):
|
|
data.update(schema_version="1.1", revision=2)
|
|
stage.update(status="draft", revision=2)
|
|
stage["artifact_path"] = (folder / "stages" / stage["stage_id"] / ("r" + str(stage["revision"])) / "artifact.json").relative_to(ROOT).as_posix()
|
|
artifacts[stage["stage_id"]] = data
|
|
cuts = artifacts["07-cuts"]["payload"]
|
|
cuts["narration_duration_ms"] = 2000
|
|
cuts["cuts"] = [dict(id="c1", start_ms=0, end_ms=1000, unit_ids=["u1"], narration_text="첫 장면"),
|
|
dict(id="c2", start_ms=1000, end_ms=2000, unit_ids=["u1"], narration_text="다음 장면")]
|
|
cuts["captions"] = []
|
|
source = folder / "stages/08-scenes/r2/sample.blend"
|
|
source.parent.mkdir(parents=True, exist_ok=True)
|
|
source.write_bytes("검사 전 원본".encode("utf-8"))
|
|
narration = folder / "narration.wav"
|
|
with wave.open(str(narration), "wb") as audio:
|
|
audio.setparams((1, 2, 24000, 0, "NONE", "not compressed"))
|
|
audio.writeframes(b"\0\0" * 48000)
|
|
artifacts["06-tts"]["payload"]["result"]["narration"].update(
|
|
path=narration.relative_to(ROOT).as_posix(), duration_ms=2000)
|
|
tts = artifacts["06-tts"]["payload"]
|
|
ids = [unit["id"] for unit in tts["request"]["units"]]
|
|
tts["result"]["files"] = []
|
|
tts["result"]["unit_ranges"] = []
|
|
for index, unit_id in enumerate(ids):
|
|
path = folder / (unit_id + ".wav")
|
|
with wave.open(str(path), "wb") as audio:
|
|
audio.setparams((1, 2, 24000, 0, "NONE", "not compressed"))
|
|
audio.writeframes(b"\0\0" * 6000)
|
|
tts["result"]["files"].append(dict(unit_id=unit_id, media=dict(path=path.relative_to(ROOT).as_posix(), kind="audio", verified=True,
|
|
duration_ms=250, width=None, height=None)))
|
|
tts["result"]["unit_ranges"].append(dict(unit_id=unit_id, start_ms=index * 250, end_ms=(index + 1) * 250))
|
|
for cut in cuts["cuts"]:
|
|
cut["unit_ids"] = ids[:4] if cut["id"] == "c1" else ids[4:]
|
|
scenes = artifacts["08-scenes"]
|
|
scenes["payload"]["scenes"] = [dict(cut_id=cut["id"], description="움직이는 구", start_state="왼쪽", end_state="오른쪽", camera="고정") for cut in cuts["cuts"]]
|
|
scenes["payload"]["connections"] = [dict(from_cut_id="c1", to_cut_id="c2", mode="cut", reason="시점 변경")]
|
|
scenes["payload"]["preview"] = dict(scene_file=dict(path=source.relative_to(ROOT).as_posix(), verified=False, sha256=None, blender_version=None), render_profile=deepcopy(PROFILE), samples=[])
|
|
for stage in project["stages"]:
|
|
if stage["stage_id"] in artifacts:
|
|
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
|
|
write(folder / "project.json", project)
|
|
job = dict(mode="preview", project=(folder / "project.json").relative_to(ROOT).as_posix(),
|
|
artifact=next(s["artifact_path"] for s in project["stages"] if s["stage_id"] == "08-scenes"),
|
|
output_revision=2, source=source.relative_to(ROOT).as_posix(), scene="SceneA",
|
|
frame=None, start=0, end=10, cut_id=["c1"], frame_id=None, reference_id=None,
|
|
require_object=["Ball"])
|
|
return folder, project, artifacts, job
|
|
|
|
|
|
def production_fixture():
|
|
folder, project, artifacts, job = fixture()
|
|
for stage in project["stages"]:
|
|
if stage["stage_id"] in ("08-scenes", "09-entities"):
|
|
stage["status"] = "complete"
|
|
for ref in artifacts["09-entities"]["inputs"]:
|
|
if ref["stage_id"] == "08-scenes":
|
|
ref["revision"] = 2
|
|
artifacts["09-entities"]["payload"]["appearances"] = [dict(cut_id=c, character_ids=[], background_ids=[]) for c in ("c1", "c2")]
|
|
preview = artifacts["08-scenes"]["payload"]["preview"]
|
|
preview["scene_file"].update(verified=True, sha256=run.digest(ROOT / job["source"]), blender_version="5.2.2 LTS")
|
|
preview["samples"] = [dict(id="sample", cut_ids=["c1"], media=dict(path=job["source"].replace(".blend", ".png"),
|
|
kind="image", verified=True, duration_ms=None, width=90, height=160))]
|
|
def chunk(kind, body):
|
|
return struct.pack(">I", len(body)) + kind + body + struct.pack(">I", zlib.crc32(kind + body))
|
|
png = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", 90, 160, 8, 2, 0, 0, 0))
|
|
png += chunk(b"IDAT", zlib.compress((b"\0" + b"\0" * 270) * 160)) + chunk(b"IEND", b"")
|
|
(ROOT / preview["samples"][0]["media"]["path"]).write_bytes(png)
|
|
review = next(r for r in project["reviews"] if r["review_id"] == "scenes")
|
|
review["decision"] = "approved"
|
|
review["artifact_refs"] = [dict(stage_id=s, revision=1 if s == "07-cuts" else 2) for s in ("07-cuts", "08-scenes", "09-entities")]
|
|
review["feedback"] = "회귀 검사 전용 승인 기록이며 실제 영상의 사용자 승인이 아닙니다"
|
|
reference = json.loads((ROOT / "templates/stages/10-references.json").read_text(encoding="utf-8"))
|
|
reference.update(project_id=project["project_id"], revision=2,
|
|
inputs=[dict(stage_id=s, revision=1 if s == "02-research" else 2) for s in ("02-research", "08-scenes", "09-entities")])
|
|
reference["payload"]["request"]["render_profile"] = deepcopy(PROFILE)
|
|
reference["payload"]["request"]["items"] = [dict(id="ball", kind="object", entity_id=None, prompt="검사용 구")]
|
|
pointer = next(s for s in project["stages"] if s["stage_id"] == "10-references")
|
|
pointer.update(status="draft", revision=2,
|
|
artifact_path=(folder / "stages/10-references/r2/artifact.json").relative_to(ROOT).as_posix())
|
|
artifacts["10-references"] = reference
|
|
source = folder / "stages/10-references/r2/model.blend"
|
|
source.parent.mkdir(parents=True, exist_ok=True)
|
|
source.write_bytes(b"uninspected fixture")
|
|
job.update(mode="production", artifact=pointer["artifact_path"], source=source.relative_to(ROOT).as_posix(),
|
|
frame=0, start=None, end=None, reference_id="ball", cut_id=[])
|
|
for stage in project["stages"]:
|
|
if stage["stage_id"] in artifacts:
|
|
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
|
|
write(folder / "project.json", project)
|
|
return folder, project, artifacts, job
|
|
|
|
|
|
def endpoint_fixture(motion=False):
|
|
folder, project, artifacts, job = production_fixture()
|
|
reference = artifacts["10-references"]
|
|
reference["payload"]["result"] = dict(scene_file=dict(path=job["source"], verified=True,
|
|
sha256=run.digest(ROOT / job["source"]), blender_version="5.2.2 LTS"),
|
|
items=[dict(reference_id="ball", object_name="Ball", media=deepcopy(artifacts["08-scenes"]["payload"]["preview"]["samples"][0]["media"]))])
|
|
reference_media = reference["payload"]["result"]["items"][0]["media"]
|
|
image = folder / "stages/10-references/r2/reference.png"
|
|
shutil.copyfile(ROOT / reference_media["path"], image)
|
|
reference_media["path"] = image.relative_to(ROOT).as_posix()
|
|
frames = [dict(id=cut + "-" + position, cut_id=cut, position=position, prompt="상태 설명", reference_ids=["ball"],
|
|
scene_name="SceneA" if cut == "c1" else "SceneB", frame_index=index)
|
|
for cut, start in (("c1", 0), ("c2", 10)) for position, index in (("start", start), ("end", start + 10))]
|
|
source = folder / "stages/11-images/r2/animation.blend"
|
|
source.parent.mkdir(parents=True, exist_ok=True)
|
|
source.write_bytes(b"uninspected endpoint fixture")
|
|
images = json.loads((ROOT / "templates/stages/11-images.json").read_text(encoding="utf-8"))
|
|
images.update(project_id=project["project_id"], revision=2,
|
|
inputs=[dict(stage_id=s, revision=1 if s in ("02-research", "03-script", "07-cuts") else 2)
|
|
for s in ("02-research", "03-script", "07-cuts", "08-scenes", "09-entities", "10-references")])
|
|
images["payload"]["request"].update(frames=frames, scene_path=source.relative_to(ROOT).as_posix(), render_profile=deepcopy(PROFILE))
|
|
artifacts["11-images"] = images
|
|
for stage in project["stages"]:
|
|
if stage["stage_id"] in ("10-references", "11-images"):
|
|
stage.update(revision=2, status="complete" if stage["stage_id"] == "10-references" or motion else "draft",
|
|
artifact_path=(folder / "stages" / stage["stage_id"] / "r2/artifact.json").relative_to(ROOT).as_posix())
|
|
job.update(artifact=(folder / "stages/11-images/r2/artifact.json").relative_to(ROOT).as_posix(),
|
|
source=source.relative_to(ROOT).as_posix(), frame_id="c1-start", reference_id=None)
|
|
if motion:
|
|
images["payload"]["result"] = dict(scene_file=dict(path=job["source"], verified=True,
|
|
sha256=run.digest(source), blender_version="5.2.2 LTS"),
|
|
frames=[dict(frame_id=f["id"], media=deepcopy(reference["payload"]["result"]["items"][0]["media"])) for f in frames])
|
|
for item in images["payload"]["result"]["frames"]:
|
|
image = source.parent / (item["frame_id"] + ".png")
|
|
shutil.copyfile(ROOT / item["media"]["path"], image)
|
|
item["media"]["path"] = image.relative_to(ROOT).as_posix()
|
|
clips = [dict(cut_id=c, start_frame_id=c + "-start", end_frame_id=c + "-end", duration_ms=1000,
|
|
motion_prompt="구의 움직임", scene_name="SceneA" if c == "c1" else "SceneB", start_frame=start, end_frame=start + 10)
|
|
for c, start in (("c1", 0), ("c2", 10))]
|
|
artifact = json.loads((ROOT / "templates/stages/12-motion.json").read_text(encoding="utf-8"))
|
|
artifact.update(project_id=project["project_id"], revision=2,
|
|
inputs=[dict(stage_id=s, revision=1 if s == "07-cuts" else 2) for s in ("07-cuts", "08-scenes", "11-images")])
|
|
artifact["payload"]["request"].update(clips=clips, scene_path=job["source"], render_profile=deepcopy(PROFILE))
|
|
artifacts["12-motion"] = artifact
|
|
pointer = next(s for s in project["stages"] if s["stage_id"] == "12-motion")
|
|
pointer.update(revision=2, status="draft", artifact_path=(folder / "stages/12-motion/r2/artifact.json").relative_to(ROOT).as_posix())
|
|
job.update(artifact=pointer["artifact_path"], frame=None, start=0, end=10, cut_id=["c1"], frame_id=None)
|
|
for stage in project["stages"]:
|
|
if stage["stage_id"] in artifacts:
|
|
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
|
|
write(folder / "project.json", project)
|
|
return folder, project, artifacts, job
|
|
|
|
|
|
class RequestTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.folder, self.project, self.artifacts, self.job = fixture()
|
|
|
|
def persist(self):
|
|
write(self.folder / "project.json", self.project)
|
|
for stage in self.project["stages"]:
|
|
if stage["stage_id"] in self.artifacts:
|
|
write(ROOT / stage["artifact_path"], self.artifacts[stage["stage_id"]])
|
|
|
|
def test_fixture_uses_stage_id_for_reordered_inputs(self):
|
|
sample = fixture()
|
|
sample[2]["09-entities"]["inputs"].reverse()
|
|
with patch(__name__ + ".fixture", return_value=sample):
|
|
_, _, artifacts, job = production_fixture()
|
|
self.assertEqual(run.prepare(ROOT, job)["frames"], [0])
|
|
refs = {r["stage_id"]: r["revision"] for r in artifacts["09-entities"]["inputs"]}
|
|
self.assertEqual(refs["08-scenes"], 2)
|
|
self.assertEqual(refs["03-script"], 1)
|
|
|
|
def test_preview_requires_no_new_scenes_approval(self):
|
|
validated = run.prepare(ROOT, self.job)
|
|
self.assertEqual(validated["profile"], PROFILE)
|
|
self.assertEqual(validated["frames"], list(range(10)))
|
|
|
|
def test_production_rejects_preview_and_old_scenes(self):
|
|
self.job["mode"] = "production"
|
|
with self.assertRaises(ValueError):
|
|
run.prepare(ROOT, self.job)
|
|
folder, project, artifacts, job = production_fixture()
|
|
self.assertEqual(run.prepare(ROOT, job)["frames"], [0])
|
|
review = next(r for r in project["reviews"] if r["review_id"] == "scenes")
|
|
for decision in ("pending", "changes_requested", "stale", "old_approved", "no_feedback"):
|
|
review.update(decision="approved", feedback="회귀 검사 기록")
|
|
review["artifact_refs"][1]["revision"] = 2
|
|
if decision == "old_approved":
|
|
review["artifact_refs"][1]["revision"] = 1
|
|
elif decision == "no_feedback":
|
|
review["feedback"] = ""
|
|
else:
|
|
review["decision"] = decision
|
|
write(folder / "project.json", project)
|
|
with self.subTest(decision=decision), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_bad_pointers_inputs_and_revision(self):
|
|
for change in ("revision", "pointer", "stale", "input"):
|
|
folder, project, artifacts, job = fixture()
|
|
if change == "revision":
|
|
job["output_revision"] = 1
|
|
elif change == "pointer":
|
|
project["stages"][7]["revision"] = 1
|
|
elif change == "stale":
|
|
project["stages"][2]["status"] = "stale"
|
|
else:
|
|
artifacts["08-scenes"]["inputs"][0]["revision"] = 99
|
|
write(ROOT / job["artifact"], artifacts["08-scenes"])
|
|
write(folder / "project.json", project)
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_ranges_aspect_grid_and_foreign_source(self):
|
|
for change in ("long", "negative", "unknown_cut", "foreign", "aspect", "grid"):
|
|
folder, project, artifacts, job = fixture()
|
|
if change == "long":
|
|
job["end"] = 51
|
|
elif change == "negative":
|
|
job["start"] = -1
|
|
elif change == "unknown_cut":
|
|
job["cut_id"] = ["foreign"]
|
|
elif change == "foreign":
|
|
job["source"] = "../foreign.blend"
|
|
elif change == "aspect":
|
|
artifacts["08-scenes"]["payload"]["preview"]["render_profile"]["width"] = 100
|
|
else:
|
|
artifacts["07-cuts"]["payload"]["cuts"][0]["end_ms"] = 999
|
|
for stage in project["stages"]:
|
|
if stage["stage_id"] in artifacts:
|
|
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_source_hash_changed_is_rejected(self):
|
|
metadata = dict(path=self.job["source"], verified=True, sha256="0" * 64, blender_version="5.2.2 LTS")
|
|
with self.assertRaises(ValueError):
|
|
run.check_source(ROOT, metadata)
|
|
|
|
def test_eevee_does_not_claim_an_unsupported_seed(self):
|
|
profile = self.artifacts["08-scenes"]["payload"]["preview"]["render_profile"]
|
|
profile.update(engine="BLENDER_EEVEE", device="GPU", seed=5)
|
|
self.persist()
|
|
with self.assertRaises(ValueError):
|
|
run.prepare(ROOT, self.job)
|
|
|
|
def test_production_missing_or_foreign_preview_is_rejected(self):
|
|
for change in ("missing", "foreign", "hash"):
|
|
folder, project, artifacts, job = production_fixture()
|
|
preview = artifacts["08-scenes"]["payload"]["preview"]
|
|
if change == "missing":
|
|
preview["samples"][0]["media"]["path"] += ".missing"
|
|
elif change == "foreign":
|
|
preview["scene_file"]["path"] = self.job["source"]
|
|
preview["scene_file"]["sha256"] = run.digest(ROOT / self.job["source"])
|
|
else:
|
|
preview["scene_file"]["sha256"] = "a" * 64
|
|
write(folder / "stages/08-scenes/r2/artifact.json", artifacts["08-scenes"])
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_attempt_identity_includes_scene_hash_profile_and_range(self):
|
|
job = run.prepare(ROOT, self.job)
|
|
key = run.identity(job, "a" * 64)
|
|
for name, value in (("scene", "SceneB"), ("frames", [10]), ("profile", dict(PROFILE, seed=4))):
|
|
other = dict(job, **{name: value})
|
|
self.assertNotEqual(key, run.identity(other, "a" * 64))
|
|
self.assertNotEqual(key, run.identity(job, "b" * 64))
|
|
|
|
def test_endpoint_and_motion_requests_match_actual_cuts(self):
|
|
for motion in (False, True):
|
|
folder, project, artifacts, job = endpoint_fixture(motion)
|
|
self.assertEqual(len(run.prepare(ROOT, job)["frames"]), 10 if motion else 1)
|
|
for change in ("scene", "endpoint", "reference", "profile", "continuous", "cut_scene"):
|
|
folder, project, artifacts, job = endpoint_fixture(motion)
|
|
stage = "12-motion" if motion else "11-images"
|
|
request = artifacts[stage]["payload"]["request"]
|
|
if change == "scene":
|
|
job["scene"] = "ForeignScene"
|
|
elif change == "endpoint":
|
|
artifacts["11-images"]["payload"]["request"]["frames"][3]["frame_index"] = 19
|
|
elif change == "reference":
|
|
artifacts["11-images"]["payload"]["request"]["frames"][0]["reference_ids"] = ["missing"]
|
|
elif change == "profile":
|
|
request["render_profile"]["fps"] = 20
|
|
elif change == "continuous":
|
|
artifacts["08-scenes"]["payload"]["connections"][0]["mode"] = "continuous"
|
|
else:
|
|
for target in artifacts["11-images"]["payload"]["request"]["frames"]:
|
|
target["scene_name"] = "SceneA"
|
|
if motion:
|
|
artifacts["12-motion"]["payload"]["request"]["clips"][1]["scene_name"] = "SceneA"
|
|
for pointer in project["stages"]:
|
|
if pointer["stage_id"] in artifacts:
|
|
write(ROOT / pointer["artifact_path"], artifacts[pointer["stage_id"]])
|
|
with self.subTest(motion=motion, change=change), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_current_10_entity_reference_must_exist(self):
|
|
folder, project, artifacts, job = production_fixture()
|
|
artifacts["10-references"]["payload"]["request"]["items"][0].update(kind="character", entity_id="ABSENT")
|
|
write(ROOT / job["artifact"], artifacts["10-references"])
|
|
with self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_new_renderer_rejects_legacy_project_configuration(self):
|
|
self.project.update(schema_version="1.0")
|
|
self.project["settings"].pop("visual_engine")
|
|
self.project["settings"]["image_model"] = "gpt-image-2.5-sunburst"
|
|
self.persist()
|
|
with self.assertRaises(ValueError):
|
|
run.prepare(ROOT, self.job)
|
|
|
|
def test_complete_06_unit_files_need_actual_metadata(self):
|
|
for change in ("missing", "unverified", "duration", "range", "truncated", "id"):
|
|
folder, project, artifacts, job = fixture()
|
|
tts = artifacts["06-tts"]["payload"]
|
|
media = tts["result"]["files"][0]["media"]
|
|
if change == "missing":
|
|
media["path"] += ".missing"
|
|
elif change == "unverified":
|
|
media["verified"] = False
|
|
elif change == "duration":
|
|
media["duration_ms"] = 251
|
|
elif change == "range":
|
|
tts["result"]["unit_ranges"][0]["end_ms"] = 251
|
|
elif change == "id":
|
|
tts["result"]["files"][0]["unit_id"] = "foreign"
|
|
else:
|
|
path = ROOT / media["path"]
|
|
path.write_bytes(path.read_bytes()[:-2])
|
|
write(folder / "stages/06-tts/r1/artifact.json", artifacts["06-tts"])
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_old_visual_revision_source_and_media_are_rejected(self):
|
|
for input_stage in ("08-scenes", "10-references", "11-images"):
|
|
for change in ("source", "media"):
|
|
folder, project, artifacts, job = endpoint_fixture(motion=True)
|
|
payload = artifacts[input_stage]["payload"]
|
|
result = payload["preview"] if input_stage == "08-scenes" else payload["result"]
|
|
key = {"08-scenes": "samples", "10-references": "items", "11-images": "frames"}[input_stage]
|
|
metadata = result["scene_file"] if change == "source" else result[key][0]["media"]
|
|
old = ROOT / metadata["path"].replace("/r2/", "/r1/")
|
|
old.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(ROOT / metadata["path"], old)
|
|
metadata["path"] = old.relative_to(ROOT).as_posix()
|
|
if input_stage == "11-images" and change == "source":
|
|
artifacts["12-motion"]["payload"]["request"]["scene_path"] = metadata["path"]
|
|
job["source"] = metadata["path"]
|
|
for pointer in project["stages"]:
|
|
if pointer["stage_id"] in artifacts:
|
|
write(ROOT / pointer["artifact_path"], artifacts[pointer["stage_id"]])
|
|
with self.subTest(stage=input_stage, change=change), self.assertRaises(ValueError):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_partial_11_allows_the_next_endpoint(self):
|
|
folder, project, artifacts, job = endpoint_fixture(motion=True)
|
|
images = artifacts["11-images"]["payload"]
|
|
images["result"]["frames"] = images["result"]["frames"][:1]
|
|
for frame in images["request"]["frames"]:
|
|
frame["scene_name"] = "SceneA"
|
|
artifacts["08-scenes"]["payload"]["connections"][0]["mode"] = "continuous"
|
|
pointer = next(s for s in project["stages"] if s["stage_id"] == "11-images")
|
|
pointer["status"] = "draft"
|
|
job.update(artifact=pointer["artifact_path"], frame_id="c1-end", frame=10,
|
|
start=None, end=None, cut_id=[])
|
|
write(folder / "project.json", project)
|
|
write(folder / "stages/08-scenes/r2/artifact.json", artifacts["08-scenes"])
|
|
write(ROOT / job["artifact"], artifacts["11-images"])
|
|
self.assertEqual(run.prepare(ROOT, job)["frames"], [10])
|
|
|
|
def test_partial_11_rejects_different_completed_boundary_paths(self):
|
|
folder, project, artifacts, job = endpoint_fixture(motion=True)
|
|
images = artifacts["11-images"]["payload"]
|
|
images["result"]["frames"] = images["result"]["frames"][1:3]
|
|
for frame in images["request"]["frames"]:
|
|
frame["scene_name"] = "SceneA"
|
|
artifacts["08-scenes"]["payload"]["connections"][0]["mode"] = "continuous"
|
|
pointer = next(s for s in project["stages"] if s["stage_id"] == "11-images")
|
|
pointer["status"] = "draft"
|
|
job.update(artifact=pointer["artifact_path"], frame_id="c1-end", frame=10,
|
|
start=None, end=None, cut_id=[])
|
|
write(folder / "project.json", project)
|
|
write(folder / "stages/08-scenes/r2/artifact.json", artifacts["08-scenes"])
|
|
write(ROOT / job["artifact"], artifacts["11-images"])
|
|
with self.assertRaisesRegex(ValueError, "같은 실제 PNG"):
|
|
run.prepare(ROOT, job)
|
|
|
|
def test_12_rejects_partial_or_unverified_11_marked_complete(self):
|
|
for change in ("partial", "unverified"):
|
|
folder, project, artifacts, job = endpoint_fixture(motion=True)
|
|
frames = artifacts["11-images"]["payload"]["result"]["frames"]
|
|
if change == "partial":
|
|
artifacts["11-images"]["payload"]["result"]["frames"] = frames[:1]
|
|
else:
|
|
frames[1]["media"]["verified"] = False
|
|
write(folder / "stages/11-images/r2/artifact.json", artifacts["11-images"])
|
|
with self.subTest(change=change), self.assertRaisesRegex(ValueError, "완료 선행 결과"):
|
|
run.prepare(ROOT, job)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|