Download tests/integration/test_e2e_mock.py from desiree/spi: direct link, hf CLI and curl.
- Browser
- Download file 3.21 kB
-
https://huggingface.co/desiree/spi/resolve/main/tests/integration/test_e2e_mock.py
- Command line
-
hf download hf://desiree/spi/tests/integration/test_e2e_mock.py
-
curl -L -o test_e2e_mock.py https://huggingface.co/desiree/spi/resolve/main/tests/integration/test_e2e_mock.py
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" |