tests/test_render_pipeline.py

from pathlib import Path

from backend.render_pipeline import RenderPipeline


class FakeSynthesizer:
    def __init__(self) -> None:
        self.calls = []

    def synthesize(self, *, text: str, output_path: Path, **kwargs) -> dict:
        self.calls.append({"text": text, "output_path": output_path, **kwargs})
        output_path.write_bytes(b"RIFFfakewave")
        return {
            "duration_seconds": max(1, len(text.split()) // 2),
            "sample_rate": 24000,
            "backend": "local",
            "model": kwargs["voice_config"].model,
        }


def _chapters() -> list:
    return [
        {"id": "c1", "title": "One", "text": "hello world " * 20, "included": True, "est_minutes": 1},
        {"id": "c2", "title": "Two", "text": "next chapter " * 20, "included": True, "est_minutes": 1},
    ]


def test_render_pipeline_streams_ordered_progress_events(tmp_path: Path) -> None:
    synthesizer = FakeSynthesizer()
    pipeline = RenderPipeline(session_root=tmp_path, synthesizer=synthesizer)

    events = list(
        pipeline.render(
            session_id="session-a",
            book={"title": "Book"},
            chapters=_chapters(),
            voice_config={"mode": "auto"},
            diffusion_steps=32,
            speed=1.0,
        )
    )

    event_types = [event["type"] for event in events]
    assert event_types[0] == "started"
    assert "chapter_started" in event_types
    assert "chapter_done" in event_types
    assert event_types[-1] == "completed"
    assert len(synthesizer.calls) == 2
    assert all(event["backend"] == "local" for event in events)
    assert all(event["model"] == "omnivoice" for event in events)


def test_render_pipeline_can_be_cancelled(tmp_path: Path) -> None:
    synthesizer = FakeSynthesizer()
    pipeline = RenderPipeline(session_root=tmp_path, synthesizer=synthesizer)
    iterator = pipeline.render(
        session_id="session-a",
        book={"title": "Book"},
        chapters=_chapters(),
        voice_config={"mode": "auto"},
        diffusion_steps=32,
        speed=1.0,
    )

    first = next(iterator)
    pipeline.cancel("session-a")
    remaining = list(iterator)

    assert first["type"] == "started"
    assert remaining[-1]["type"] == "cancelled"