474 lines
24 KiB
Python
474 lines
24 KiB
Python
"""Untitled magazine entries: continuation detection, merging, suggested titles, CLI and routes."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
_CONFIG = tempfile.mkdtemp(prefix="ttvc-magazine-untitled-test-")
|
|
os.environ.setdefault("CONFIG_DIR", _CONFIG)
|
|
|
|
from fastapi import HTTPException # noqa: E402
|
|
|
|
from core import magazine_untitled as mu # noqa: E402
|
|
|
|
|
|
def blk(bid, page, order, kind, text, x=40, w=240):
|
|
return {"id": bid, "page": page, "reading_order": order, "type": kind, "text": text,
|
|
"bbox": [x, 100 + order * 50, w, 40]}
|
|
|
|
|
|
def art(aid, title, blocks, pages, status="approved", **extra):
|
|
return {"id": aid, "title": title, "author": "", "pages": pages, "blocks": blocks, "images": [],
|
|
"status": status, "confidence": 0.55, "section": "", "continued_to": [], "continued_from": [],
|
|
"audio_status": "none", **extra}
|
|
|
|
|
|
LONG = "Die Entwickler setzen inzwischen auf Agenten, die selbstständig Code schreiben und prüfen. "
|
|
|
|
|
|
def sample():
|
|
"""Page 10 opens a feature, 11 and 12 continue it as 'Page N' stand-ins, 13 is an advert,
|
|
14 is a new signed-off article, 15 holds only a picture credit."""
|
|
blocks = [
|
|
blk("b10-1", 10, 0, "title", "Per KI von der Idee zum Programm"),
|
|
blk("b10-2", 10, 1, "byline", "Von Anna Beispiel"),
|
|
blk("b10-3", 10, 2, "body", LONG * 3 + "Die Agenten arbeiten dabei ohne Pause und schreiben Hunder-"),
|
|
blk("b11-1", 11, 0, "body", "te Zeilen am Tag. " + LONG * 3 + "Erst danach kommt der Mensch wieder ins"),
|
|
blk("b12-1", 12, 0, "body", "Spiel und prüft die Ergebnisse. " + LONG * 2),
|
|
blk("b12-2", 12, 1, "heading", "Blick nach vorn"),
|
|
blk("b12-3", 12, 2, "body", LONG * 2 + "Das Projekt ist damit abgeschlossen. (hag) ct"),
|
|
blk("b13-1", 13, 0, "ad", "Jetzt abonnieren! Drei Hefte gratis."),
|
|
blk("b14-1", 14, 0, "body", "Recht an der eigenen Stimme. " + LONG * 3),
|
|
blk("b15-1", 15, 0, "body", "Bild: Agentur"),
|
|
]
|
|
layout = {"page_count": 15, "pages": [{"page": p, "width": 600, "height": 800} for p in range(10, 16)],
|
|
"blocks": blocks, "images": [{"id": "img-13", "page": 13}]}
|
|
articles = [
|
|
art("article-001", "Per KI von der Idee zum Programm", ["b10-1", "b10-2", "b10-3"], [10]),
|
|
art("article-002", "Page 11", ["b11-1"], [11]),
|
|
art("article-003", "Page 12", ["b12-1", "b12-2", "b12-3"], [12]),
|
|
art("article-004", "Page 13", ["b13-1"], [13]),
|
|
art("article-005", "Page 14", ["b14-1"], [14]),
|
|
art("article-006", "Page 15", ["b15-1"], [15]),
|
|
]
|
|
return layout, articles
|
|
|
|
|
|
class PlaceholderTests(unittest.TestCase):
|
|
def test_placeholder_titles(self):
|
|
for t in ("", " ", "Page 33", "page 2", "Seite 7", "Untitled"):
|
|
self.assertTrue(mu.is_placeholder_title(t), t)
|
|
for t in ("Page 33 of the issue", "Inhalt", "Wer haftet, wenn KI einkauft?"):
|
|
self.assertFalse(mu.is_placeholder_title(t), t)
|
|
|
|
|
|
class ContinuationScoreTests(unittest.TestCase):
|
|
def score(self, tail, head, **kw):
|
|
args = dict(adjacent=True, similar_columns=True, starts_with_headline=False)
|
|
args.update(kw)
|
|
return mu.continuation_score(tail, {"type": "body", "text": head}, **args)
|
|
|
|
def test_lowercase_start_after_open_sentence_is_high(self):
|
|
s, ev = self.score("und schreiben Hunder-", "te Zeilen am Tag.")
|
|
self.assertGreaterEqual(s, mu.HIGH_AT)
|
|
self.assertIn("starts lower-case / mid-sentence", ev)
|
|
self.assertIn("previous text ends in a hyphenated word break", ev)
|
|
|
|
def test_signed_off_previous_article_blocks_a_merge(self):
|
|
s, ev = self.score("Das Projekt ist abgeschlossen. (hag) ct", "Recht an der eigenen Stimme.")
|
|
self.assertLess(s, mu.SUGGEST_AT)
|
|
self.assertIn("previous article is signed off", ev)
|
|
|
|
def test_own_headline_counts_against(self):
|
|
base, _ = self.score("kommt der Mensch wieder ins", "Spiel und prüft.")
|
|
with_head, ev = self.score("kommt der Mensch wieder ins", "Spiel und prüft.", starts_with_headline=True)
|
|
self.assertLess(with_head, base)
|
|
self.assertIn("starts with its own headline", ev)
|
|
|
|
def test_finished_sentence_far_away_is_low(self):
|
|
s, _ = self.score("Ende.", "Neuer Anfang.", adjacent=False, similar_columns=False)
|
|
self.assertLess(s, mu.SUGGEST_AT)
|
|
|
|
|
|
class ClassificationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.layout, self.articles = sample()
|
|
self.ann = mu.annotate(self.layout, self.articles, language="de")
|
|
|
|
def test_kinds(self):
|
|
kinds = {k: v["untitled_kind"] for k, v in self.ann.items()}
|
|
self.assertIsNone(kinds["article-001"])
|
|
self.assertEqual(kinds["article-002"], "continuation")
|
|
self.assertEqual(kinds["article-003"], "continuation")
|
|
self.assertEqual(kinds["article-004"], "advert")
|
|
self.assertEqual(kinds["article-006"], "junk")
|
|
# previous article signed off, so page 14 is not a continuation
|
|
self.assertNotEqual(kinds["article-005"], "continuation")
|
|
|
|
def test_suggestion_points_at_the_previous_entry_with_evidence(self):
|
|
s = self.ann["article-002"]["merge_suggestion"]
|
|
self.assertEqual(s["into"], "article-001")
|
|
self.assertEqual(s["level"], "high")
|
|
self.assertTrue(s["evidence"])
|
|
self.assertEqual(self.ann["article-003"]["merge_suggestion"]["into"], "article-002")
|
|
|
|
def test_display_fields(self):
|
|
a1, a4, a2 = self.ann["article-001"], self.ann["article-004"], self.ann["article-002"]
|
|
self.assertEqual(a1["display_title"], "Per KI von der Idee zum Programm")
|
|
self.assertEqual(a1["title_source"], "title")
|
|
self.assertEqual(a4["display_title"], "Anzeige")
|
|
self.assertEqual(a4["title_source"], "suggested")
|
|
self.assertEqual(a4["page_label"], "Seite 13")
|
|
self.assertNotIn("Page", a2["display_title"])
|
|
|
|
def test_previous_entry_skips_advert_pages_but_not_long_gaps(self):
|
|
layout, articles = sample()
|
|
# page 14 follows page 12 across the advert on 13 -> previous entry is page 12's
|
|
index = mu._Index(layout, articles)
|
|
prev, _ = mu.previous_entry(index, articles[4])
|
|
self.assertEqual(prev["id"], "article-003")
|
|
layout["blocks"].append(blk("b20-1", 20, 0, "body", "weiter geht es hier. " + LONG))
|
|
articles.append(art("article-007", "Page 20", ["b20-1"], [20]))
|
|
prev, _ = mu.previous_entry(mu._Index(layout, articles), articles[-1])
|
|
self.assertIsNone(prev)
|
|
|
|
def test_empty_stand_in(self):
|
|
layout, articles = sample()
|
|
articles.append(art("article-008", "Page 16", [], [16], status="detected"))
|
|
layout["images"].append({"id": "img-16", "page": 16})
|
|
ann = mu.annotate(layout, articles, language="de")
|
|
self.assertEqual(ann["article-008"]["untitled_kind"], "image_only")
|
|
self.assertEqual(ann["article-008"]["display_title"], "Bildseite")
|
|
|
|
def test_contents_heading(self):
|
|
layout, articles = sample()
|
|
layout["blocks"].append(blk("b16-1", 16, 0, "title", "Inhalt"))
|
|
layout["blocks"].append(blk("b16-2", 16, 1, "body", "Seite 12 Agenten. Seite 30 Recht. " * 3))
|
|
articles.append(art("article-009", "", ["b16-1", "b16-2"], [16]))
|
|
ann = mu.annotate(layout, articles, language="de")
|
|
self.assertEqual(ann["article-009"]["untitled_kind"], "contents")
|
|
self.assertEqual(ann["article-009"]["suggested_title"], "Inhalt")
|
|
|
|
def test_article_head_gets_its_headline(self):
|
|
layout, articles = sample()
|
|
layout["blocks"].append(blk("b17-1", 17, 0, "title", "Wer haftet, wenn KI einkauft?"))
|
|
layout["blocks"].append(blk("b17-2", 17, 1, "body", LONG * 3))
|
|
articles.append(art("article-010", "Page 17", ["b17-1", "b17-2"], [17]))
|
|
ann = mu.annotate(layout, articles, language="de")
|
|
self.assertEqual(ann["article-010"]["untitled_kind"], "article_head")
|
|
self.assertEqual(ann["article-010"]["suggested_title"], "Wer haftet, wenn KI einkauft?")
|
|
self.assertEqual(ann["article-010"]["suggested_title_source"], "heading")
|
|
|
|
|
|
class LlmCheckTests(unittest.TestCase):
|
|
def test_llm_only_consulted_in_the_ambiguous_band(self):
|
|
layout, articles = sample()
|
|
check = Mock(return_value=True)
|
|
index = mu._Index(layout, articles)
|
|
mu.merge_suggestion(index, articles[1], llm_check=check) # clear case: high already
|
|
check.assert_not_called()
|
|
# a page that starts a new capitalised sentence after a finished one: ambiguous
|
|
layout["blocks"][3]["text"] = "Neue Zeilen am Tag. " + LONG * 3
|
|
layout["blocks"][2]["text"] = LONG * 3
|
|
index = mu._Index(layout, articles)
|
|
without = mu.merge_suggestion(index, articles[1])
|
|
with_yes = mu.merge_suggestion(index, articles[1], llm_check=check)
|
|
check.assert_called_once()
|
|
self.assertIn("LLM: continues", with_yes["evidence"])
|
|
self.assertGreater(with_yes["confidence"], without["confidence"] if without else 0)
|
|
no = mu.merge_suggestion(index, articles[1], llm_check=Mock(return_value=False))
|
|
self.assertIsNone(no)
|
|
|
|
def test_llm_failure_is_ignored(self):
|
|
layout, articles = sample()
|
|
layout["blocks"][3]["text"] = "Neue Zeilen am Tag. " + LONG * 3
|
|
layout["blocks"][2]["text"] = LONG * 3
|
|
index = mu._Index(layout, articles)
|
|
plain = mu.merge_suggestion(index, articles[1])
|
|
broken = mu.merge_suggestion(index, articles[1], llm_check=Mock(side_effect=RuntimeError("down")))
|
|
self.assertEqual(plain, broken)
|
|
|
|
def test_continuation_check_parses_json(self):
|
|
self.assertTrue(mu.llm_continuation_check(lambda m: '{"continues": true}')("a", "b"))
|
|
self.assertFalse(mu.llm_continuation_check(lambda m: '```json\n{"continues": false}\n```')("a", "b"))
|
|
self.assertIsNone(mu.llm_continuation_check(lambda m: "maybe")("a", "b"))
|
|
|
|
|
|
class TitleTests(unittest.TestCase):
|
|
def test_deterministic_sentence_skips_a_leading_fragment_and_trims(self):
|
|
layout = {"blocks": [blk("x1", 5, 0, "body", "te Zeilen am Tag. Danach prüft der Mensch die "
|
|
"Ergebnisse sehr gründlich und mit viel Geduld jeden Tag.")]}
|
|
index = mu._Index(layout, [])
|
|
title, source = mu.deterministic_title(index, art("a", "Page 5", ["x1"], [5]), "untitled_article")
|
|
self.assertEqual(source, "sentence")
|
|
self.assertTrue(title.startswith("Danach prüft der Mensch"))
|
|
self.assertLessEqual(len(title.replace(" …", "").split()), mu.MAX_TITLE_WORDS)
|
|
|
|
def test_type_labels_follow_the_language(self):
|
|
index = mu._Index({"blocks": []}, [])
|
|
self.assertEqual(mu.deterministic_title(index, art("a", "", [], []), "advert", "de")[0], "Anzeige")
|
|
self.assertEqual(mu.deterministic_title(index, art("a", "", [], []), "advert", "en")[0], "Advertisement")
|
|
self.assertEqual(mu.page_label(33, "en"), "page 33")
|
|
|
|
def test_validate_llm_title(self):
|
|
text = "Agenten schreiben Code, Menschen prüfen die Ergebnisse"
|
|
self.assertEqual(mu.validate_llm_title('"Agenten schreiben Code"', text), "Agenten schreiben Code")
|
|
self.assertEqual(mu.validate_llm_title("<think>hm</think>**Agenten prüfen Ergebnisse**", text),
|
|
"Agenten prüfen Ergebnisse")
|
|
for bad in ("", "Titel: Agenten schreiben Code", "Hier ist ein Titel", "Agenten",
|
|
"Eins zwei drei vier fünf sechs sieben acht neun", "Zeile eins\nZeile zwei",
|
|
"Quantencomputer erobern Weltraum", "A" * 80):
|
|
self.assertIsNone(mu.validate_llm_title(bad, text), bad)
|
|
|
|
def test_llm_title_falls_back_on_errors(self):
|
|
text = "Agenten schreiben Code"
|
|
self.assertEqual(mu.llm_title(lambda m: "Agenten schreiben Code", text, "de"), "Agenten schreiben Code")
|
|
self.assertIsNone(mu.llm_title(Mock(side_effect=TimeoutError()), text, "de"))
|
|
|
|
def test_prompt_is_strict_and_in_the_magazine_language(self):
|
|
msgs = mu._title_messages("Text", "de")
|
|
self.assertIn("ausschließlich", msgs[0]["content"])
|
|
self.assertIn("Deutsch", msgs[0]["content"])
|
|
self.assertIn("unvertrauenswürdige", msgs[0]["content"])
|
|
|
|
def test_cached_llm_title_is_used_only_while_the_text_matches(self):
|
|
layout, articles = sample()
|
|
index = mu._Index(layout, articles)
|
|
cache = {"entries": {"article-005": {"title": "Recht an der eigenen Stimme",
|
|
"hash": mu.text_hash(index, articles[4])}}}
|
|
ann = mu.annotate(layout, articles, cache, language="de")
|
|
self.assertEqual(ann["article-005"]["suggested_title"], "Recht an der eigenen Stimme")
|
|
self.assertEqual(ann["article-005"]["suggested_title_source"], "llm")
|
|
cache["entries"]["article-005"]["hash"] = "stale"
|
|
ann = mu.annotate(layout, articles, cache, language="de")
|
|
self.assertNotEqual(ann["article-005"]["suggested_title_source"], "llm")
|
|
|
|
|
|
class MergeTests(unittest.TestCase):
|
|
def test_merge_orders_blocks_and_marks_audio_stale(self):
|
|
layout, articles = sample()
|
|
articles[0]["blocks"] = ["b10-3", "b10-1", "b10-2"]
|
|
rec = mu.merge_entries(layout, articles, "article-002", "article-001")
|
|
merged = articles[0]
|
|
self.assertEqual(merged["blocks"], ["b10-1", "b10-2", "b10-3", "b11-1"])
|
|
self.assertEqual(merged["pages"], [10, 11])
|
|
self.assertTrue(merged["audio_stale"])
|
|
self.assertEqual(merged["merged_from"], ["article-002"])
|
|
self.assertNotIn("article-002", [a["id"] for a in articles])
|
|
self.assertEqual(rec["status"], "approved")
|
|
self.assertEqual(merged["title"], "Per KI von der Idee zum Programm")
|
|
|
|
def test_merge_never_replaces_a_real_title_and_adopts_one_for_a_placeholder(self):
|
|
layout, articles = sample()
|
|
articles[1]["title"] = "Echter Titel"
|
|
mu.merge_entries(layout, articles, "article-002", "article-001")
|
|
self.assertEqual(articles[0]["title"], "Per KI von der Idee zum Programm")
|
|
layout, articles = sample()
|
|
articles[2]["title"] = "Durchstarten"
|
|
mu.merge_entries(layout, articles, "article-003", "article-002")
|
|
self.assertEqual(next(a for a in articles if a["id"] == "article-002")["title"], "Durchstarten")
|
|
|
|
def test_unreviewed_source_makes_the_result_unreviewed(self):
|
|
layout, articles = sample()
|
|
articles[1]["status"] = "detected"
|
|
mu.merge_entries(layout, articles, "article-002", "article-001")
|
|
self.assertEqual(articles[0]["status"], "detected")
|
|
|
|
def test_bad_ids(self):
|
|
layout, articles = sample()
|
|
with self.assertRaises(ValueError):
|
|
mu.merge_entries(layout, articles, "article-001", "article-001")
|
|
with self.assertRaises(KeyError):
|
|
mu.merge_entries(layout, articles, "article-404", "article-001")
|
|
|
|
|
|
class PlanTests(unittest.TestCase):
|
|
def test_chain_collapses_into_the_root_and_input_is_untouched(self):
|
|
layout, articles = sample()
|
|
snapshot = json.dumps(articles)
|
|
result = mu.plan(layout, articles)
|
|
self.assertEqual(json.dumps(articles), snapshot)
|
|
ids = [a["id"] for a in result["articles"]]
|
|
self.assertNotIn("article-002", ids)
|
|
self.assertNotIn("article-003", ids)
|
|
root = result["articles"][0]
|
|
self.assertEqual(root["pages"], [10, 11, 12])
|
|
self.assertEqual(root["merged_from"], ["article-002", "article-003"])
|
|
titled = {t["id"] for t in result["titles"]}
|
|
self.assertNotIn("article-001", titled)
|
|
self.assertIn("article-004", titled)
|
|
|
|
def test_duplicate_title_tail_is_merged_and_named(self):
|
|
layout, articles = sample()
|
|
articles[0]["title"] = "Page 10" # the head is a stand-in ...
|
|
articles[2]["title"] = "Per KI von der Idee zum Programm" # ... and the tail kept the title
|
|
result = mu.plan(layout, articles)
|
|
root = result["articles"][0]
|
|
self.assertEqual(root["pages"], [10, 11, 12])
|
|
self.assertEqual(root["title"], "Per KI von der Idee zum Programm")
|
|
|
|
def test_threshold_controls_merges(self):
|
|
layout, articles = sample()
|
|
self.assertEqual(mu.plan(layout, articles, min_confidence=1.01)["merges"], [])
|
|
|
|
def test_excluded_entries_are_left_untouched(self):
|
|
layout, articles = sample()
|
|
base = mu.plan(layout, articles)
|
|
self.assertTrue(base["merges"])
|
|
src = base["merges"][0]["source"]
|
|
result = mu.plan(layout, articles, exclude=[src])
|
|
self.assertTrue(all(m["source"] != src for m in result["merges"]))
|
|
self.assertNotIn(src, {t["id"] for t in result["titles"]})
|
|
kept = next(a for a in result["articles"] if a["id"] == src)
|
|
orig = next(a for a in articles if a["id"] == src)
|
|
self.assertEqual(kept["title"], orig["title"])
|
|
|
|
def test_apply_titles_only_touches_placeholders(self):
|
|
_, articles = sample()
|
|
n = mu.apply_titles(articles, [{"id": "article-001", "after": "X"}, {"id": "article-004", "after": "Anzeige",
|
|
"source": "type"}])
|
|
self.assertEqual(n, 1)
|
|
self.assertEqual(articles[0]["title"], "Per KI von der Idee zum Programm")
|
|
self.assertEqual(articles[3]["title"], "Anzeige")
|
|
self.assertEqual(articles[3]["title_placeholder"], "Page 13")
|
|
|
|
|
|
class CliTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = Path(tempfile.mkdtemp(prefix="untitled-cli-"))
|
|
layout, articles = sample()
|
|
(self.dir / "layout.json").write_text(json.dumps(layout), "utf-8")
|
|
(self.dir / "articles.json").write_text(json.dumps({"articles": articles}), "utf-8")
|
|
(self.dir / "audio").mkdir()
|
|
(self.dir / "audio" / "article-002.mp3").write_bytes(b"ID3")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def test_dry_run_writes_nothing(self):
|
|
from scripts import magazine_untitled as cli
|
|
before = sorted(p.name for p in self.dir.rglob("*"))
|
|
text = (self.dir / "articles.json").read_text("utf-8")
|
|
s = cli.run(self.dir, apply=False, min_confidence=mu.HIGH_AT)
|
|
self.assertEqual(sorted(p.name for p in self.dir.rglob("*")), before)
|
|
self.assertEqual((self.dir / "articles.json").read_text("utf-8"), text)
|
|
self.assertEqual(len(s["merges"]), 2)
|
|
self.assertEqual(s["audio_kept_orphaned"], ["article-002"])
|
|
|
|
def test_apply_backs_up_merges_and_keeps_audio(self):
|
|
from scripts import magazine_untitled as cli
|
|
s = cli.run(self.dir, apply=True, min_confidence=mu.HIGH_AT)
|
|
backups = list(self.dir.glob("articles.backup-untitled-*.json"))
|
|
self.assertEqual(len(backups), 1)
|
|
self.assertEqual(len(json.loads(backups[0].read_text("utf-8"))["articles"]), 6)
|
|
after = json.loads((self.dir / "articles.json").read_text("utf-8"))["articles"]
|
|
self.assertEqual(len(after), 4)
|
|
self.assertTrue(after[0]["audio_stale"])
|
|
self.assertEqual(next(a for a in after if a["id"] == "article-004")["title"], "Anzeige")
|
|
self.assertTrue((self.dir / "audio" / "article-002.mp3").is_file())
|
|
self.assertGreaterEqual(s["titles_applied"], 1)
|
|
|
|
def test_llm_titles_with_mocked_chat(self):
|
|
from scripts import magazine_untitled as cli
|
|
chat = Mock(return_value="Recht an der eigenen Stimme")
|
|
s = cli.run(self.dir, apply=False, min_confidence=mu.HIGH_AT, chat=chat)
|
|
by_id = {t["id"]: t for t in s["titles"]}
|
|
self.assertEqual(by_id["article-005"]["source"], "llm")
|
|
self.assertEqual(by_id["article-005"]["after"], "Recht an der eigenen Stimme")
|
|
|
|
|
|
class RouteTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.dir = Path(tempfile.mkdtemp(prefix="untitled-routes-"))
|
|
layout, articles = sample()
|
|
(self.dir / "layout.json").write_text(json.dumps(layout), "utf-8")
|
|
(self.dir / "articles.json").write_text(json.dumps({"articles": articles}), "utf-8")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def articles(self):
|
|
return json.loads((self.dir / "articles.json").read_text("utf-8"))["articles"]
|
|
|
|
async def test_articles_listing_carries_display_fields(self):
|
|
from routes.magazine import magazine_articles
|
|
with patch("routes.magazine._require_magazine", return_value=(self.dir, {})):
|
|
out = await magazine_articles("a" * 32)
|
|
a4 = next(a for a in out["articles"] if a["id"] == "article-004")
|
|
self.assertEqual(a4["display_title"], "Anzeige")
|
|
self.assertEqual(a4["page_label"], "Seite 13")
|
|
a2 = next(a for a in out["articles"] if a["id"] == "article-002")
|
|
self.assertEqual(a2["merge_suggestion"]["into"], "article-001")
|
|
self.assertEqual(self.articles()[1]["title"], "Page 11") # nothing saved
|
|
|
|
async def test_merge_into_previous_checks_the_target_and_is_saved(self):
|
|
from routes.magazine import magazine_merge_into_previous
|
|
request = AsyncMock()
|
|
request.body.return_value = b"{}"
|
|
with patch("routes.magazine._require_magazine", return_value=(self.dir, {})):
|
|
request.json.return_value = {"into": "article-005"}
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_merge_into_previous("a" * 32, "article-002", request)
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
request.json.return_value = {"into": "article-001"}
|
|
out = await magazine_merge_into_previous("a" * 32, "article-002", request)
|
|
self.assertEqual(out["article"]["pages"], [10, 11])
|
|
with self.assertRaises(HTTPException) as err: # a titled, finished article: nothing proposed
|
|
await magazine_merge_into_previous("a" * 32, "article-001", request)
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
saved = self.articles()
|
|
self.assertEqual(len(saved), 5)
|
|
self.assertTrue(saved[0]["audio_stale"])
|
|
|
|
async def test_accept_title_never_overwrites_a_real_title(self):
|
|
from routes.magazine import magazine_accept_title
|
|
request = AsyncMock()
|
|
request.body.return_value = b"{}"
|
|
request.json.return_value = {}
|
|
with patch("routes.magazine._require_magazine", return_value=(self.dir, {})):
|
|
with self.assertRaises(HTTPException) as err:
|
|
await magazine_accept_title("a" * 32, "article-001", request)
|
|
self.assertEqual(err.exception.status_code, 409)
|
|
out = await magazine_accept_title("a" * 32, "article-004", request)
|
|
self.assertEqual(out["article"]["title"], "Anzeige")
|
|
self.assertEqual(self.articles()[3]["title"], "Anzeige")
|
|
self.assertEqual(self.articles()[3]["title_origin"], "type")
|
|
|
|
async def test_untitled_report(self):
|
|
from routes.magazine import magazine_untitled_report
|
|
with patch("routes.magazine._require_magazine", return_value=(self.dir, {})):
|
|
out = await magazine_untitled_report("a" * 32)
|
|
self.assertEqual(out["kinds"]["continuation"], 2)
|
|
self.assertEqual(out["merge_suggestions"], 2)
|
|
|
|
def test_fresh_render_clears_audio_stale(self):
|
|
from routes.magazine import _clear_audio_stale
|
|
arts = self.articles()
|
|
arts[0].update(audio_stale=True, audio_stale_reason="merged")
|
|
(self.dir / "articles.json").write_text(json.dumps({"articles": arts}), "utf-8")
|
|
_clear_audio_stale(self.dir, "article-001")
|
|
self.assertNotIn("audio_stale", self.articles()[0])
|
|
|
|
def test_suggest_titles_job_uses_the_llm_and_caches(self):
|
|
from routes import magazine as rm
|
|
resp = Mock(status_code=200)
|
|
resp.json.return_value = {"choices": [{"message": {"content": "Recht an der eigenen Stimme"}}]}
|
|
with patch.object(rm, "_LIB", self.dir.parent), \
|
|
patch.object(rm, "_load_settings", return_value={}), \
|
|
patch.object(rm, "_llm_chat_fast", return_value=resp), \
|
|
patch.object(rm, "background_job_update") as upd:
|
|
rm._suggest_titles_sync(self.dir.name, "job")
|
|
self.assertEqual(upd.call_args.kwargs.get("stage"), "done")
|
|
cache = mu.load_title_cache(self.dir)
|
|
self.assertEqual(cache["entries"]["article-005"]["title"], "Recht an der eigenen Stimme")
|
|
self.assertEqual(self.articles()[4]["title"], "Page 14") # titles themselves unchanged
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|