diff --git a/README.md b/README.md index 515ec27..415a659 100644 --- a/README.md +++ b/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 `.parts/` diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..4584de7 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +pythonpath = . diff --git a/tests/test_rede_korrigieren.py b/tests/test_rede_korrigieren.py new file mode 100644 index 0000000..171a819 --- /dev/null +++ b/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