"""Word-timing job: planning, sidecar client, v2 record format and graceful fallback (sidecar mocked).""" from __future__ import annotations import json import os import shutil import tempfile import time import unittest from pathlib import Path from unittest import mock _CONFIG = tempfile.mkdtemp(prefix="ttvc-wordalign-test-") os.environ.setdefault("CONFIG_DIR", _CONFIG) import requests from core import epub_export, word_alignment as wa def units_for(record, blocks): return epub_export.plan_sentences(blocks, record["duration_ms"], record, None) BLOCKS = [{"id": "b1", "text": "Erster Satz hier. Zweiter Satz."}, {"id": "b2", "text": "Dritter Satz."}] RECORD = {"version": 1, "interview": False, "duration_ms": 6000, "segments": [ {"chars": len(BLOCKS[0]["text"]), "times": [[0, 2000], [2000, 3000]]}, {"chars": len(BLOCKS[1]["text"]), "times": [[3000, 6000]]}]} def fake_post_factory(calls): def post(url, wav, spans, language, **_): calls.append((url, spans, language)) # one word per token, evenly spread over the span out = [] for sp in spans: toks = sp["text"].split() step = (sp["end"] - sp["start"]) / len(toks) out.append([{"word": t, "start": round(sp["start"] + i * step, 3), "end": round(sp["start"] + (i + 1) * step, 3), "aligned": True} for i, t in enumerate(toks)]) return out return post class PlanningTests(unittest.TestCase): def test_exact_blocks_get_one_job_per_sentence(self): units, q = units_for(RECORD, BLOCKS) self.assertEqual(q, "exact") jobs = wa.alignment_jobs(units, BLOCKS, False) self.assertEqual([(j["start"], j["end"], j["text"]) for j in jobs], [(0, 2000, "Erster Satz hier."), (2000, 3000, "Zweiter Satz."), (3000, 6000, "Dritter Satz.")]) def test_anchored_block_is_aligned_as_one_span(self): rec = json.loads(json.dumps(RECORD)) rec["segments"][0]["times"] = [[0, 3000]] # one utterance for two sentences units, q = units_for(rec, BLOCKS) self.assertEqual(q, "anchored") jobs = wa.alignment_jobs(units, BLOCKS, False) self.assertEqual(jobs[0]["text"], "Erster Satz hier. Zweiter Satz.") self.assertEqual(jobs[0]["units"], [0, 1]) def test_speaker_label_is_not_sent_and_gets_a_zero_length_slot(self): blocks = [{"id": "b1", "text": "Frage Eins: Wie geht es? Gut."}] rec = {"version": 1, "interview": True, "duration_ms": 4000, "segments": [ {"chars": len("Wie geht es? Gut."), "times": [[0, 2000], [2000, 4000]]}]} units, q = epub_export.plan_sentences(blocks, 4000, rec, [len("Wie geht es? Gut.")]) self.assertEqual(q, "exact") jobs = wa.alignment_jobs(units, blocks, True) self.assertEqual(jobs[0]["text"], "Wie geht es?") self.assertEqual(jobs[0]["lead"], 2) class JobTests(unittest.TestCase): def setUp(self): self.d = Path(tempfile.mkdtemp(prefix="ttvc-wa-")) self.addCleanup(shutil.rmtree, self.d, True) from pydub import AudioSegment from pydub.generators import Sine self.mp3 = self.d / "a1.mp3" Sine(300).to_audio_segment(duration=6000).export(self.mp3, format="mp3") self.rec_path = self.d / "a1.timings.json" self.rec_path.write_text(json.dumps(RECORD)) self.units, _ = units_for(RECORD, BLOCKS) self.jobs = wa.alignment_jobs(self.units, BLOCKS, False) def test_words_are_written_as_version_2_and_old_fields_survive(self): calls = [] got = wa.align_words(self.mp3, self.units, self.jobs, "de", url="http://x", post=fake_post_factory(calls)) seg_words = wa.build_segment_words(self.units, got, 2) self.assertTrue(wa.write_words(self.mp3, seg_words)) rec = json.loads(self.rec_path.read_text()) self.assertEqual(rec["version"], 2) self.assertEqual(rec["segments"][0]["times"], RECORD["segments"][0]["times"]) # v1 data untouched self.assertEqual(len(rec["segments"][0]["words"]), 5) # 3 + 2 tokens self.assertEqual(len(rec["segments"][1]["words"]), 2) first, last = rec["segments"][0]["words"][0], rec["segments"][0]["words"][-1] self.assertAlmostEqual(first[0], 0, delta=150) self.assertAlmostEqual(last[1], 3000, delta=150) # the v1 reader still accepts it from core import magazine as mag self.assertIsNotNone(mag.read_audio_timings(self.mp3)) self.assertEqual(calls[0][2], "de") def test_chunks_are_bounded_and_progress_reported(self): many = [{"units": [0], "start": i * 10000, "end": i * 10000 + 9000, "text": "a", "lead": 0} for i in range(7)] self.assertTrue(all(len(c) <= 4 for c in wa._chunks(many))) # 45 s per request seen = [] wa.align_words(self.mp3, self.units, self.jobs, "de", url="http://x", post=fake_post_factory([]), progress=lambda a, b: seen.append((a, b))) self.assertEqual(seen[-1], (3, 3)) def test_incomplete_segment_gets_no_words(self): got = {0: [[0, 1]] * 3} # sentence 1 missing for segment 0 self.assertEqual(wa.build_segment_words(self.units, got, 2), {}) class ClientFallbackTests(unittest.TestCase): def test_retries_with_backoff_then_raises_when_sidecar_is_down(self): sleeps = [] with mock.patch.object(wa.requests, "post", side_effect=requests.ConnectionError("down")) as p: with self.assertRaises(wa.AlignerError): wa._post_chunk("http://x", b"w", [{"start": 0, "end": 1, "text": "a"}], "de", sleep=sleeps.append) self.assertEqual(p.call_count, wa.ATTEMPTS) self.assertEqual(sleeps, [1, 2]) def test_client_error_is_not_retried(self): resp = mock.Mock(status_code=422) with mock.patch.object(wa.requests, "post", return_value=resp) as p: with self.assertRaises(wa.AlignerError): wa._post_chunk("http://x", b"w", [], "de", sleep=lambda s: None) self.assertEqual(p.call_count, 1) def test_busy_then_ok(self): ok = mock.Mock(status_code=200); ok.json.return_value = {"spans": [[]]} with mock.patch.object(wa.requests, "post", side_effect=[mock.Mock(status_code=503), ok]): self.assertEqual(wa._post_chunk("http://x", b"w", [], "de", sleep=lambda s: None), [[]]) def test_failed_job_is_recorded_and_not_restarted_within_cooldown(self): def run(progress): raise wa.AlignerError("aligner unreachable") with mock.patch.dict(os.environ, {"ALIGNER_URL": "http://x"}): self.assertTrue(wa.should_autostart("docF", "a1")) self.assertTrue(wa.start("docF", "a1", run)) for _ in range(100): if wa.status("docF", "a1")["state"] != "running": break time.sleep(0.02) st = wa.status("docF", "a1") self.assertEqual(st["state"], "failed") self.assertFalse(wa.should_autostart("docF", "a1")) def test_disabled_without_url(self): with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}): self.assertFalse(wa.should_autostart("docX", "a1")) class RouteTests(unittest.TestCase): """GET /timings carries words when present and never waits for the job.""" def setUp(self): from fastapi import FastAPI from fastapi.testclient import TestClient import routes.magazine as routes self.routes = routes self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-")) self.addCleanup(shutil.rmtree, self.lib, True) p = mock.patch.object(routes, "_LIB", self.lib); p.start(); self.addCleanup(p.stop) app = FastAPI(); app.include_router(routes.router) self.client = TestClient(app) self.doc = self.client.post("/api/magazine/docs", json={"title": "T"}).json()["id"] d = self.d = self.lib / self.doc blocks = [{"id": "b1", "page": 1, "bbox": [0, 0, 10, 10], "type": "body", "text": BLOCKS[0]["text"], "reading_order": 1}, {"id": "b2", "page": 1, "bbox": [0, 40, 10, 10], "type": "body", "text": BLOCKS[1]["text"], "reading_order": 2}] (d / "layout.json").write_text(json.dumps({"pages": [{"page": 1, "width": 600, "height": 800}], "blocks": blocks, "images": []})) (d / "articles.json").write_text(json.dumps({"articles": [ {"id": "a1", "title": "T", "pages": [1], "blocks": ["b1", "b2"], "images": [], "status": "approved"}]})) (d / "audio").mkdir(exist_ok=True) (d / "audio" / "a1.mp3").write_bytes(b"x") self.rec = d / "audio" / "a1.timings.json" self.rec.write_text(json.dumps(dict(RECORD, interview=True))) self.url = f"/api/magazine/{self.doc}/articles/a1/timings" def test_without_sidecar_the_answer_is_unchanged_and_instant(self): with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}): body = self.client.get(self.url).json() self.assertTrue(body["available"]) self.assertEqual(body["words"], "none") self.assertNotIn("words", body["sentences"][0]) def test_words_are_returned_per_sentence(self): rec = json.loads(self.rec.read_text()); rec["version"] = 2 rec["segments"][0]["words"] = [[0, 400], [500, 900], [1000, 1900], [2100, 2500], [2600, 2900]] rec["segments"][1]["words"] = [[3100, 3500], [3600, 5900]] self.rec.write_text(json.dumps(rec)) body = self.client.get(self.url).json() self.assertEqual(body["words"], "ready") self.assertEqual(body["sentences"][0]["words"], [[0.0, 0.4], [0.5, 0.9], [1.0, 1.9]]) self.assertEqual(body["sentences"][1]["words"], [[2.1, 2.5], [2.6, 2.9]]) self.assertEqual(body["sentences"][2]["words"][1], [3.6, 5.9]) def test_mismatching_words_are_ignored(self): rec = json.loads(self.rec.read_text()); rec["segments"][0]["words"] = [[0, 1]] self.rec.write_text(json.dumps(rec)) with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}): body = self.client.get(self.url).json() self.assertNotIn("words", body["sentences"][0]) def test_first_request_starts_the_job_in_background(self): started = [] with mock.patch.dict(os.environ, {"ALIGNER_URL": "http://x"}), \ mock.patch.object(self.routes, "_start_word_alignment", lambda *a: started.append(a) or True): body = self.client.get(self.url).json() self.assertEqual(body["words"], "running") self.assertEqual(len(started), 1) self.assertTrue(body["available"]) def test_explicit_endpoint_requires_configuration_and_timings(self): url = f"/api/magazine/{self.doc}/articles/a1/align-words" with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}): self.assertEqual(self.client.post(url).status_code, 503) with mock.patch.dict(os.environ, {"ALIGNER_URL": "http://x"}): self.rec.unlink() self.assertEqual(self.client.post(url).status_code, 404) self.assertIn("state", self.client.get(url).json()) if __name__ == "__main__": unittest.main()