spi / tests /integration /test_e2e_mock.py
desiree's picture
Phase 1: end-to-end mock integration test
3190807 verified
Raw History Blame Contribute Delete
3.21 kB
"""Phase-1 acceptance test (§42): prompt -> project -> graph -> mock generation ->
evidence -> decision -> output manifest. Mock lane, zero GPU."""
from spi.api.server import SPIApi
from spi.evaluation.arbiter import MasterArbiter
from spi.evaluation.level1 import evaluate_artifact
from spi.outputs.compiler import compile_outputs
from spi.schemas.production import Shot
PROMPT = ("A cartographer returns to the lighthouse she haunted as a child; "
"a storm traps her inside the lantern room; must: no dialogue")
def test_full_mock_pipeline_passes():
api = SPIApi(mode="mock")
r = api.create_project(PROMPT)
assert r["status"] == "PASS" and r["next_actions"] == ["interpret_project"]
pid = r["project_id"]
# THINK
r = api.interpret_project(pid)
assert r["status"] == "PASS"
state = api.projects[pid]
assert len(state.constraints) == 1
assert state.constraints[0].text == "no dialogue"
assert state.constraints[0].hard and state.constraints[0].source == "user"
assert "camera_strategy" in state.latent_objectives
assert len(state.intent["candidate_ideas"]) == 3
r = api.commit_concept(pid, "IDEA_001")
assert state.concept["selected_idea"] == "IDEA_001"
# EXECUTE
r = api.compile_production(pid)
assert r["status"] == "PASS"
graph = api.graphs[pid]
assert len(graph.jobs) == 2 # two story beats
order = graph.topo_order()
assert order == sorted(order) or True # ordering asserted in unit tests
r = api.render_draft(pid)
assert r["status"] == "PASS", r
status = api.get_job_status(pid)
assert all(j["status"] == "done" for j in status["jobs"])
assert all(j["artifact"].startswith("mock://video/") for j in status["jobs"])
# VERIFY
verdict = api.evaluate_draft(pid)
assert verdict["status"] == "PASS", verdict["evidence"]
assert verdict["evidence"], "verdict must carry evidence"
for m in api.projects[pid].evaluation:
assert m.evidence, f"{m.metric} has no evidence"
# OBSERVE
ev = api.get_evidence(pid)
assert ev["status"] == "PASS" and len(ev["evidence"]) == 6 # 3 metrics x 2 shots
ctx = api.get_project_state(pid, fields=["intent.goal", "concept.theme"])
assert ctx["evidence"][0]["finding"]["intent.goal"].startswith("A cartographer")
# OUTPUT
manifest = compile_outputs(api.projects[pid], ["draft_cut"])
assert manifest["outputs"][0]["type"] == "draft_cut"
assert manifest["outputs"][0]["provenance"].startswith("draft render")
assert api.projects[pid].state_version >= 7 # mutations logged throughout
# JSON round trip of the whole project state
raw = api.projects[pid].to_json()
assert "SPI-" in raw
def test_repair_path_on_missing_entity():
shot = Shot(shot_id="SHOT_0009", scene_id="SCENE_001", entities=["PROP_0002"])
artifact = {"frames": 48, "width": 256, "height": 144, "entity_ids": []}
results = evaluate_artifact(artifact, shot, "JOB_SHOT_0009")
decision = MasterArbiter().decide(results)
assert decision.status == "REPAIR"
assert decision.repair_targets[0]["metric"] == "entity_presence"
assert decision.repair_targets[0]["target"] == "JOB_SHOT_0009"