File size: 2,259 Bytes
af4851b cd0ff97 af4851b cd0ff97 af4851b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | 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"
|