| import pytest |
|
|
| from spi.compiler.production_graph import CycleError, ProductionGraph |
| from spi.renderers.base import RendererError, request_hash |
| from spi.renderers.mock import MockVideoRenderer |
| from spi.scheduler.scheduler import Scheduler |
| from spi.schemas.production import GenerationJob |
|
|
|
|
| def _job(jid, deps=None, priority=5, kind="video", shot_id="SHOT_0001"): |
| return GenerationJob(job_id=jid, depends_on=deps or [], priority=priority, |
| kind=kind, shot_id=shot_id, request={"prompt": jid}) |
|
|
|
|
| def test_topo_respects_deps_and_priority(): |
| g = ProductionGraph([_job("A"), _job("B"), _job("C", deps=["A"], priority=1), |
| _job("D", deps=["A", "C"])]) |
| order = g.topo_order() |
| assert order.index("A") < order.index("C") < order.index("D") |
| assert order.index("A") < order.index("B") |
|
|
|
|
| def test_cycle_detected(): |
| g = ProductionGraph([_job("A"), _job("B"), _job("C")]) |
| g.add_edge("A", "B") |
| g.add_edge("B", "C") |
| g.add_edge("C", "A") |
| with pytest.raises(CycleError): |
| g.topo_order() |
| assert any("cycle" in p for p in g.validate()) |
|
|
|
|
| def test_unknown_dependency_reported(): |
| |
| |
| |
| g = ProductionGraph([_job("A")]) |
| g.jobs["A"].depends_on.append("GHOST") |
| problems = g.validate() |
| assert any("GHOST" in p for p in problems) |
| with pytest.raises(ValueError): |
| g.topo_order() |
|
|
|
|
| def test_video_job_without_shot_reported(): |
| j = _job("A", kind="video", shot_id=None) |
| g = ProductionGraph([j]) |
| assert any("shot_id" in p for p in g.validate()) |
|
|
|
|
| def test_ready_jobs_ordering(): |
| g = ProductionGraph([_job("low", priority=9), _job("high", priority=1)]) |
| assert g.ready_jobs()[0] == "high" |
| g.mark("high", "done") |
| assert g.ready_jobs() == ["low"] |
|
|
|
|
| def test_request_hash_stable(): |
| r1 = {"prompt": "x", "seed": 1} |
| r2 = {"seed": 1, "prompt": "x"} |
| assert request_hash(r1) == request_hash(r2) |
|
|
|
|
| def test_mock_renderer_caches_identical_requests(): |
| r = MockVideoRenderer() |
| a1 = r.generate({"prompt": "same", "num_frames": 48, "seed": 7}) |
| a2 = r.generate({"seed": 7, "prompt": "same", "num_frames": 48}) |
| assert a1["request_hash"] == a2["request_hash"] |
| assert a1["cached"] is False and a2["cached"] is True |
|
|
|
|
| def test_scheduler_executes_all_and_records(): |
| g = ProductionGraph([_job("A"), _job("B", deps=["A"])]) |
| summary = Scheduler(g, MockVideoRenderer()).run() |
| assert summary["done"] == 2 and summary["failed"] == 0 |
| assert all(j.status == "done" for j in g.jobs.values()) |
| assert all(j.runtime_s is not None for j in g.jobs.values()) |
| assert g.jobs["A"].artifact_ref.startswith("mock://video/") |
|
|
|
|
| def test_scheduler_retries_then_fails(): |
| class ExplodingRenderer(MockVideoRenderer): |
| def generate(self, request): |
| raise RendererError("simulated OOM") |
|
|
| g = ProductionGraph([_job("A")]) |
| summary = Scheduler(g, ExplodingRenderer(), max_retries=2).run() |
| assert summary["failed"] == 1 |
| assert g.jobs["A"].status == "failed" |
| assert g.jobs["A"].retries == 3 |
| assert "OOM" in g.jobs["A"].error |
