test: pytest-Suite ohne Ollama-Abhängigkeit

33 Tests mit Fakes für die Modellaufrufe: Abschnittsbildung, Glossar
(jede Ersetzung trifft die Testdaten), Streaming-Auswertung und
Fehlerfälle, Satztrennung, Absatzgliederung inkl. Teilen/Verbinden,
Zwischenspeicher und Kommandozeile.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

README.md

@@ -75,6 +75,17 @@ Ersetzungen ohne Treffer werden gemeldet (Tippfehler im Glossar).
 
 Ein Lauf dauert mit `qwen3:32b` vollständig auf einer RTX 3090 etwa 6 Minuten.
 
+## Tests
+
+```bash
+python3 -m pytest
+```
+
+Die Tests brauchen kein Ollama: Modellaufrufe werden durch Fakes ersetzt. Geprüft
+werden Abschnittsbildung, Glossar (inkl. „jede Ersetzung in `ramstein.glossar.txt`
+trifft die Testdaten“), Streaming-Auswertung, Absatzgliederung (Text bleibt
+unverändert, 3–10 Sätze), Zwischenspeicher/Fortsetzen und die Kommandozeile.
+
 ## Zwischenstände
 
 Jeder fertige Abschnitt und jede Absatzgliederung wird in `<Ausgabe>.parts/`

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+pythonpath = .

tests/test_rede_korrigieren.py

@@ -0,0 +1,315 @@
+"""Tests for rede_korrigieren.py; the Ollama model is always replaced by fakes."""
+import io
+import json
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+
+import rede_korrigieren as rk
+
+REPO = Path(__file__).resolve().parent.parent
+SAMPLE = REPO / "testdaten" / "ramstein.de.txt"
+GLOSSARY = REPO / "ramstein.glossar.txt"
+
+
+def flat(text):
+    return " ".join(text.split())
+
+
+class FakeResponse(io.BytesIO):
+    """Stands in for urlopen(): iterable by lines, readable, context manager."""
+
+
+def fake_urlopen(body):
+    def urlopen(request, timeout=None):
+        return FakeResponse(body)
+    return urlopen
+
+
+def identity_correct(chunk, index, system):
+    return chunk.strip()
+
+
+def every_nth_start(n):
+    """Fake find_paragraph_starts: a paragraph starts every n sentences."""
+    def find(sentences, first, system):
+        last = first + len(sentences) - 1
+        return {k for k in range(max(first, 2), last + 1) if (k - 1) % n == 0}
+    return find
+
+
+def no_model_calls(*args, **kwargs):
+    raise AssertionError("Modell wurde unerwartet aufgerufen")
+
+
+def run_main(monkeypatch, *args):
+    monkeypatch.setattr(sys, "argv", ["rede_korrigieren.py", *map(str, args)])
+    rk.main()
+
+
+# === Chunking and word counts ===
+
+def test_split_chunks_keeps_text_and_ends_at_sentence():
+    text = flat(SAMPLE.read_text(encoding="utf-8"))
+    chunks = rk.split_chunks(text)
+
+    assert "".join(chunks) == text
+    assert len(chunks) > 1
+    for chunk in chunks[:-1]:
+        assert rk.SENTENCE_END.search(chunk.rstrip())
+        assert rk.CHUNK_SIZE <= len(chunk.split()) <= rk.CHUNK_SIZE + rk.MAX_EXTRA_WORDS
+
+
+def test_split_chunks_cuts_at_limit_without_sentence_end(monkeypatch):
+    monkeypatch.setattr(rk, "CHUNK_SIZE", 3)
+    monkeypatch.setattr(rk, "MAX_EXTRA_WORDS", 2)
+    chunks = rk.split_chunks("a b c d e f g h")
+    assert [len(c.split()) for c in chunks] == [5, 3]
+
+
+def test_split_chunks_empty():
+    assert rk.split_chunks("") == []
+
+
+def test_count_words_ignores_review_marks():
+    assert rk.count_words("Ramstein [prüfen: Rammstein] liegt") == 2
+
+
+# === Glossary ===
+
+def test_read_glossary(tmp_path):
+    path = tmp_path / "g.txt"
+    path.write_text(
+        "# Kommentar\n\nKremmel => Kreml\n  Willly Brand =>  Willy Brandt \n"
+        "Multi-Domain Task Force\nkaputt =>\n",
+        encoding="utf-8",
+    )
+    terms, replacements = rk.read_glossary(path)
+    assert terms == ["Multi-Domain Task Force"]
+    assert replacements == [("Kremmel", "Kreml"), ("Willly Brand", "Willy Brandt")]
+
+
+def test_apply_replacements_whole_words_in_order(capsys):
+    text = "Egon Bars Politik, Egon Bar. Barbara bleibt."
+    result = rk.apply_replacements(
+        text, [("Egon Bars", "Egon Bahrs"), ("Egon Bar", "Egon Bahr"), ("Fehlt", "x")]
+    )
+    assert result == "Egon Bahrs Politik, Egon Bahr. Barbara bleibt."
+    assert "Ersetzung ohne Treffer: „Fehlt“" in capsys.readouterr().out
+
+
+def test_build_system_appends_terms():
+    assert rk.build_system([]) == rk.SYSTEM
+    system = rk.build_system(["Kreml", "Egon Bahr"])
+    assert system.startswith(rk.SYSTEM)
+    assert "- Kreml\n- Egon Bahr" in system
+
+
+def test_sample_glossary_replacements_all_match(capsys):
+    """Every rule in ramstein.glossar.txt must hit the sample transcript."""
+    _, replacements = rk.read_glossary(GLOSSARY)
+    rk.apply_replacements(flat(SAMPLE.read_text(encoding="utf-8")), replacements)
+    assert "ohne Treffer" not in capsys.readouterr().out
+
+
+# === Model calls (urlopen faked) ===
+
+def stream_body(pieces, done_reason="stop"):
+    lines = [{"message": {"content": p}, "done": False} for p in pieces]
+    lines.append({"message": {"content": ""}, "done": True, "done_reason": done_reason,
+                  "eval_count": 10, "eval_duration": 1_000_000_000})
+    return b"\n".join(json.dumps(line).encode("utf-8") for line in lines) + b"\n"
+
+
+def test_correct_chunk_joins_stream(monkeypatch, capsys):
+    monkeypatch.setattr(rk.urllib.request, "urlopen",
+                        fake_urlopen(stream_body(["Ich grüße ", "Sie. ", "\n"])))
+    assert rk.correct_chunk("ich grüße sie", 1, rk.SYSTEM) == "Ich grüße Sie."
+    assert "10.0 Tokens/s" in capsys.readouterr().out
+
+
+def test_correct_chunk_truncated(monkeypatch):
+    monkeypatch.setattr(rk.urllib.request, "urlopen",
+                        fake_urlopen(stream_body(["Text"], done_reason="length")))
+    with pytest.raises(RuntimeError, match="abgeschnitten"):
+        rk.correct_chunk("text", 1, rk.SYSTEM)
+
+
+def test_correct_chunk_empty(monkeypatch):
+    monkeypatch.setattr(rk.urllib.request, "urlopen", fake_urlopen(stream_body([" "])))
+    with pytest.raises(RuntimeError, match="leere Modellantwort"):
+        rk.correct_chunk("text", 1, rk.SYSTEM)
+
+
+def test_correct_chunk_ollama_error(monkeypatch):
+    monkeypatch.setattr(rk.urllib.request, "urlopen",
+                        fake_urlopen(b'{"error": "model not found"}\n'))
+    with pytest.raises(RuntimeError, match="model not found"):
+        rk.correct_chunk("text", 1, rk.SYSTEM)
+
+
+def test_find_paragraph_starts_filters_numbers(monkeypatch):
+    content = json.dumps({"absatzanfaenge": [1, 10, 12, 15, 20, "x"]})
+    body = json.dumps({"message": {"content": content}}).encode("utf-8")
+    monkeypatch.setattr(rk.urllib.request, "urlopen", fake_urlopen(body))
+    # Window holds sentences 10..15: 10 and 15 allowed, 1/20/"x" dropped
+    assert rk.find_paragraph_starts(["s"] * 6, 10, rk.PARAGRAPH_SYSTEM) == {10, 12, 15}
+
+
+def test_find_paragraph_starts_never_first_sentence(monkeypatch):
+    content = json.dumps({"absatzanfaenge": [1, 3]})
+    body = json.dumps({"message": {"content": content}}).encode("utf-8")
+    monkeypatch.setattr(rk.urllib.request, "urlopen", fake_urlopen(body))
+    assert rk.find_paragraph_starts(["s"] * 5, 1, rk.PARAGRAPH_SYSTEM) == {3}
+
+
+def test_find_paragraph_starts_bad_json(monkeypatch):
+    body = json.dumps({"message": {"content": "kein JSON"}}).encode("utf-8")
+    monkeypatch.setattr(rk.urllib.request, "urlopen", fake_urlopen(body))
+    with pytest.raises(RuntimeError, match="unbrauchbare Antwort"):
+        rk.find_paragraph_starts(["s"] * 3, 1, rk.PARAGRAPH_SYSTEM)
+
+
+# === Sentences and paragraphs ===
+
+def test_split_sentences_keeps_closing_quotes():
+    text = "Er sagte: „Ja.“ Dann ging er. Wirklich?! [applaus] Gut… Ende."
+    sentences = rk.split_sentences(text)
+    assert sentences == ["Er sagte: „Ja.“", "Dann ging er.", "Wirklich?!",
+                         "[applaus] Gut…", "Ende."]
+    assert " ".join(sentences) == text
+
+
+def test_group_paragraphs():
+    assert rk.group_paragraphs(7, {3, 6}) == [(1, 2), (3, 5), (6, 7)]
+    assert rk.group_paragraphs(3, set()) == [(1, 3)]
+    assert rk.group_paragraphs(3, {1, 9}) == [(1, 3)]
+
+
+@pytest.mark.parametrize("paragraphs, expected", [
+    ([(1, 2), (3, 6)], [(1, 6)]),                          # short first joins next
+    ([(1, 4), (5, 5), (6, 9)], [(1, 5), (6, 9)]),          # single joins previous
+    ([(1, 9), (10, 10), (11, 14)], [(1, 10), (11, 14)]),   # fits exactly at maximum
+    ([(1, 10), (11, 11), (12, 15)], [(1, 10), (11, 15)]),  # previous full: joins next
+    ([(1, 10), (11, 11)], [(1, 10), (11, 11)]),            # no room anywhere
+])
+def test_merge_short_paragraphs(paragraphs, expected):
+    assert rk.merge_short_paragraphs(paragraphs) == expected
+
+
+def paragraph_lengths(text):
+    return [len(rk.split_sentences(p)) for p in text.split("\n\n")]
+
+
+def test_build_paragraphs_keeps_text_and_limits(tmp_path, monkeypatch):
+    text = flat(SAMPLE.read_text(encoding="utf-8"))
+    monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5))
+
+    result = rk.build_paragraphs(text, tmp_path)
+
+    assert flat(result) == text
+    lengths = paragraph_lengths(result)
+    assert min(lengths) >= rk.MIN_PARAGRAPH_SENTENCES
+    assert max(lengths) <= rk.MAX_PARAGRAPH_SENTENCES
+
+
+def test_build_paragraphs_splits_long_and_falls_back(tmp_path, monkeypatch, capsys):
+    text = flat(SAMPLE.read_text(encoding="utf-8"))
+
+    def find(sentences, first, system):
+        if system == rk.PARAGRAPH_SYSTEM:
+            return every_nth_start(25)(sentences, first, system)  # far too long
+        if first < 100:
+            return every_nth_start(6)(sentences, first, system)   # split works
+        return set()                                              # split fails
+
+    monkeypatch.setattr(rk, "find_paragraph_starts", find)
+    result = rk.build_paragraphs(text, tmp_path)
+
+    out = capsys.readouterr().out
+    assert "Teile langen Absatz" in out
+    assert "gleichmäßig" in out
+    assert flat(result) == text
+    assert max(paragraph_lengths(result)) <= rk.MAX_PARAGRAPH_SENTENCES
+
+
+def test_build_paragraphs_uses_cache(tmp_path, monkeypatch):
+    text = flat(SAMPLE.read_text(encoding="utf-8"))
+    monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5))
+    first = rk.build_paragraphs(text, tmp_path)
+
+    monkeypatch.setattr(rk, "find_paragraph_starts", no_model_calls)
+    assert rk.build_paragraphs(text, tmp_path) == first
+
+
+# === Command line and end-to-end ===
+
+@pytest.fixture
+def sample_copy(tmp_path):
+    path = tmp_path / "ramstein.de.txt"
+    shutil.copy(SAMPLE, path)
+    return path
+
+
+def test_main_default_output_resume_and_glossary(sample_copy, monkeypatch, capsys):
+    monkeypatch.setattr(rk, "correct_chunk", identity_correct)
+    monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5))
+    run_main(monkeypatch, "-i", sample_copy, "-g", GLOSSARY)
+
+    target = sample_copy.with_name("ramstein.de.korrigiert.txt")
+    result = target.read_text(encoding="utf-8")
+    assert "Ramstein" in result and "Rammstein" not in result
+    assert "Kreml" in result and "1983" in result
+    assert (sample_copy.parent / "ramstein.de.korrigiert.parts").is_dir()
+    assert "WARNUNG" not in capsys.readouterr().out
+
+    # Second run: everything cached, no model calls
+    monkeypatch.setattr(rk, "correct_chunk", no_model_calls)
+    monkeypatch.setattr(rk, "find_paragraph_starts", no_model_calls)
+    run_main(monkeypatch, "-i", sample_copy, "-g", GLOSSARY)
+    assert target.read_text(encoding="utf-8") == result
+
+
+def test_main_output_option_creates_directories(sample_copy, monkeypatch):
+    monkeypatch.setattr(rk, "correct_chunk", identity_correct)
+    monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5))
+    target = sample_copy.parent / "neu" / "unter" / "rede.txt"
+    run_main(monkeypatch, "-i", sample_copy, "-o", target)
+    assert target.is_file()
+    assert (target.parent / "rede.parts").is_dir()
+
+
+def test_main_warns_on_word_count_deviation(sample_copy, monkeypatch, capsys):
+    def shortening(chunk, index, system):
+        words = chunk.split()
+        return " ".join(words[: int(len(words) * 0.9)]) if index == 2 else chunk.strip()
+
+    monkeypatch.setattr(rk, "correct_chunk", shortening)
+    monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5))
+    run_main(monkeypatch, "-i", sample_copy)
+    out = capsys.readouterr().out
+    assert "WARNUNG: Abschnitt 2:" in out
+    assert "1 Abschnitt(e) mit auffälliger Wortzahl" in out
+
+
+def test_main_empty_input(tmp_path, monkeypatch):
+    empty = tmp_path / "leer.txt"
+    empty.write_text("  \n", encoding="utf-8")
+    with pytest.raises(SystemExit, match="leer"):
+        run_main(monkeypatch, "-i", empty)
+
+
+@pytest.mark.parametrize("args, message", [
+    ([], "required"),
+    (["-i", "gibtsnicht.txt"], "Eingabedatei nicht gefunden"),
+    (["-i", "{sample}", "-o", "{sample}"], "nicht die Eingabedatei"),
+    (["-i", "{sample}", "-g", "gibtsnicht.txt"], "Glossar nicht gefunden"),
+])
+def test_cli_errors(sample_copy, monkeypatch, capsys, args, message):
+    args = [a.format(sample=sample_copy) for a in args]
+    with pytest.raises(SystemExit) as exc:
+        run_main(monkeypatch, *args)
+    assert exc.value.code == 2
+    assert message in capsys.readouterr().err