301 lines
15 KiB
Python
301 lines
15 KiB
Python
"""EPUB 3 Media Overlay export: package structure, timing tiers, escaping, and the timing record."""
|
|
import json
|
|
import os
|
|
import re
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
from xml.etree import ElementTree as ET
|
|
|
|
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-test-config-"))
|
|
|
|
from core import epub_export as ee
|
|
from core import magazine as mag
|
|
|
|
NS = {"o": "http://www.idpf.org/2007/opf", "s": "http://www.w3.org/ns/SMIL", "x": "http://www.w3.org/1999/xhtml"}
|
|
|
|
|
|
def blocks():
|
|
return [
|
|
{"type": "title", "text": "Die Zukunft der Batterien"},
|
|
{"type": "byline", "text": "Von Jane Smith"},
|
|
{"type": "body", "text": "Erster Satz. Zweiter Satz! Dritter Satz?"},
|
|
{"type": "body", "text": "Ein einzelner Absatz."},
|
|
]
|
|
|
|
|
|
def exact_timings(bl=None):
|
|
bl = bl or blocks()
|
|
times, t = [], 0
|
|
segs = []
|
|
for b in bl:
|
|
n = len(ee.split_sentences(b["text"]))
|
|
seg = []
|
|
for _ in range(n):
|
|
seg.append([t, t + 1000]); t += 1000
|
|
segs.append({"chars": len(b["text"]), "times": seg})
|
|
return {"version": 1, "interview": False, "duration_ms": t, "segments": segs}
|
|
|
|
|
|
class PlanTests(unittest.TestCase):
|
|
def test_exact_uses_the_recorded_times(self):
|
|
units, q = ee.plan_sentences(blocks(), 99999, exact_timings())
|
|
self.assertEqual(q, "exact")
|
|
self.assertEqual(len(units), 1 + 1 + 3 + 1)
|
|
self.assertEqual((units[0]["start"], units[0]["end"]), (0, 1000))
|
|
self.assertEqual(units[-1]["end"], 6000)
|
|
|
|
def test_a_sentence_count_mismatch_keeps_the_paragraph_anchored(self):
|
|
t = exact_timings()
|
|
t["segments"][2]["times"] = [[2000, 5000]] # recorded as one utterance, text splits into three
|
|
units, q = ee.plan_sentences(blocks(), 99999, t)
|
|
self.assertEqual(q, "anchored")
|
|
para = [u for u in units if u["block"] == 2]
|
|
self.assertAlmostEqual(para[0]["start"], 2000)
|
|
self.assertAlmostEqual(para[-1]["end"], 5000)
|
|
self.assertTrue(all(a["end"] <= b["start"] + 1e-6 for a, b in zip(para, para[1:])))
|
|
|
|
def test_changed_text_falls_back_to_an_estimate_over_the_real_duration(self):
|
|
t = exact_timings()
|
|
t["segments"][3]["chars"] += 1 # text was edited after the audio was rendered
|
|
units, q = ee.plan_sentences(blocks(), 10000, t)
|
|
self.assertEqual(q, "estimated")
|
|
self.assertAlmostEqual(units[-1]["end"], 10000, places=3)
|
|
self.assertTrue(all(a["end"] <= b["start"] + 1e-6 for a, b in zip(units, units[1:])))
|
|
|
|
def test_no_record_and_interview_audio_are_estimated(self):
|
|
self.assertEqual(ee.plan_sentences(blocks(), 5000, None)[1], "estimated")
|
|
t = exact_timings(); t["interview"] = True
|
|
self.assertEqual(ee.plan_sentences(blocks(), 5000, t)[1], "estimated")
|
|
|
|
def test_empty_blocks_yield_nothing(self):
|
|
self.assertEqual(ee.plan_sentences([{"type": "body", "text": " "}], 1000, None), ([], "estimated"))
|
|
|
|
def test_clock_format(self):
|
|
self.assertEqual(ee.clock(0), "0:00:00.000")
|
|
self.assertEqual(ee.clock(3_723_456), "1:02:03.456")
|
|
self.assertEqual(ee.clock(-5), "0:00:00.000")
|
|
|
|
|
|
class BuildTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-epub-"))
|
|
self.mp3 = self.dir / "a.mp3"
|
|
self.mp3.write_bytes(b"ID3fakeaudio" * 50)
|
|
self.out = self.dir / "book.epub"
|
|
|
|
def article(self, **kw):
|
|
base = {"title": "Batterien & <Zukunft>", "author": "J", "blocks": blocks(), "mp3": self.mp3,
|
|
"duration_ms": 6000, "timings": exact_timings()}
|
|
base.update(kw)
|
|
return base
|
|
|
|
def build(self, arts=None, **kw):
|
|
return ee.build_epub(self.out, title=kw.pop("title", "c't"), issue=kw.pop("issue", "9/2026"),
|
|
language="de", articles=arts or [self.article()], identifier="urn:uuid:test", **kw)
|
|
|
|
def test_mimetype_is_first_and_stored_and_every_xml_part_is_well_formed(self):
|
|
info = self.build()
|
|
with zipfile.ZipFile(self.out) as z:
|
|
first = z.infolist()[0]
|
|
self.assertEqual((first.filename, first.compress_type), ("mimetype", zipfile.ZIP_STORED))
|
|
self.assertEqual(z.read("mimetype"), b"application/epub+zip")
|
|
for name in z.namelist():
|
|
if name.endswith((".xhtml", ".smil", ".opf", ".xml")):
|
|
ET.fromstring(z.read(name)) # raises on malformed XML
|
|
self.assertEqual(info["quality"], {"Batterien & <Zukunft>": "exact"})
|
|
|
|
def test_manifest_references_exist_ids_are_unique_and_overlay_is_linked(self):
|
|
self.build()
|
|
with zipfile.ZipFile(self.out) as z:
|
|
opf = ET.fromstring(z.read("OEBPS/content.opf"))
|
|
items = opf.findall(".//o:manifest/o:item", NS)
|
|
ids = [i.get("id") for i in items]
|
|
self.assertEqual(len(ids), len(set(ids)))
|
|
for i in items:
|
|
self.assertIn("OEBPS/" + i.get("href"), z.namelist())
|
|
xhtml = next(i for i in items if i.get("media-type") == "application/xhtml+xml" and i.get("id") == "a001")
|
|
self.assertEqual(xhtml.get("media-overlay"), "a001smil")
|
|
spine = [r.get("idref") for r in opf.findall(".//o:spine/o:itemref", NS)]
|
|
self.assertEqual(spine, ["a001"])
|
|
props = {m.get("property") for m in opf.findall(".//o:metadata/o:meta", NS)}
|
|
self.assertTrue({"media:duration", "media:active-class", "dcterms:modified"} <= props)
|
|
|
|
def test_every_smil_text_target_exists_and_clips_are_ordered(self):
|
|
self.build()
|
|
with zipfile.ZipFile(self.out) as z:
|
|
page = ET.fromstring(z.read("OEBPS/a001.xhtml"))
|
|
span_ids = {e.get("id") for e in page.iter("{%s}span" % NS["x"])}
|
|
smil = ET.fromstring(z.read("OEBPS/a001.smil"))
|
|
pars = smil.findall(".//s:par", NS)
|
|
self.assertEqual(len(pars), 6)
|
|
last_end = -1.0
|
|
for p in pars:
|
|
self.assertIn(p.find("s:text", NS).get("src").split("#")[1], span_ids)
|
|
a = p.find("s:audio", NS)
|
|
def secs(v):
|
|
h, m, s = v.split(":"); return int(h) * 3600 + int(m) * 60 + float(s)
|
|
b, e = secs(a.get("clipBegin")), secs(a.get("clipEnd"))
|
|
self.assertLess(b, e)
|
|
self.assertGreaterEqual(b, last_end - 1e-9)
|
|
last_end = e
|
|
self.assertEqual(a.get("src"), "audio/a001.mp3")
|
|
|
|
def test_hostile_text_is_escaped_not_interpreted(self):
|
|
bl = [{"type": "title", "text": "T"}, {"type": "body", "text": "Ein <script>alert(1)</script> & \"Test\"."}]
|
|
self.build([self.article(blocks=bl, timings=None, title="x</title><b>")])
|
|
with zipfile.ZipFile(self.out) as z:
|
|
raw = z.read("OEBPS/a001.xhtml").decode()
|
|
self.assertNotIn("<script>", raw)
|
|
self.assertNotIn("<b>", raw)
|
|
ET.fromstring(raw)
|
|
|
|
def test_headings_and_classes_follow_block_types_and_audio_is_stored_not_recompressed(self):
|
|
self.build()
|
|
with zipfile.ZipFile(self.out) as z:
|
|
page = z.read("OEBPS/a001.xhtml").decode()
|
|
self.assertIn("<h1>", page)
|
|
self.assertIn('<p class="byline">', page)
|
|
self.assertEqual(z.getinfo("OEBPS/audio/a001.mp3").compress_type, zipfile.ZIP_STORED)
|
|
self.assertEqual(z.read("OEBPS/audio/a001.mp3"), self.mp3.read_bytes())
|
|
|
|
def test_several_articles_keep_order_and_an_estimated_one_is_declared(self):
|
|
arts = [self.article(title="Erster"), self.article(title="Zweiter", timings=None)]
|
|
info = self.build(arts)
|
|
self.assertEqual(info["quality"], {"Erster": "exact", "Zweiter": "estimated"})
|
|
with zipfile.ZipFile(self.out) as z:
|
|
opf = z.read("OEBPS/content.opf").decode()
|
|
self.assertIn("estimated", re.search(r"<dc:description>(.*?)</dc:description>", opf).group(1))
|
|
nav = z.read("OEBPS/nav.xhtml").decode()
|
|
self.assertLess(nav.index("Erster"), nav.index("Zweiter"))
|
|
|
|
def test_nothing_to_package_is_an_error_and_leaves_no_file(self):
|
|
with self.assertRaises(ValueError):
|
|
ee.build_epub(self.out, title="t", issue="", language="de", articles=[])
|
|
with self.assertRaises(ValueError):
|
|
self.build([self.article(blocks=[{"type": "body", "text": " "}])])
|
|
self.assertEqual([p.name for p in self.dir.iterdir() if p.suffix in (".epub", ".part")], [])
|
|
|
|
def test_a_missing_audio_file_leaves_no_partial_book(self):
|
|
with self.assertRaises(OSError):
|
|
self.build([self.article(mp3=self.dir / "missing.mp3")])
|
|
self.assertEqual([p.name for p in self.dir.iterdir() if p.suffix in (".epub", ".part")], [])
|
|
|
|
|
|
class TimingRecordTests(unittest.TestCase):
|
|
def test_write_and_read_round_trip_without_any_text(self):
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-tm-"))
|
|
mp3 = d / "article-001.mp3"
|
|
mp3.write_bytes(b"x")
|
|
segs = [{"text": "Erster Satz. Zweiter."}, {"text": "Dritter."}]
|
|
mag.write_audio_timings(mp3, segs, [0, 0, 1], [(0, 900), (900, 1500), (1500, 2100)], interview=False)
|
|
rec = mag.read_audio_timings(mp3)
|
|
self.assertEqual(rec["duration_ms"], 2100)
|
|
self.assertEqual(rec["segments"], [{"chars": 21, "times": [[0, 900], [900, 1500]]}, {"chars": 8, "times": [[1500, 2100]]}])
|
|
self.assertNotIn("Satz", (d / "article-001.timings.json").read_text())
|
|
|
|
def test_missing_or_corrupt_record_reads_as_none(self):
|
|
d = Path(tempfile.mkdtemp(prefix="ttvc-tm-"))
|
|
mp3 = d / "x.mp3"
|
|
self.assertIsNone(mag.read_audio_timings(mp3))
|
|
(d / "x.timings.json").write_text("not json")
|
|
self.assertIsNone(mag.read_audio_timings(mp3))
|
|
(d / "x.timings.json").write_text(json.dumps({"version": 2, "segments": []}))
|
|
self.assertIsNone(mag.read_audio_timings(mp3))
|
|
|
|
|
|
class Mp3DurationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from unittest import mock
|
|
self.mock = mock
|
|
self.mp3 = Path(tempfile.mkdtemp(prefix="ttvc-dur-")) / "a.mp3"
|
|
self.mp3.write_bytes(b"not really audio")
|
|
|
|
def test_falls_back_to_ffprobe_when_mutagen_is_missing(self):
|
|
import subprocess, sys
|
|
done = subprocess.CompletedProcess([], 0, stdout="12.5\n")
|
|
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
|
self.mock.patch.object(mag.subprocess, "run", return_value=done):
|
|
self.assertEqual(mag.mp3_duration_seconds(self.mp3), 12.5)
|
|
|
|
def test_unreadable_audio_is_none_not_an_exception(self):
|
|
import subprocess, sys
|
|
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
|
self.mock.patch.object(mag.subprocess, "run", side_effect=subprocess.CalledProcessError(1, "ffprobe")):
|
|
self.assertIsNone(mag.mp3_duration_seconds(self.mp3))
|
|
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
|
self.mock.patch.object(mag.subprocess, "run", side_effect=FileNotFoundError()):
|
|
self.assertIsNone(mag.mp3_duration_seconds(self.mp3))
|
|
|
|
def test_garbage_or_zero_output_is_none(self):
|
|
import subprocess, sys
|
|
for out in ("N/A\n", "0\n", ""):
|
|
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
|
self.mock.patch.object(mag.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, stdout=out)):
|
|
self.assertIsNone(mag.mp3_duration_seconds(self.mp3))
|
|
|
|
|
|
class EndpointTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from unittest import mock
|
|
import routes.magazine as routes
|
|
self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-"))
|
|
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": "Test Mag", "issue": "1/2026"}).json()["id"]
|
|
self.d = self.lib / self.doc
|
|
|
|
def seed(self, approved=True, with_audio=True):
|
|
layout = {"page_count": 1, "pages": [{"page": 1, "width": 595, "height": 842}], "images": [], "blocks": [
|
|
{"id": "t", "page": 1, "bbox": [0, 0, 1, 1], "type": "title", "text": "Kopf", "reading_order": 0},
|
|
{"id": "b", "page": 1, "bbox": [0, 2, 1, 1], "type": "body", "text": "Ein Satz. Noch ein Satz.", "reading_order": 1}]}
|
|
(self.d / "layout.json").write_text(json.dumps(layout))
|
|
(self.d / "articles.json").write_text(json.dumps({"articles": [{"id": "article-001", "title": "Kopf", "pages": [1],
|
|
"blocks": ["t", "b"], "images": [], "status": "approved" if approved else "detected", "confidence": .9}]}))
|
|
import shutil
|
|
shutil.rmtree(self.d / "audio", ignore_errors=True)
|
|
if with_audio:
|
|
(self.d / "audio").mkdir()
|
|
(self.d / "audio" / "article-001.mp3").write_bytes(b"ID3" + b"0" * 200)
|
|
(self.d / "audio" / "article-001.timings.json").write_text(json.dumps(
|
|
{"version": 1, "interview": False, "duration_ms": 3000, "segments": [
|
|
{"chars": 4, "times": [[0, 500]]}, {"chars": 24, "times": [[500, 1700], [1700, 3000]]}]}))
|
|
|
|
def test_downloads_a_valid_epub(self):
|
|
self.seed()
|
|
r = self.client.get(f"/api/magazine/{self.doc}/epub")
|
|
self.assertEqual(r.status_code, 200, r.text[:200])
|
|
self.assertEqual(r.headers["content-type"], "application/epub+zip")
|
|
self.assertIn("Test%20Mag.epub", r.headers["content-disposition"])
|
|
import io
|
|
with zipfile.ZipFile(io.BytesIO(r.content)) as z:
|
|
self.assertEqual(z.namelist()[0], "mimetype")
|
|
self.assertIn("exact", z.read("OEBPS/content.opf").decode())
|
|
|
|
def test_unapproved_or_silent_articles_are_not_packaged(self):
|
|
self.seed(approved=False)
|
|
self.assertEqual(self.client.get(f"/api/magazine/{self.doc}/epub").status_code, 404)
|
|
self.seed(with_audio=False)
|
|
self.assertEqual(self.client.get(f"/api/magazine/{self.doc}/epub").status_code, 404)
|
|
|
|
def test_unknown_article_and_bad_document(self):
|
|
self.seed()
|
|
self.assertEqual(self.client.get(f"/api/magazine/{self.doc}/epub?article=nope").status_code, 404)
|
|
self.assertIn(self.client.get("/api/magazine/xx/epub").status_code, (400, 404))
|
|
|
|
def test_the_temporary_file_is_removed_after_sending(self):
|
|
import glob
|
|
self.seed()
|
|
before = set(glob.glob(os.path.join(tempfile.gettempdir(), "mag-epub-*")))
|
|
self.client.get(f"/api/magazine/{self.doc}/epub")
|
|
self.assertEqual(set(glob.glob(os.path.join(tempfile.gettempdir(), "mag-epub-*"))) - before, set())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|