"""Offline-Tests fuer die Kernlogik (kein Netzwerk noetig)."""
from __future__ import annotations
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from src.model import ComponentType, Document, ReviewStatus, Scope # noqa: E402
from src.review.staging import ReviewGate # noqa: E402
from src.transformers.content_filter import filter_document # noqa: E402
from src.transformers.tagger import apply_tags # noqa: E402
def make_doc(body="Ein ausreichend langer Inhalt fuer den Test. " * 3, scope=Scope.EXTERN, domain="support"):
return Document(
tool="tool-x",
scope=scope,
domain=domain,
title="Mein Titel",
body_md=body,
url="https://example.com/x",
component_type=ComponentType.PAGE,
tags=["faq"],
)
def test_scope_visibility():
assert Scope.ALLGEMEIN.visible_extern and Scope.ALLGEMEIN.visible_intern
assert Scope.INTERN.visible_intern and not Scope.INTERN.visible_extern
assert Scope.EXTERN.visible_extern and not Scope.EXTERN.visible_intern
def test_tagger_adds_tool_and_scope_and_dedupes():
doc = make_doc()
apply_tags(doc)
assert "tool:tool-x" in doc.tags
assert "scope:extern" in doc.tags
assert "faq" in doc.tags
assert len(doc.tags) == len(set(doc.tags))
def test_frontmatter_roundtrip():
doc = make_doc()
apply_tags(doc)
md = doc.to_markdown()
assert md.startswith("---")
assert 'tool: "tool-x"' in md
assert 'scope: "extern"' in md
assert "content_hash:" in md
def test_confidential_in_extern_is_pending():
rules = {"blacklist_keywords": ["Vertraulich"], "regex_redact": []}
doc = make_doc(body="Dies ist Vertraulich und sollte nie extern erscheinen. " * 3)
filter_document(doc, rules)
assert doc.review_status == ReviewStatus.PENDING
def test_confidential_in_intern_is_pending():
rules = {"blacklist_keywords": ["Vertraulich"], "regex_redact": []}
doc = make_doc(body="Interne Notiz: Vertraulich. " * 5, scope=Scope.INTERN)
filter_document(doc, rules)
assert doc.review_status == ReviewStatus.PENDING
def test_clean_doc_approved():
rules = {"blacklist_keywords": ["Vertraulich"], "regex_redact": []}
doc = make_doc()
filter_document(doc, rules)
assert doc.review_status == ReviewStatus.APPROVED
def test_trusted_source_not_pending():
# oeffentliche, vertrauenswuerdige Quelle (z.B. INB) wird trotz Keyword nicht hart abgelehnt
rules = {"blacklist_keywords": ["Vertraulich"], "regex_redact": []}
doc = make_doc(body="Regulierungstext mit dem Wort Vertraulich darin. " * 3, scope=Scope.ALLGEMEIN)
filter_document(doc, rules, trusted=True)
assert doc.review_status == ReviewStatus.APPROVED
def test_domain_tag_and_path(tmp_path):
gate = ReviewGate(tmp_path)
doc = make_doc(domain="regulierung", scope=Scope.ALLGEMEIN)
apply_tags(doc)
assert "domain:regulierung" in doc.tags
doc.review_status = ReviewStatus.APPROVED
p = gate.write(doc)
assert "regulierung" in str(p) and "allgemein" in str(p)
def test_pdf_heading_split():
from src.connectors.pdf_parser import PdfConnector
text = (
"Vorwort zum Dokument.\n\n"
"Anlage 1.0 Abkuerzungen\nInhalt der Anlage 1.\n\n"
"Anlage 2.0 Entgelte\nInhalt der Anlage 2."
)
segs = PdfConnector()._split_by_headings(text, None)
titles = [t for t, _ in segs]
assert any("Anlage 1.0" in t for t in titles)
assert any("Anlage 2.0" in t for t in titles)
assert len(segs) >= 3 # Einleitung + 2 Anlagen
def test_pdf_strip_noise_removes_footer_and_pagenums():
from src.connectors.pdf_parser import PdfConnector
text = "Echter Inhalt\nINB 2026, Redaktionsstand 11.06.2026 3\n42\nWeiterer Inhalt"
out = PdfConnector._strip_noise(text, [r"^INB \d{4}, Redaktionsstand.*$"])
assert "Redaktionsstand" not in out
assert "\n42\n" not in f"\n{out}\n"
assert "Echter Inhalt" in out and "Weiterer Inhalt" in out
def test_pdf_discover_from_sitemap_picks_newest(monkeypatch):
"""pdf-Strategie findet PDFs in einer (Index-)Sitemap per Regex, filtert Nicht-Treffer
raus und liefert die neueste Version zuerst (hoechste blob-ID)."""
from src.connectors.pdf_parser import PdfConnector
index = 'https://x/sub.xml'
sub = (
""
"https://x/resource/blob/100/aa/Handbuch-pathOS-Webportal-Version-0-6-data.pdf"
"https://x/resource/blob/200/bb/Handbuch-pathOS-Webportal-Version-1-01-data.pdf"
"https://x/resource/blob/300/cc/Anderes-Dokument-data.pdf"
"https://x/web/seite-12345"
""
)
class Resp:
def __init__(self, text):
self.content = text.encode("utf-8")
def fake_get(url, **kw):
return Resp(index if url.endswith("index.xml") else sub)
monkeypatch.setattr("requests.get", fake_get)
c = PdfConnector()
# nur die neueste Handbuch-Version (blob 200 > 100), Nicht-PDF und anderes Dok raus
newest = c._discover_from_sitemap("https://x/index.xml", "Handbuch-pathOS-Webportal", max_pdfs=1)
assert newest == ["https://x/resource/blob/200/bb/Handbuch-pathOS-Webportal-Version-1-01-data.pdf"]
# ohne Limit: beide Handbuch-Treffer (neueste zuerst), 'Anderes-Dokument' rausgefiltert
all_hits = c._discover_from_sitemap("https://x/index.xml", "Handbuch-pathOS-Webportal")
assert len(all_hits) == 2
assert all_hits[0].endswith("Version-1-01-data.pdf")
assert all("Anderes-Dokument" not in u for u in all_hits)
# _discover_pdfs routet eine .xml-URL automatisch in die Sitemap-Discovery
from src.model import Scope, Source, Strategy
src = Source(url="https://x/index.xml", strategy=Strategy.PDF, scopes=[Scope.INTERN],
domain="pathos", options={"url_pattern": "Handbuch-pathOS-Webportal", "max_pdfs": 1})
assert c._discover_pdfs(src) == newest
def test_md_converter_strips_macro_ids():
from src.transformers.md_converter import storage_to_markdown
html = (
'Text vorher
'
''
'Green'
'Text nachher
'
)
md = storage_to_markdown(html)
assert "d1143f0d" not in md
assert "Green" not in md
assert "Text vorher" in md and "Text nachher" in md
def test_sitemap_select_newest_and_filter():
from src.connectors.web_crawler import WebCrawlerConnector
locs = [
"https://x/web/.../kund-inneninformationen/alt-1000000",
"https://x/web/.../kund-inneninformationen/neu-2000000",
"https://x/web/.../impressum", # passt nicht zum pattern
"https://x/web/.../kund-inneninformationen/mitte-1500000",
]
sel = WebCrawlerConnector._select_details(locs, r"kund-inneninformationen/[^/]+-\d{6,}$", 2)
assert len(sel) == 2
assert sel[0].endswith("neu-2000000") # neueste zuerst
assert sel[1].endswith("mitte-1500000")
assert all("impressum" not in u for u in sel)
# limit=None -> alle Treffer (3 Kundeninfos, impressum gefiltert)
alle = WebCrawlerConnector._select_details(locs, r"kund-inneninformationen/[^/]+-\d{6,}$", None)
assert len(alle) == 3
def test_frontmatter_parser():
from src.store import parse_frontmatter
text = (
'---\n'
'domain: "andi"\n'
'scope: "extern"\n'
'tags: ["tool:andi", "faq"]\n'
'review_status: "approved"\n'
'---\n\n'
'# Titel\n\nInhalt hier.\n'
)
meta, body = parse_frontmatter(text)
assert meta["domain"] == "andi"
assert meta["scope"] == "extern"
assert meta["tags"] == ["tool:andi", "faq"]
assert "Inhalt hier." in body
assert not body.startswith("---")
def test_regex_redaction():
rules = {"blacklist_keywords": [], "regex_redact": [r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"]}
doc = make_doc(body="Server unter 192.168.0.1 erreichbar. " * 3)
filter_document(doc, rules)
assert "192.168.0.1" not in doc.body_md
assert "[REDACTED]" in doc.body_md
def test_review_gate_routing(tmp_path):
gate = ReviewGate(tmp_path)
approved = make_doc()
apply_tags(approved)
approved.review_status = ReviewStatus.APPROVED
pending_doc = make_doc()
apply_tags(pending_doc)
pending_doc.review_status = ReviewStatus.PENDING
p1 = gate.write(approved)
p2 = gate.write(pending_doc)
assert "processed" in str(p1)
assert "staging" in str(p2)
gate.write_report()
assert (tmp_path / "staging" / "review_report.json").exists()
def test_tool_level_scope_intern_no_split():
from src.model import Scope, Tool
t = Tool.from_dict({
"id": "crm", "name": "CRM", "domain": "crm", "scope": "intern",
"sources": [{"url": "u", "strategy": "confluence_tree"}],
})
assert t.sources[0].scope == Scope.INTERN
assert not t.sources[0].options.get("scope_split")
def test_tool_level_scope_mixed_defaults_intern():
# 'mixed' splittet NICHT mehr; Quellen ohne eigenen Scope sind intern (restriktiv)
from src.model import Scope, Tool
t = Tool.from_dict({
"id": "andi", "name": "AnDi", "domain": "andi", "scope": "mixed",
"sources": [{"url": "u", "strategy": "confluence_tree"}],
})
assert t.sources[0].scopes == [Scope.INTERN]
assert not t.sources[0].options.get("scope_split")
def test_source_scope_intern_extern_combo():
# 'intern,extern' -> Quelle wird fuer BEIDE Scopes genutzt (ohne Trennung)
from src.model import Scope, Tool
t = Tool.from_dict({
"id": "x", "name": "X", "domain": "x", "scope": "mixed",
"sources": [{"url": "u", "strategy": "confluence_page", "scope": "intern,extern"}],
})
assert t.sources[0].scopes == [Scope.INTERN, Scope.EXTERN]
def test_tool_level_scope_extern_propagates():
from src.model import Scope, Tool
t = Tool.from_dict({
"id": "x", "name": "X", "domain": "x", "scope": "extern",
"sources": [{"url": "u", "strategy": "crawler"}],
})
assert t.sources[0].scope == Scope.EXTERN
def test_approval_allowlist_overrides_pending():
# ein intern-pending-Dokument wird per URL-Allowlist freigegeben
rules = {"blacklist_keywords": ["Vertraulich"], "regex_redact": []}
doc = make_doc(body="Interne Notiz: Vertraulich. " * 5, scope=Scope.INTERN)
approvals = {"urls": {doc.url}, "hashes": set()}
filter_document(doc, rules, approvals=approvals)
assert doc.review_status == ReviewStatus.APPROVED
assert any("freigegeben" in n for n in doc.review_notes)
def test_approval_allowlist_by_hash():
rules = {"blacklist_keywords": ["Vertraulich"], "regex_redact": []}
doc = make_doc(body="Interne Notiz: Vertraulich. " * 5, scope=Scope.INTERN)
approvals = {"urls": set(), "hashes": {doc.content_hash}}
filter_document(doc, rules, approvals=approvals)
assert doc.review_status == ReviewStatus.APPROVED
def test_owners_inherited_and_in_frontmatter():
from src.model import Tool
t = Tool.from_dict({
"id": "x", "name": "X", "domain": "x", "scope": "intern",
"owners": "a@db.de, b@db.de",
"sources": [{"url": "u", "strategy": "confluence_tree"}],
})
assert t.owners == ["a@db.de", "b@db.de"]
assert t.sources[0].owners == ["a@db.de", "b@db.de"] # vererbt
def test_owners_in_document_markdown():
doc = make_doc()
doc.owners = ["a@db.de"]
md = doc.to_markdown()
assert 'owners: ["a@db.de"]' in md
def test_detect_strategy():
from src.strategy_detect import detect_strategy
assert detect_strategy("https://x/INB-2026-Hauptdokument-data.pdf") == "pdf"
assert detect_strategy("https://arija-confluence.jaas.service.deutschebahn.com/spaces/BES/pages/123/X") == "confluence_tree"
assert detect_strategy("https://www.dbinfrago.com/.../sitemap_index.xml") == "sitemap"
assert detect_strategy("https://www.example.com/faq") == "crawler"
# nicht-http -> unbekannt (neue Strategie noetig)
assert detect_strategy("ftp://fileshare/intern.docx") is None
assert detect_strategy("sharepoint://team/site") is None
def test_detect_strategy_gitlab():
from src.strategy_detect import detect_strategy
assert detect_strategy("https://git.tech.rz.db.de/gruppe/projekt") == "gitlab_md"
def test_parse_gitlab_url():
from src.connectors.gitlab_md import parse_gitlab_url
i = parse_gitlab_url("https://git.tech.rz.db.de/einfachbahn-lab/tools/x/-/tree/main/docs")
assert i["host"] == "git.tech.rz.db.de"
assert i["project"] == "einfachbahn-lab/tools/x"
assert i["ref"] == "main"
assert i["path"] == "docs"
i2 = parse_gitlab_url("https://git.tech.rz.db.de/gruppe/projekt")
assert i2["project"] == "gruppe/projekt" and i2["ref"] == "" and i2["path"] == ""
def test_file_strategy_reads_markdown(tmp_path):
from src.connectors.file_source import FileConnector
from src.model import Scope, Source, Strategy
d = tmp_path / "files" / "mein-tool"
d.mkdir(parents=True)
(d / "handbuch.md").write_text("# Handbuch\n\nInhalt des Handbuchs.", encoding="utf-8")
src = Source(url=str(d), strategy=Strategy.FILE, scopes=[Scope.INTERN], domain="mein-tool")
docs = FileConnector().fetch(src, None)
assert len(docs) == 1
assert docs[0].scope == Scope.INTERN
assert "Handbuch" in docs[0].body_md
def test_detect_strategy_file_path():
from src.strategy_detect import detect_strategy
assert detect_strategy("files/mein-tool/handbuch.pdf") == "file"
assert detect_strategy("files/mein-tool") == "file"
def test_quality_score():
from src.quality import quality_score
assert quality_score("") == 0
long = "# Titel\n\n" + ("Ein gut strukturierter Absatz mit Inhalt. " * 40) + "\n- Punkt eins\n- Punkt zwei"
assert quality_score(long) >= 70
assert quality_score("zu kurz") < 30
def test_confluence_known_versions(tmp_path):
from src.connectors.confluence import ConfluenceConnector
d = tmp_path / "processed" / "intern" / "andi"
d.mkdir(parents=True)
(d / "seite.md").write_text(
'---\n'
'domain: "andi"\n'
'scope: "intern"\n'
'source_version: "7"\n'
'url: "https://x/pages/12345/Titel"\n'
'---\n\n# Titel\n\nInhalt.\n',
encoding="utf-8",
)
# andere Domaene -> darf nicht auftauchen
d2 = tmp_path / "processed" / "intern" / "other"
d2.mkdir(parents=True)
(d2 / "x.md").write_text(
'---\ndomain: "other"\nsource_version: "3"\nurl: "https://x/pages/999/Y"\n---\n\nX\n',
encoding="utf-8",
)
conn = ConfluenceConnector(data_dir=str(tmp_path))
known = conn._known_versions("andi")
assert set(known.keys()) == {"12345"}
assert known["12345"]["version"] == "7"
assert known["12345"]["files"]
assert known["12345"]["files"][0]["scope"] == "intern"
def test_issue_parser_strategy_per_link():
import sys as _sys
from pathlib import Path as _Path
_sys.path.insert(0, str(_Path(__file__).resolve().parents[1] / "scripts"))
from issue_to_source import build_entry, parse_issue
issue = (
"### Tool / Domaene\nMaTeo\n\n"
"### Verantwortlich (owners, 1-2)\nresp@db.de\n\n"
"### Kontakt (Ansprechpartner, optional)\nteam@db.de\n\n"
"### Quellen (eine pro Zeile: URL | Strategie | Scope)\n"
"- https://arija-confluence.x/spaces/BES/pages/123/X | confluence_tree | intern\n"
"- https://www.example.com/faq | crawler | extern\n"
"- https://www.example.com/handbuch-data.pdf\n"
)
data = parse_issue(issue)
assert data["tool"] == "MaTeo"
assert data["owners"] == ["resp@db.de"]
assert data["contact"] == "team@db.de"
assert len(data["sources"]) == 3
assert data["sources"][0]["strategy"] == "confluence_tree"
assert data["sources"][1]["scope"] == "extern"
# dritte Zeile ohne Strategie -> Auto-Erkennung
assert data["sources"][2]["strategy"] == ""
entry = build_entry(data)
assert "strategy: confluence_tree" in entry
assert 'scope: "extern"' in entry
assert "strategy: pdf" in entry # automatisch erkannt
assert 'contact: "team@db.de"' in entry
def test_contact_default_and_separate_from_owners():
from src.model import DEFAULT_CONTACT, Tool
# kein contact angegeben -> Default; owners bleibt davon getrennt
t = Tool.from_dict({
"id": "x", "name": "X", "domain": "x", "scope": "intern",
"owners": "verantwortlich@db.de",
"sources": [{"url": "u", "strategy": "confluence_tree"}],
})
s = t.sources[0]
assert s.owners == ["verantwortlich@db.de"]
assert s.contact_address == DEFAULT_CONTACT # einfachbahn@deutschebahn.com
# expliziter Tool-contact wird vererbt
t2 = Tool.from_dict({
"id": "y", "name": "Y", "domain": "y", "scope": "intern",
"owners": "resp@db.de", "contact": "team-y@db.de",
"sources": [{"url": "u", "strategy": "confluence_tree"}],
})
assert t2.sources[0].contact_address == "team-y@db.de"
assert t2.sources[0].owners == ["resp@db.de"]
def test_contact_in_document_markdown():
doc = make_doc()
doc.owners = ["resp@db.de"]
doc.contact = "einfachbahn@deutschebahn.com"
md = doc.to_markdown()
assert 'owners: ["resp@db.de"]' in md
assert 'contact: "einfachbahn@deutschebahn.com"' in md
def test_site_build_creates_pages(tmp_path):
"""build() erzeugt alle Pages + JSON aus einem Mini-Datenbestand."""
from src.review.staging import ReviewGate
from src.site import build
from src.store import write_run_meta
# einen approved-Datensatz nach output/processed schreiben
gate = ReviewGate(tmp_path)
doc = make_doc(scope=Scope.EXTERN, domain="pathos")
apply_tags(doc)
doc.review_status = ReviewStatus.APPROVED
gate.write(doc)
write_run_meta(str(tmp_path), gate.summary())
out = tmp_path / "public"
build(str(tmp_path), str(out), staging_dir=str(tmp_path / "staging"))
for name in ("index.html", "hilfe.html", "chatbot.html", "config.html",
"changelog.html", "data.json", "config.json"):
assert (out / name).exists(), f"{name} fehlt"
import json
payload = json.loads((out / "data.json").read_text(encoding="utf-8"))
assert payload["docs"] and payload["docs"][0]["scope"] == "extern"
# last_run/last_change in data.json (fuer die dezente Fusszeile)
assert payload["meta"]["last_run"]
assert payload["meta"]["last_change"]
# Footer-Platzhalter ist vorhanden
assert 'id="lastrun"' in (out / "index.html").read_text(encoding="utf-8")
# Nav-Platzhalter muss ersetzt sein
assert "__NAV__" not in (out / "index.html").read_text(encoding="utf-8")
# Changelog gerendert, Platzhalter ersetzt
cl = (out / "changelog.html").read_text(encoding="utf-8")
assert "__CHANGELOG__" not in cl and "__VERSION__" not in cl
assert "Changelog" in cl
def test_confluence_url_uses_webui_links():
"""URL kommt aus _links.webui (nicht das nicht-aufloesbare /pages/)."""
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_page_by_id(self, pid, expand=None):
return {
"title": "Titel",
"body": {"storage": {"value": "Ein ausreichend langer Inhalt.
"}},
"version": {"number": 3},
"_links": {"base": "https://x", "webui": "/spaces/BES/pages/329715401/Titel"},
}
src = Source(url="https://x/spaces/BES/pages/329715401/Titel",
strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="d")
docs = ConfluenceConnector()._page_to_doc(FakeClient(), "329715401", src, None)
assert docs and docs[0].url == "https://x/spaces/BES/pages/329715401/Titel"
def test_confluence_url_fallback_viewpage():
"""Ohne _links -> Fallback auf viewpage.action?pageId=."""
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_page_by_id(self, pid, expand=None):
return {"title": "T", "body": {"storage": {"value": "x"}}, "version": {"number": 1}}
src = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="d")
docs = ConfluenceConnector()._page_to_doc(FakeClient(), "329715401", src, None)
assert docs and docs[0].url.endswith("/pages/viewpage.action?pageId=329715401")
def test_meta_fingerprint_changes_with_tags_owners_contact():
from src.model import Scope, Source, Strategy, source_meta_fingerprint
base = Source(url="u", strategy=Strategy.CONFLUENCE_TREE, scopes=[Scope.INTERN],
domain="d", tags=["a", "b"], owners=["x@db.de"], contact="t@db.de")
fp0 = source_meta_fingerprint(base)
# gleiche Werte, andere Reihenfolge/Case -> gleicher Fingerprint (stabil)
same = Source(url="u", strategy=Strategy.CONFLUENCE_TREE, scopes=[Scope.INTERN],
domain="d", tags=["B", "A"], owners=["x@db.de"], contact="t@db.de")
assert source_meta_fingerprint(same) == fp0
# Tag geaendert -> anderer Fingerprint
changed = Source(url="u", strategy=Strategy.CONFLUENCE_TREE, scopes=[Scope.INTERN],
domain="d", tags=["a", "c"], owners=["x@db.de"], contact="t@db.de")
assert source_meta_fingerprint(changed) != fp0
# scope/domain aendern -> Fingerprint UNVERAENDERT (steuern den Pfad, separat)
other_scope = Source(url="u", strategy=Strategy.CONFLUENCE_TREE, scopes=[Scope.EXTERN],
domain="other", tags=["a", "b"], owners=["x@db.de"], contact="t@db.de")
assert source_meta_fingerprint(other_scope) == fp0
def test_meta_fingerprint_in_frontmatter():
doc = make_doc()
apply_tags(doc)
doc.meta_fingerprint = "abc123"
md = doc.to_markdown()
assert 'meta_fingerprint: "abc123"' in md
def test_patch_frontmatter_updates_only_targeted_fields(tmp_path):
from src.store import parse_frontmatter, patch_frontmatter_file
p = tmp_path / "seite.md"
p.write_text(
'---\n'
'domain: "andi"\n'
'tool: "andi"\n'
'scope: "intern"\n'
'tags: ["domain:andi", "alt"]\n'
'owners: ["alt@db.de"]\n'
'contact: "alt@db.de"\n'
'meta_fingerprint: ""\n'
'url: "https://x/pages/1/Y"\n'
'---\n\n# Titel\n\nInhalt bleibt.\n',
encoding="utf-8",
)
ok = patch_frontmatter_file(p, tags=["domain:andi", "neu"], owners=["neu@db.de"],
contact="neu@db.de", meta_fingerprint="ff00")
assert ok
meta, body = parse_frontmatter(p.read_text(encoding="utf-8"))
assert meta["tags"] == ["domain:andi", "neu"]
assert meta["owners"] == ["neu@db.de"]
assert meta["contact"] == "neu@db.de"
assert meta["meta_fingerprint"] == "ff00"
# nicht angefasste Felder + Body bleiben erhalten
assert meta["scope"] == "intern"
assert meta["url"] == "https://x/pages/1/Y"
assert "Inhalt bleibt." in body
def test_reconcile_meta_rewrites_tags_and_fingerprint(tmp_path):
from src.model import Scope, Source, Strategy, source_meta_fingerprint
from src.store import parse_frontmatter, reconcile_meta
p = tmp_path / "seite.md"
p.write_text(
'---\n'
'domain: "pathos"\n'
'tool: "pathos"\n'
'scope: "extern"\n'
'tags: ["domain:pathos", "tool:pathos", "scope:extern", "alt"]\n'
'owners: []\n'
'contact: "einfachbahn@deutschebahn.com"\n'
'meta_fingerprint: "oldfp"\n'
'url: "https://x/p/1"\n'
'---\n\nInhalt.\n',
encoding="utf-8",
)
src = Source(url="u", strategy=Strategy.SITEMAP, scopes=[Scope.EXTERN],
domain="pathos", tags=["neu", "web"], owners=["resp@db.de"], contact="team@db.de")
fp = source_meta_fingerprint(src)
n = reconcile_meta([str(p)], src, fp)
assert n == 1
meta, _ = parse_frontmatter(p.read_text(encoding="utf-8"))
# augmentierte Tags mit der im File hinterlegten domain/tool/scope rekonstruiert
assert meta["tags"] == ["domain:pathos", "tool:pathos", "scope:extern", "neu", "web"]
assert meta["owners"] == ["resp@db.de"]
assert meta["contact"] == "team@db.de"
assert meta["meta_fingerprint"] == fp
def test_prune_scope_orphans_removes_old_placement(tmp_path):
"""Scope-Wechsel: alte intern-Datei wird entfernt, neue extern-Datei bleibt,
eine andere URL derselben Domaene bleibt unangetastet."""
from src.store import prune_scope_orphans
proc = tmp_path / "processed"
# alte Ablage (intern) der gewechselten Seite
old = proc / "intern" / "pathos"
old.mkdir(parents=True)
(old / "seite.md").write_text(
'---\ndomain: "pathos"\nscope: "intern"\nurl: "https://x/p/1"\n---\n\nInhalt.\n',
encoding="utf-8",
)
# neue Ablage (extern) derselben Seite (in diesem Lauf geschrieben)
new = proc / "extern" / "pathos"
new.mkdir(parents=True)
(new / "seite.md").write_text(
'---\ndomain: "pathos"\nscope: "extern"\nurl: "https://x/p/1"\n---\n\nInhalt.\n',
encoding="utf-8",
)
# unbeteiligte zweite Seite (intern) -> muss bleiben
(old / "andere.md").write_text(
'---\ndomain: "pathos"\nscope: "intern"\nurl: "https://x/p/2"\n---\n\nInhalt.\n',
encoding="utf-8",
)
produced = {("pathos", "https://x/p/1"): {"extern"}}
deleted = prune_scope_orphans(str(tmp_path), produced)
assert len(deleted) == 1
assert not (old / "seite.md").exists() # alte intern-Ablage weg
assert (new / "seite.md").exists() # neue extern-Ablage bleibt
assert (old / "andere.md").exists() # andere URL unberuehrt
def test_prune_scope_orphans_keeps_combo_scopes(tmp_path):
"""'intern,extern': beide Scopes erzeugt -> nichts wird geloescht."""
from src.store import prune_scope_orphans
proc = tmp_path / "processed"
for sc in ("intern", "extern"):
d = proc / sc / "pathos"
d.mkdir(parents=True)
(d / "seite.md").write_text(
f'---\ndomain: "pathos"\nscope: "{sc}"\nurl: "https://x/p/1"\n---\n\nInhalt.\n',
encoding="utf-8",
)
produced = {("pathos", "https://x/p/1"): {"intern", "extern"}}
deleted = prune_scope_orphans(str(tmp_path), produced)
assert deleted == []
assert (proc / "intern" / "pathos" / "seite.md").exists()
assert (proc / "extern" / "pathos" / "seite.md").exists()
def test_write_run_meta_tracks_last_run_and_change(tmp_path):
"""last_run aktualisiert sich jeden Lauf; last_change bleibt stehen, solange
sich der Bestand nicht aendert, und springt bei einer Aenderung mit."""
import json
import time
from src.review.staging import ReviewGate
from src.store import write_run_meta
gate = ReviewGate(tmp_path)
doc = make_doc(scope=Scope.EXTERN, domain="pathos")
apply_tags(doc)
doc.meta_fingerprint = "fp1"
doc.review_status = ReviewStatus.APPROVED
gate.write(doc)
p1 = write_run_meta(str(tmp_path), {"approved": 1})
m1 = json.loads(p1.read_text(encoding="utf-8"))
assert m1["documents"] == 1
assert m1["by_scope"]["extern"] == 1
assert m1["last_run"] == m1["last_change"]
assert m1["last_run_summary"] == {"approved": 1}
# zweiter Lauf ohne Aenderung -> last_change bleibt, last_run kann gleich/neuer sein
time.sleep(1)
m2 = json.loads(write_run_meta(str(tmp_path)).read_text(encoding="utf-8"))
assert m2["last_change"] == m1["last_change"]
assert m2["content_signature"] == m1["content_signature"]
# Inhalt aendert sich -> last_change springt mit (eigene URL: anderes Dokument,
# nicht durch Gate-Dedup mit dem ersten zusammengefasst)
doc2 = make_doc(body="Komplett neuer Inhalt fuer den Test. " * 4, scope=Scope.EXTERN, domain="pathos")
doc2.url = "https://example.com/y"
apply_tags(doc2)
doc2.review_status = ReviewStatus.APPROVED
gate.write(doc2)
m3 = json.loads(write_run_meta(str(tmp_path)).read_text(encoding="utf-8"))
assert m3["content_signature"] != m1["content_signature"]
assert m3["last_change"] != m1["last_change"]
def test_processed_signature_reacts_to_metadata(tmp_path):
"""Reine Metadaten-Aenderung (meta_fingerprint) aendert die Signatur."""
from src.store import patch_frontmatter_file, processed_signature
d = tmp_path / "processed" / "extern" / "pathos"
d.mkdir(parents=True)
f = d / "seite.md"
f.write_text(
'---\ndomain: "pathos"\nscope: "extern"\n'
'content_hash: "deadbeef"\nmeta_fingerprint: "fp1"\n'
'url: "https://x/p/1"\n---\n\nInhalt.\n',
encoding="utf-8",
)
sig_before, n, _ = processed_signature(str(tmp_path))
assert n == 1
patch_frontmatter_file(f, meta_fingerprint="fp2")
sig_after, _, _ = processed_signature(str(tmp_path))
assert sig_after != sig_before
# ---------------------------------------------------------------------------
# Chunking (deterministisch, ohne Embeddings)
# ---------------------------------------------------------------------------
def test_chunk_effective_opts_override():
from src.transformers.chunker import effective_opts
# Default off; chunking.yaml-Defaults < per-Quelle
opts = effective_opts({"chunk": "headings", "chunk_target_tokens": 700},
{"chunk_min_chars": 50, "chunk_target_tokens": 500})
assert opts["chunk"] == "headings"
assert opts["chunk_target_tokens"] == 700 # per-Quelle schlaegt Default
assert opts["chunk_min_chars"] == 50 # aus chunking.yaml-Default
# ohne Override bleibt es off
assert effective_opts({}, {})["chunk"] == "off"
def test_split_sections_and_ziffer():
from src.transformers.chunker import split_sections
md = (
"# Titel\n\nEinleitung.\n\n"
"## 7.3.1.1 Kapazitaet\n\nText A der ausreichend lang ist.\n\n"
"## 7.3.1.2 Entgelte\n\nText B der ausreichend lang ist.\n"
)
secs = split_sections(md, level=2)
titles = [s["title"] for s in secs]
assert any("Kapazitaet" in t for t in titles)
ziffern = [s["ziffer"] for s in secs if s["ziffer"]]
assert "7.3.1.1" in ziffern and "7.3.1.2" in ziffern
def test_merge_small_sections():
from src.transformers.chunker import _merge_small
secs = [
{"title": "A", "ziffer": "", "text": "X" * 300},
{"title": "B", "ziffer": "", "text": "kurz"},
]
merged = _merge_small(secs, min_chars=200)
assert len(merged) == 1
assert "kurz" in merged[0]["text"]
def test_chunk_document_produces_chunks_with_context_and_parent():
from src.transformers.chunker import chunk_document
body = (
"## 1.1 Erster Abschnitt\n\n" + ("Inhalt eins. " * 30) + "\n\n"
"## 1.2 Zweiter Abschnitt\n\n" + ("Inhalt zwei. " * 30) + "\n"
)
doc = make_doc(body=body, scope=Scope.ALLGEMEIN, domain="regulierung")
doc.title = "INB 2026"
opts = {"chunk": "headings", "chunk_level": 2, "chunk_context_header": True,
"chunk_min_chars": 50, "chunk_target_tokens": 500, "chunk_max_tokens": 1800}
chunks = chunk_document(doc, opts)
assert len(chunks) == 2
assert all(c.kind == "chunk" for c in chunks)
assert all(c.parent_hash == doc.content_hash for c in chunks)
assert all(c.parent_url == doc.url for c in chunks)
# Contextual-Retrieval-Vorspann je Chunk
assert chunks[0].body_md.startswith("> Kontext: INB 2026 >")
# Frontmatter markiert kind=chunk
md = chunks[0].to_markdown()
assert 'kind: "chunk"' in md and 'ziffer: "1.1"' in md
def test_chunk_document_off_returns_empty():
from src.transformers.chunker import chunk_document
doc = make_doc()
assert chunk_document(doc, {"chunk": "off"}) == []
assert chunk_document(doc, {}) == []
def test_collect_locations_only_enabled():
from src.chunk import collect_locations
from src.model import Source, Tool
tools = [Tool.from_dict({
"id": "x", "name": "X", "domain": "x", "scope": "intern",
"sources": [{"url": "u", "strategy": "confluence_tree"}], # kein chunk -> off
})]
general = [Source.from_dict({
"url": "u2", "strategy": "pdf", "scope": "allgemein", "domain": "regulierung",
"options": {"chunk": "headings"},
})]
locs = collect_locations(tools, general, {})
assert ("allgemein", "regulierung", "") in locs
assert all(loc[1] != "x" for loc in locs) # x ist off -> nicht enthalten
def test_chunk_location_writes_files(tmp_path):
from src.chunk import chunk_location
proc = tmp_path / "processed" / "allgemein" / "regulierung"
proc.mkdir(parents=True)
doc = make_doc(
body="## 1.1 A\n\n" + ("Inhalt. " * 40) + "\n\n## 1.2 B\n\n" + ("Mehr. " * 40),
scope=Scope.ALLGEMEIN, domain="regulierung",
)
doc.title = "INB"
doc.review_status = ReviewStatus.APPROVED
(proc / "inb.md").write_text(doc.to_markdown(), encoding="utf-8")
opts = {"chunk": "headings", "chunk_level": 2, "chunk_min_chars": 50,
"chunk_target_tokens": 500, "chunk_max_tokens": 1800, "chunk_context_header": True}
n_docs, n_chunks, skipped = chunk_location(tmp_path, ("allgemein", "regulierung", ""), opts)
assert n_docs == 1 and n_chunks == 2 and skipped == 0
written = list((tmp_path / "chunks" / "allgemein" / "regulierung" / "inb").glob("*.md"))
assert len(written) == 2
# Chunk-Datei traegt kind=chunk + parent-Verknuepfung + chunk_fingerprint
txt = sorted(written)[0].read_text(encoding="utf-8")
assert 'kind: "chunk"' in txt and "parent_hash:" in txt and "chunk_fingerprint:" in txt
def _write_full_doc(proc: Path, name: str, body: str, title: str = "INB",
domain: str = "regulierung", scope: Scope = Scope.ALLGEMEIN):
proc.mkdir(parents=True, exist_ok=True)
doc = Document(tool="allgemein", scope=scope, domain=domain,
title=title, body_md=body, url=f"https://x/{name}",
component_type=ComponentType.PDF, review_status=ReviewStatus.APPROVED)
(proc / f"{name}.md").write_text(doc.to_markdown(), encoding="utf-8")
def test_chunk_location_incremental_skips_unchanged(tmp_path):
from src.chunk import chunk_location
proc = tmp_path / "processed" / "allgemein" / "regulierung"
body = "## 1.1 A\n\n" + ("Inhalt. " * 40) + "\n\n## 1.2 B\n\n" + ("Mehr. " * 40)
_write_full_doc(proc, "inb", body)
opts = {"chunk": "headings", "chunk_level": 2, "chunk_min_chars": 50}
d1, c1, s1 = chunk_location(tmp_path, ("allgemein", "regulierung", ""), opts)
assert d1 == 1 and s1 == 0
# zweiter Lauf, nichts geaendert -> alles uebersprungen, keine neuen Schreibvorgaenge
files = sorted((tmp_path / "chunks" / "allgemein" / "regulierung" / "inb").glob("*.md"))
before = {f.name: f.read_text(encoding="utf-8") for f in files}
d2, c2, s2 = chunk_location(tmp_path, ("allgemein", "regulierung", ""), opts)
assert d2 == 0 and c2 == 0 and s2 == 1
after = {f.name: f.read_text(encoding="utf-8") for f in files}
assert before == after # byte-identisch -> kein Git-Churn
def test_chunk_location_rechunks_on_option_change(tmp_path):
from src.chunk import chunk_location
proc = tmp_path / "processed" / "allgemein" / "regulierung"
body = "## 1.1 A\n\n" + ("Inhalt. " * 40) + "\n\n## 1.2 B\n\n" + ("Mehr. " * 40)
_write_full_doc(proc, "inb", body)
chunk_location(tmp_path, ("allgemein", "regulierung", ""), {"chunk": "headings", "chunk_min_chars": 50})
# Strategie/Option geaendert -> Fingerprint weicht ab -> neu chunken
d, c, s = chunk_location(tmp_path, ("allgemein", "regulierung", ""),
{"chunk": "recursive", "chunk_min_chars": 50})
assert d == 1 and s == 0
def test_chunk_modes_faq_and_recursive():
from src.transformers.chunker import chunk_document
faq_body = "### Frage 1\n\nAntwort eins.\n\n### Frage 2\n\nAntwort zwei.\n"
doc = make_doc(body=faq_body, scope=Scope.ALLGEMEIN, domain="x")
faq = chunk_document(doc, {"chunk": "faq", "chunk_min_chars": 1, "chunk_context_header": False})
assert len(faq) == 2
rec_body = "## H\n\n" + ("Satz. " * 200)
doc2 = make_doc(body=rec_body, scope=Scope.ALLGEMEIN, domain="x")
rec = chunk_document(doc2, {"chunk": "recursive", "chunk_target_tokens": 50,
"chunk_max_tokens": 60, "chunk_context_header": False})
# recursive ignoriert Ueberschriften, teilt rein groessenbasiert -> mehrere Teile
assert len(rec) >= 2
def test_chunk_options_fingerprint_stable_and_sensitive():
from src.transformers.chunker import chunk_options_fingerprint
a = chunk_options_fingerprint({"chunk": "headings", "chunk_level": 2})
b = chunk_options_fingerprint({"chunk": "headings", "chunk_level": 2})
assert a == b # stabil
assert a != chunk_options_fingerprint({"chunk": "headings", "chunk_level": 3})
assert a != chunk_options_fingerprint({"chunk": "recursive"})
# off/False sind gleichbedeutend -> gleicher Hash
assert chunk_options_fingerprint({"chunk": "off"}) == chunk_options_fingerprint({"chunk": False})
def test_prune_orphans_removes_deactivated_and_missing_parent(tmp_path):
from src.chunk import chunk_location, prune_orphans
proc = tmp_path / "processed" / "allgemein" / "regulierung"
body = "## 1.1 A\n\n" + ("Inhalt. " * 40) + "\n\n## 1.2 B\n\n" + ("Mehr. " * 40)
_write_full_doc(proc, "inb-a", body)
_write_full_doc(proc, "inb-b", body)
loc = ("allgemein", "regulierung", "")
opts = {"chunk": "headings", "chunk_min_chars": 50}
chunk_location(tmp_path, loc, opts)
assert (tmp_path / "chunks" / "allgemein" / "regulierung" / "inb-a").exists()
# Voll-Dokument inb-b geloescht -> dessen Chunks gelten als verwaist
(proc / "inb-b.md").unlink()
removed = prune_orphans(tmp_path, {loc: opts})
assert any("inb-b" in r for r in removed)
assert not (tmp_path / "chunks" / "allgemein" / "regulierung" / "inb-b").exists()
assert (tmp_path / "chunks" / "allgemein" / "regulierung" / "inb-a").exists()
# Ort komplett deaktiviert (nicht mehr in locations) -> alles weg
removed2 = prune_orphans(tmp_path, {})
assert any("inb-a" in r for r in removed2)
assert not (tmp_path / "chunks" / "allgemein" / "regulierung" / "inb-a").exists()
def test_split_large_hard_bounds_single_block():
"""Ein ueberlanger Block OHNE Absatztrenner wird hart auf das Budget begrenzt."""
from src.transformers.chunker import _split_large, _tokens
one_line = "X" * 20000 # ein einziger Block, keine \n\n
parts = _split_large(one_line, target_tokens=500, max_tokens=1800)
assert len(parts) > 1
assert all(_tokens(p) <= 1800 for p in parts)
assert "".join(parts) == one_line # verlustfrei
def test_build_index_links_full_doc_and_chunks(tmp_path):
"""Der Index listet pro Voll-Dokument Herkunft + (falls vorhanden) Chunks."""
from src.chunk import chunk_location
from src.store import build_index, write_index
proc = tmp_path / "processed" / "allgemein" / "regulierung"
body = "## 1.1 A\n\n" + ("Inhalt. " * 40) + "\n\n## 1.2 B\n\n" + ("Mehr. " * 40)
_write_full_doc(proc, "inb", body)
# ein zweites Dokument OHNE Chunks (anderer Bereich)
other = tmp_path / "processed" / "extern" / "pathos"
_write_full_doc(other, "seite", "Kurzer Inhalt.", title="PathOS",
domain="pathos", scope=Scope.EXTERN)
chunk_location(tmp_path, ("allgemein", "regulierung", ""), {"chunk": "headings", "chunk_min_chars": 50})
idx = build_index(tmp_path)
assert idx["documents_total"] == 2
assert idx["chunks_total"] >= 2
by_url = {d["title"]: d for d in idx["documents"]}
inb = next(d for d in idx["documents"] if d["domain"] == "regulierung")
pathos = next(d for d in idx["documents"] if d["domain"] == "pathos")
assert inb["chunks"] and inb["chunks"]["count"] >= 2
assert inb["chunks"]["path"].startswith("chunks/allgemein/regulierung/inb")
assert pathos["chunks"] is None # kein Chunking -> kein Eintrag
assert idx["by_domain"]["regulierung"]["chunks"] >= 2
assert "allgemein" in idx["by_domain"]["regulierung"]["scopes"]
assert by_url # smoke
# write_index schreibt nach output/_index.json und ist deterministisch
p = write_index(tmp_path)
assert p.name == "_index.json"
first = p.read_text(encoding="utf-8")
second_payload = write_index(tmp_path).read_text(encoding="utf-8")
assert first == second_payload # kein Churn bei unveraendertem Bestand
def test_chunk_parent_hash_matches_full_doc_content_hash(tmp_path):
"""Chunk.parent_hash entspricht exakt dem content_hash der Voll-Datei (Linking)."""
from src.chunk import chunk_location
from src.store import parse_frontmatter
proc = tmp_path / "processed" / "allgemein" / "regulierung"
body = "## 1.1 A\n\n" + ("Inhalt. " * 40) + "\n\n## 1.2 B\n\n" + ("Mehr. " * 40)
_write_full_doc(proc, "inb", body)
full_meta, _ = parse_frontmatter((proc / "inb.md").read_text(encoding="utf-8"))
full_hash = full_meta["content_hash"]
chunk_location(tmp_path, ("allgemein", "regulierung", ""), {"chunk": "headings", "chunk_min_chars": 50})
chunk_files = sorted((tmp_path / "chunks" / "allgemein" / "regulierung" / "inb").glob("*.md"))
for cf in chunk_files:
cmeta, _ = parse_frontmatter(cf.read_text(encoding="utf-8"))
assert cmeta["parent_hash"] == full_hash # exakt verknuepft
assert cf.name.endswith(".md") # Dateiname behaelt Endung
def test_filter_invalid_redaction_pattern_does_not_crash():
"""Ein ungueltiges Regex in regex_redact bricht die Verarbeitung nicht ab."""
rules = {"blacklist_keywords": [], "regex_redact": ["(ungeschlossen", r"\d+"]}
doc = make_doc(body="Wert 12345 im Text. " * 4)
filter_document(doc, rules) # darf nicht werfen
assert "[REDACTED]" in doc.body_md # gueltiges Pattern greift weiter
assert any("ungueltiges Redaction-Pattern" in n for n in doc.review_notes)
def test_is_off_does_not_match_int_zero():
from src.transformers.chunker import _is_off
assert _is_off("off") and _is_off(False) and _is_off(None) and _is_off("")
assert _is_off("OFF") and _is_off("no") and _is_off("false")
assert not _is_off(0) # int 0 ist KEIN 'off'
assert not _is_off("headings") and not _is_off("recursive")
def test_chunk_overlap_prepends_previous_tail():
from src.transformers.chunker import chunk_document
body = "## A\n\n" + ("Alpha. " * 30) + "\n\n## B\n\n" + ("Beta. " * 30)
doc = make_doc(body=body, scope=Scope.ALLGEMEIN, domain="x")
doc.title = "T"
base = {"chunk": "headings", "chunk_level": 2, "chunk_min_chars": 1,
"chunk_context_header": False}
no_ov = chunk_document(doc, {**base, "chunk_overlap": 0})
ov = chunk_document(doc, {**base, "chunk_overlap": 8})
assert len(no_ov) == len(ov) == 2
# zweiter Chunk enthaelt mit Overlap einen Tail des ersten (Alpha), ohne Overlap nicht
assert "Alpha" in ov[1].body_md
assert "Alpha" not in no_ov[1].body_md
# Overlap aendert den Fingerprint -> inkrementelles Re-Chunking greift
assert ov[0].chunk_fingerprint != no_ov[0].chunk_fingerprint
def test_append_run_log_appends_and_caps(tmp_path):
import json as _json
from src.store import append_run_log, read_last_run
for i in range(5):
append_run_log(tmp_path, {"documents": i, "sources_failed": 0, "errors": []}, cap=3)
path = tmp_path / "run_log.jsonl"
lines = [ln for ln in path.read_text(encoding="utf-8").splitlines() if ln.strip()]
assert len(lines) == 3 # auf cap gekappt
docs = [_json.loads(ln)["documents"] for ln in lines]
assert docs == [2, 3, 4] # die letzten 3
assert all("ts" in _json.loads(ln) for ln in lines) # Zeitstempel gesetzt
# read_last_run liefert den juengsten Eintrag
assert read_last_run(tmp_path)["documents"] == 4
def test_read_last_run_empty(tmp_path):
from src.store import read_last_run
assert read_last_run(tmp_path) == {}
def test_run_log_captures_source_errors(tmp_path):
"""End-to-end (offline): ein Connector-Fehler landet als Quelle-Fehler im Log."""
import json as _json
from src.store import append_run_log
entry = {
"documents": 3,
"by_status": {"approved": 3},
"sources_total": 2,
"sources_failed": 1,
"doc_errors": 0,
"errors": [{"source": "pathos/pathos", "strategy": "confluence_tree",
"error": "No scheme supplied"}],
}
append_run_log(tmp_path, entry)
rec = _json.loads((tmp_path / "run_log.jsonl").read_text(encoding="utf-8").strip())
assert rec["sources_failed"] == 1
assert rec["errors"][0]["source"] == "pathos/pathos"
assert "scheme" in rec["errors"][0]["error"]
# ---------------------------------------------------------------------------
# Dedup: gleiche Seite je scope/domaene nur einmal im Feed
# ---------------------------------------------------------------------------
def test_page_identity_normalizes_confluence():
from src.model import page_identity
a = page_identity("https://x/spaces/BES/pages/611945127/FAQ+PathOS+Extern")
b = page_identity("https://x/pages/viewpage.action?pageId=611945127")
assert a == b == "confluence:611945127"
# nicht-Confluence -> normalisierte URL
assert page_identity("https://www.dbinfrago.com/web/x/") == "https://www.dbinfrago.com/web/x"
assert page_identity("") == ""
def _approved(title, body="Ein ausreichend langer Inhalt fuer den Test. " * 3,
url="https://x/spaces/BES/pages/611945127/FAQ", comp=ComponentType.PAGE,
scope=Scope.EXTERN, domain="pathos"):
d = Document(tool="pathos", scope=scope, domain=domain, title=title, body_md=body,
url=url, component_type=comp, review_status=ReviewStatus.APPROVED)
return d
def test_gate_dedup_same_page_one_file(tmp_path):
gate = ReviewGate(tmp_path)
# gleiche Seite (gleiche page-id), zwei URL-Schreibweisen, beide PAGE
p1 = gate.write(_approved("FAQ", url="https://x/spaces/BES/pages/611945127/FAQ"))
p2 = gate.write(_approved("FAQ", url="https://x/pages/viewpage.action?pageId=611945127"))
assert p1 == p2 # nur eine Datei
files = list((tmp_path / "processed" / "extern" / "pathos").glob("*.md"))
assert len(files) == 1
assert gate.summary().get("approved") == 1 # kein doppelter Report-Eintrag
def test_gate_dedup_faq_beats_page(tmp_path):
gate = ReviewGate(tmp_path)
# zuerst die generische Seite (Tree), dann die spezifische FAQ -> FAQ gewinnt
gate.write(_approved("FAQ", body="## Tabelle\n\n| a | b |\n|---|---|", comp=ComponentType.PAGE))
gate.write(_approved("FAQ", body="### Frage\n\nAntwort ausfuehrlich genug fuer den Test.",
comp=ComponentType.FAQ))
files = list((tmp_path / "processed" / "extern" / "pathos").glob("*.md"))
assert len(files) == 1
text = files[0].read_text(encoding="utf-8")
assert "### Frage" in text # FAQ-Inhalt gewann
# umgekehrte Reihenfolge (FAQ zuerst) -> FAQ bleibt, PAGE wird verworfen
gate2 = ReviewGate(tmp_path / "b")
gate2.write(_approved("FAQ", body="### Frage\n\nAntwort ausfuehrlich genug fuer den Test.",
comp=ComponentType.FAQ))
gate2.write(_approved("FAQ", body="## Tabelle\n\nrohe Tabelle", comp=ComponentType.PAGE))
f2 = list((tmp_path / "b" / "processed" / "extern" / "pathos").glob("*.md"))
assert len(f2) == 1 and "### Frage" in f2[0].read_text(encoding="utf-8")
def test_gate_keeps_different_pages_same_title(tmp_path):
gate = ReviewGate(tmp_path)
# gleicher Titel, ABER verschiedene Seiten (verschiedene URLs/IDs) -> beide bleiben
gate.write(_approved("Gleicher Titel", url="https://x/spaces/BES/pages/111/A"))
gate.write(_approved("Gleicher Titel", url="https://x/spaces/BES/pages/222/B"))
files = list((tmp_path / "processed" / "extern" / "pathos").glob("*.md"))
assert len(files) == 2
def test_gate_dedup_per_scope(tmp_path):
# dieselbe Seite fuer intern UND extern -> je Scope ein Dokument (gewollt)
gate = ReviewGate(tmp_path)
gate.write(_approved("FAQ", scope=Scope.INTERN))
gate.write(_approved("FAQ", scope=Scope.EXTERN))
assert (tmp_path / "processed" / "intern" / "pathos").exists()
assert (tmp_path / "processed" / "extern" / "pathos").exists()
def test_prune_duplicate_files_removes_same_page(tmp_path):
from src.store import prune_duplicate_files
d = tmp_path / "processed" / "extern" / "pathos"
d.mkdir(parents=True)
# Basis + Hash-Duplikat derselben Seite (gleiche page-id in url)
fm = ('---\ndomain: "pathos"\nscope: "extern"\n'
'url: "https://x/spaces/BES/pages/611945127/FAQ"\n---\n\nInhalt.\n')
(d / "faq.md").write_text(fm, encoding="utf-8")
(d / "faq-0123456789abcdef.md").write_text(fm, encoding="utf-8")
# eigenstaendige andere Seite mit Hash-Suffix (andere page-id) -> bleibt
(d / "andere-fedcba9876543210.md").write_text(
'---\ndomain: "pathos"\nscope: "extern"\nurl: "https://x/spaces/BES/pages/999/Y"\n---\n\nX.\n',
encoding="utf-8")
removed = prune_duplicate_files(tmp_path)
assert any("faq-0123456789abcdef.md" in r for r in removed)
assert (d / "faq.md").exists() # Basis bleibt
assert not (d / "faq-0123456789abcdef.md").exists() # Duplikat weg
assert (d / "andere-fedcba9876543210.md").exists() # andere Seite bleibt
# ---------------------------------------------------------------------------
# Confluence-Anhaenge (eingebundene PDFs) + Bilder Variante A
# ---------------------------------------------------------------------------
def test_md_converter_keeps_image_alt_and_file_marker():
from src.transformers.md_converter import storage_to_markdown
html = (
'Vorher
'
'
'
'
'
''
'
'
'Nachher
'
)
md = storage_to_markdown(html)
assert "[Bild: screenshot.png]" in md # ac:image alt/Dateiname bleibt
assert "[Bild: Diagramm A]" in md # ![]()
alt bleibt
assert "[Anhang: Handbuch.pdf]" in md # PDF-Embed -> Marker
assert "Vorher" in md and "Nachher" in md
def test_embedded_attachments_detection():
from src.connectors.confluence import ConfluenceConnector
html = (
''
'
'
''
'
'
'
'
)
names = ConfluenceConnector._embedded_attachments(html)
assert names == ["Anlage.pdf", "Skizze.pdf"] # nur Datei-Embeds, kein Bild
def test_to_faq_maps_columns_and_separates_metadata():
"""confluence_faq: benannte Spalten (Frage/Antwort/Thema/Cluster/Ansprechpartner)
werden korrekt zugeordnet; Kopfzeile uebersprungen; Metadaten landen NICHT im
Antworttext, sondern separat -> Ueberschrift und Inhalt passen zusammen."""
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
html = (
""
"| Frage | Antwort | Ansprechpartner | "
"Cluster | Thema |
|---|
"
"| Wie melde ich eine Trasse an? | "
"Schritte:Fertig. | "
"Max Muster | Bestellung | Anmeldung |
"
"
"
)
src = Source(url="u", strategy=Strategy.CONFLUENCE_FAQ, scopes=[Scope.INTERN], domain="pathos")
md = ConfluenceConnector()._to_faq(html, src)
# genau EIN Q/A-Block (Kopfzeile nicht als FAQ)
assert md.count("### ") == 1
assert "### Wie melde ich eine Trasse an?" in md
assert "### Frage" not in md # Header nicht als Frage
# Antwort behaelt Listenstruktur
assert "- Login" in md and "- Formular" in md
# Metadaten getrennt, nicht im Antworttext gemischt
assert "*Thema:* Anmeldung" in md
assert "*Cluster:* Bestellung" in md
assert "*Ansprechpartner:* Max Muster" in md
# Antwort enthaelt die Metadatenwerte NICHT inline
antwort_teil = md.split("*Thema:*")[0]
assert "Max Muster" not in antwort_teil and "Bestellung" not in antwort_teil
def test_to_faq_fallback_without_named_header():
"""Ohne benannte Kopfzeile: Spalte0=Frage, Rest=Antwort (rueckwaertskompatibel)."""
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
html = (""
"| Was ist X? | X ist ein Tool. |
"
"
")
src = Source(url="u", strategy=Strategy.CONFLUENCE_FAQ, scopes=[Scope.INTERN], domain="d")
md = ConfluenceConnector()._to_faq(html, src)
assert "### Was ist X?" in md
assert "X ist ein Tool." in md
def test_filter_excluded_pages():
"""exclude_pages entfernt gezielt Seiten-IDs aus der Tree-Liste."""
from src.connectors.confluence import ConfluenceConnector
ids = ["355496785", "355496791", "999"]
assert ConfluenceConnector._filter_excluded(ids, {}) == ids # ohne Option unveraendert
assert ConfluenceConnector._filter_excluded(ids, {"exclude_pages": ["355496791"]}) == ["355496785", "999"]
assert ConfluenceConnector._filter_excluded(ids, {"exclude_pages": [999]}) == ["355496785", "355496791"]
def test_known_versions_records_component(tmp_path):
"""_known_versions merkt sich die Komponente (page/faq) je Datei, damit eine
FAQ-Quelle eine als 'page' abgelegte Seite nicht faelschlich ueberspringt."""
from src.connectors.confluence import ConfluenceConnector
d = tmp_path / "processed" / "intern" / "pathos"
d.mkdir(parents=True)
(d / "faq-seite.md").write_text(
'---\ndomain: "pathos"\nscope: "intern"\ncomponent_type: "page"\n'
'url: "https://x/pages/355496791"\nsource_version: "63"\n'
'meta_fingerprint: "abc"\n---\n\nInhalt.\n', encoding="utf-8")
known = ConfluenceConnector(data_dir=str(tmp_path))._known_versions("pathos")
assert "355496791" in known
assert known["355496791"]["files"][0]["comp"] == "page"
def test_attachment_docs_creates_pdf_document(tmp_path, monkeypatch):
# PDF-Parsing stubben (keine echten PDF-Bytes noetig) - lokaler Import aus pdf_parser
import src.connectors.pdf_parser as pdfmod
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
monkeypatch.setattr(pdfmod, "pymupdf_markdown",
lambda b: "Geparster PDF-Inhalt, ausreichend lang fuer den Test.")
class FakeClient:
def get_attachments_from_content(self, pid):
return {"results": [{
"title": "Skizze.pdf",
"extensions": {"mediaType": "application/pdf"},
"version": {"number": 4, "when": "2026-06-20T10:00:00.000Z"},
"_links": {"download": "/download/attachments/123/Skizze.pdf?version=4&modificationDate=1"},
}]}
def get(self, link, not_json_response=False):
return b"%PDF-1.4 fake"
html = (''
'
')
src = Source(url="https://x/spaces/BES/pages/123/Seite", strategy=Strategy.CONFLUENCE_PAGE,
scopes=[Scope.INTERN], domain="nur", tags=["nur"])
conn = ConfluenceConnector()
docs = conn._attachment_docs(FakeClient(), "123", "https://x/spaces/BES/pages/123/Seite",
"Seite", html, src, None)
assert len(docs) == 1
d = docs[0]
assert d.kind == "attachment"
assert d.attachment_name == "Skizze.pdf"
assert d.parent_url == "https://x/spaces/BES/pages/123/Seite"
assert d.component_type == ComponentType.PDF
assert "anhang" in d.tags
assert d.source_version == "4"
assert d.last_updated == "2026-06-20" # deterministisch aus modificationDate
md = d.to_markdown()
assert 'kind: "attachment"' in md and 'attachment_name: "Skizze.pdf"' in md
def test_attachment_docs_skips_non_pdf_and_disabled(tmp_path, monkeypatch):
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_attachments_from_content(self, pid):
return {"results": [{"title": "Bild.png", "extensions": {"mediaType": "image/png"},
"version": {"number": 1}, "_links": {"download": "/d/Bild.png"}}]}
def get(self, link, not_json_response=False):
return b"x"
html = (''
'
')
conn = ConfluenceConnector()
# Nicht-PDF -> kein Dokument
src = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="nur")
assert conn._attachment_docs(FakeClient(), "1", "u", "S", html, src, None) == []
# abgeschaltet via options.attachments=false
src2 = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="nur",
options={"attachments": False})
assert conn._attachment_docs(FakeClient(), "1", "u", "S", html, src2, None) == []
def test_build_index_groups_attachments(tmp_path):
from src.store import build_index
d = tmp_path / "processed" / "intern" / "nur"
d.mkdir(parents=True)
page_url = "https://x/spaces/BES/pages/417629865/Architekturskizze"
# Seite
(d / "architekturskizze.md").write_text(
f'---\ndomain: "nur"\ntool: "nur"\nscope: "intern"\nurl: "{page_url}"\n'
'content_hash: "aa"\n---\n\n[Anhang: Skizze.pdf]\n', encoding="utf-8")
# Anhang-Dokument
(d / "architekturskizze-skizze-pdf.md").write_text(
'---\ndomain: "nur"\ntool: "nur"\nscope: "intern"\nkind: "attachment"\n'
f'parent_url: "{page_url}"\nattachment_name: "Skizze.pdf"\n'
'url: "https://x/download/attachments/417629865/Skizze.pdf"\ncontent_hash: "bb"\n---\n\nInhalt.\n',
encoding="utf-8")
idx = build_index(tmp_path)
assert idx["attachments_total"] == 1
page = next(e for e in idx["documents"] if e["kind"] == "document")
assert page["attachments"] and page["attachments"][0]["name"] == "Skizze.pdf"
att = next(e for e in idx["documents"] if e["kind"] == "attachment")
assert att["parent_url"] == page_url
def test_attachment_skip_scope_aware(tmp_path, monkeypatch):
"""Scope-Erweiterung (intern -> intern,extern): neuer Scope wird neu erzeugt,
bekannter Scope+gleiche Version wird uebersprungen."""
import src.connectors.pdf_parser as pdfmod
monkeypatch.setattr(pdfmod, "pymupdf_markdown", lambda b: "PDF-Inhalt fuer den Test " * 5)
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_attachments_from_content(self, pid):
return {"results": [{
"title": "X.pdf",
"extensions": {"mediaType": "application/pdf"},
"version": {"number": 3, "when": "2026-06-20T10:00:00Z"},
"_links": {"download": "https://x.example/download/attachments/1/X.pdf?v=3"},
}]}
def get(self, link, not_json_response=False):
return b"%PDF-1.4 fake"
src = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE,
scopes=[Scope.INTERN, Scope.EXTERN], domain="pathos")
conn = ConfluenceConnector()
# intern ist schon mit Version 3 bekannt, extern nicht -> nur extern wird erzeugt
conn._att_known = {("https://x.example/download/attachments/1/X.pdf", "intern"): "3"}
html = (''
'
')
docs = conn._attachment_docs(FakeClient(), "1", "p-url", "Seite", html, src, None)
assert {d.scope for d in docs} == {Scope.EXTERN}
def test_attachment_url_absolute_passthrough(monkeypatch):
"""Cloud-typischer absoluter download-Link wird nicht mit env_base doppelt praefixiert."""
import src.connectors.pdf_parser as pdfmod
monkeypatch.setattr(pdfmod, "pymupdf_markdown", lambda b: "PDF-Inhalt " * 10)
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_attachments_from_content(self, pid):
return {"results": [{
"title": "Y.pdf",
"extensions": {"mediaType": "application/pdf"},
"version": {"number": 1, "when": "2026-06-20T10:00:00Z"},
"_links": {"download": "https://cloud.example/wiki/download/Y.pdf?ver=1"},
}]}
def get(self, link, not_json_response=False):
return b"%PDF"
src = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="x")
html = (''
'
')
docs = ConfluenceConnector()._attachment_docs(FakeClient(), "1", "p", "S", html, src, None)
assert docs and docs[0].url == "https://cloud.example/wiki/download/Y.pdf" # absolut, ohne doppelten env_base
def test_attachment_fallback_last_updated_when_no_when(monkeypatch):
"""Fehlt version.when, wird last_updated deterministisch auf v gesetzt (kein heute-Churn)."""
import src.connectors.pdf_parser as pdfmod
monkeypatch.setattr(pdfmod, "pymupdf_markdown", lambda b: "Inhalt " * 20)
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_attachments_from_content(self, pid):
return {"results": [{
"title": "Z.pdf", "extensions": {"mediaType": "application/pdf"},
"version": {"number": 7}, # KEIN 'when'
"_links": {"download": "/download/Z.pdf"},
}]}
def get(self, link, not_json_response=False):
return b"%PDF"
src = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="x")
html = (''
'
')
docs = ConfluenceConnector()._attachment_docs(FakeClient(), "1", "p", "S", html, src, None)
assert docs[0].last_updated == "v7"
def test_known_attachment_versions_scope_aware(tmp_path):
"""_known_attachment_versions trackt (url, scope) -> version aus dem Bestand."""
d = tmp_path / "processed" / "extern" / "pathos"
d.mkdir(parents=True)
(d / "a.md").write_text(
'---\ndomain: "pathos"\nscope: "extern"\nkind: "attachment"\n'
'url: "https://x/dl/A.pdf"\nsource_version: "5"\n---\nx\n', encoding="utf-8")
e = tmp_path / "processed" / "intern" / "pathos"
e.mkdir(parents=True)
(e / "b.md").write_text(
'---\ndomain: "pathos"\nscope: "intern"\nkind: "attachment"\n'
'url: "https://x/dl/A.pdf"\nsource_version: "4"\n---\nx\n', encoding="utf-8")
from src.connectors.confluence import ConfluenceConnector
known = ConfluenceConnector(data_dir=str(tmp_path))._known_attachment_versions("pathos")
assert known[("https://x/dl/A.pdf", "extern")] == "5"
assert known[("https://x/dl/A.pdf", "intern")] == "4"
def test_chunk_min_doc_chars_keeps_small_doc_whole():
"""Dokumente unter der Zeichen-Schwelle bleiben ganz, selbst wenn chunk aktiv ist."""
from src.transformers.chunker import chunk_document
short = "## A\n\n" + ("kurz. " * 20) + "\n\n## B\n\n" + ("auch. " * 20)
doc = make_doc(body=short, scope=Scope.ALLGEMEIN, domain="x")
opts = {"chunk": "headings", "chunk_min_chars": 1, "chunk_context_header": False,
"chunk_min_doc_chars": 10000}
assert chunk_document(doc, opts) == []
long_body = ("## A\n\n" + ("Satz mit ein paar Wörtern fuer mehr Inhalt. " * 200) + "\n\n## B\n\n" +
("Satz mit ein paar Wörtern fuer mehr Inhalt. " * 200))
doc2 = make_doc(body=long_body, scope=Scope.ALLGEMEIN, domain="x")
assert len(long_body) >= 10000
chunks = chunk_document(doc2, opts)
assert len(chunks) >= 2
def test_chunk_kind_filter_only_attachments():
"""chunk_kind beschraenkt das Chunking auf ein bestimmtes Document-kind."""
from src.transformers.chunker import chunk_document
body = "## A\n\n" + ("Satz. " * 80) + "\n\n## B\n\n" + ("Satz. " * 80)
page = make_doc(body=body, scope=Scope.INTERN, domain="pathos")
page.kind = "document"
att = make_doc(body=body, scope=Scope.INTERN, domain="pathos")
att.kind = "attachment"
opts = {"chunk": "headings", "chunk_min_chars": 1, "chunk_context_header": False,
"chunk_kind": "attachment"}
assert chunk_document(page, opts) == [] # Seite bleibt ganz
assert len(chunk_document(att, opts)) >= 2 # Anhang wird gechunkt
def test_chunk_fingerprint_includes_new_options():
"""Aenderung an chunk_min_doc_chars / chunk_kind wirkt sich auf den Fingerprint aus."""
from src.transformers.chunker import chunk_options_fingerprint
base = {"chunk": "headings"}
fp0 = chunk_options_fingerprint(base)
assert fp0 != chunk_options_fingerprint({**base, "chunk_min_doc_chars": 3000})
assert fp0 != chunk_options_fingerprint({**base, "chunk_kind": "attachment"})
# ---------------------------------------------------------------------------
# Reviewer-Findings: H1 (entry-Init), H2 (kind ueber _doc_from_md), M3 (last_updated)
# ---------------------------------------------------------------------------
def test_doc_from_md_carries_kind_and_parent(tmp_path):
"""_doc_from_md uebernimmt kind/parent_url/attachment_name - sonst ist chunk_kind tot."""
from src.chunk import _doc_from_md
meta = {
"scope": "intern", "tool": "pathos", "domain": "pathos",
"kind": "attachment", "parent_url": "https://x/spaces/BES/pages/123/Seite",
"attachment_name": "A.pdf", "component_type": "pdf",
"url": "https://x/download/A.pdf", "last_updated": "2026-06-10",
}
doc = _doc_from_md(meta, "Inhalt", fallback_title="A")
assert doc.kind == "attachment"
assert doc.parent_url.endswith("/pages/123/Seite")
assert doc.attachment_name == "A.pdf"
def test_chunk_kind_filter_via_offline_pipeline(tmp_path):
"""End-to-end: chunk_kind=attachment chunkt nur Anhang-Doks (echtes Frontmatter)."""
from src.chunk import chunk_location
proc = tmp_path / "processed" / "intern" / "pathos"
proc.mkdir(parents=True)
long_body = "## H1\n\n" + ("Inhalt zum Chunken " * 600) + "\n\n## H2\n\n" + ("mehr Inhalt " * 600)
# Seite (kind=document, default)
page = make_doc(body=long_body, scope=Scope.INTERN, domain="pathos")
page.title = "Seite"
page.url = "https://x/spaces/BES/pages/123/Seite"
page.review_status = ReviewStatus.APPROVED
(proc / "seite.md").write_text(page.to_markdown(), encoding="utf-8")
# Anhang
att = make_doc(body=long_body, scope=Scope.INTERN, domain="pathos")
att.title = "Seite - A.pdf"
att.url = "https://x/download/A.pdf"
att.kind = "attachment"
att.parent_url = page.url
att.attachment_name = "A.pdf"
att.review_status = ReviewStatus.APPROVED
(proc / "seite-a-pdf.md").write_text(att.to_markdown(), encoding="utf-8")
opts = {"chunk": "headings", "chunk_kind": "attachment", "chunk_min_chars": 1,
"chunk_context_header": False}
docs, chunks, skipped = chunk_location(tmp_path, ("intern", "pathos", ""), opts)
assert docs == 1 and chunks >= 2 # nur der Anhang wurde gechunkt
assert (tmp_path / "chunks" / "intern" / "pathos" / "seite-a-pdf").exists()
assert not (tmp_path / "chunks" / "intern" / "pathos" / "seite").exists()
def test_chunk_location_multiple_configs_per_location(tmp_path):
"""Am selben Ort koennen mehrere Chunk-Configs gelten: Anhang-PDFs via
chunk_kind=attachment (headings), FAQ-Seiten via chunk_component=faq (faq) -
ohne sich zu ueberschreiben; normale Seiten bleiben ungechunkt."""
from src.chunk import chunk_location
proc = tmp_path / "processed" / "intern" / "pathos"
proc.mkdir(parents=True)
body = "## H1\n\n" + ("Inhalt " * 80) + "\n\n## H2\n\n" + ("mehr " * 80)
faq_body = "### Frage A\n\n" + ("Antwort A " * 40) + "\n\n### Frage B\n\n" + ("Antwort B " * 40)
# normale Seite (component page) -> soll NICHT gechunkt werden
page = make_doc(body=body, scope=Scope.INTERN, domain="pathos")
page.title = "Seite"
page.review_status = ReviewStatus.APPROVED
(proc / "seite.md").write_text(page.to_markdown(), encoding="utf-8")
# FAQ (component faq) -> faq-Chunking (je Frage)
faq = make_doc(body=faq_body, scope=Scope.INTERN, domain="pathos")
faq.title = "FAQ"
faq.component_type = ComponentType.FAQ
faq.review_status = ReviewStatus.APPROVED
(proc / "faq.md").write_text(faq.to_markdown(), encoding="utf-8")
# Anhang-PDF (kind attachment) -> headings-Chunking
att = make_doc(body=body, scope=Scope.INTERN, domain="pathos")
att.title = "Anlage"
att.component_type = ComponentType.PDF
att.kind = "attachment"
att.review_status = ReviewStatus.APPROVED
(proc / "anlage.md").write_text(att.to_markdown(), encoding="utf-8")
opts_list = [
{"chunk": "headings", "chunk_kind": "attachment", "chunk_min_chars": 1,
"chunk_context_header": False},
{"chunk": "faq", "chunk_component": "faq", "chunk_min_chars": 1,
"chunk_context_header": False},
]
docs, chunks, _ = chunk_location(tmp_path, ("intern", "pathos", ""), opts_list)
cr = tmp_path / "chunks" / "intern" / "pathos"
assert docs == 2 # FAQ + Anlage, nicht die Seite
assert (cr / "faq").is_dir() and len(list((cr / "faq").glob("*.md"))) == 2 # je Frage
assert (cr / "anlage").is_dir() # Anhang via headings
assert not (cr / "seite").exists() # Seite ungechunkt
def test_build_index_counts_chunked_attachment(tmp_path):
"""Regression: gechunkte Anhang-PDFs (chunk_kind=attachment) erhalten im Index
einen chunks-Block und zaehlen in chunks_total mit (frueher nur Voll-Dokumente)."""
from src.chunk import chunk_location
from src.store import build_index
proc = tmp_path / "processed" / "intern" / "pathos"
proc.mkdir(parents=True)
long_body = "## H1\n\n" + ("Inhalt zum Chunken " * 600) + "\n\n## H2\n\n" + ("mehr Inhalt " * 600)
# Voll-Seite (kind=document) - wird bei chunk_kind=attachment NICHT gechunkt
page = make_doc(body="Kurze Seite.", scope=Scope.INTERN, domain="pathos")
page.title = "Seite"
page.url = "https://x/spaces/BES/pages/123/Seite"
page.review_status = ReviewStatus.APPROVED
(proc / "seite.md").write_text(page.to_markdown(), encoding="utf-8")
# Anhang-PDF (kind=attachment) - wird gechunkt
att = make_doc(body=long_body, scope=Scope.INTERN, domain="pathos")
att.title = "Seite - A.pdf"
att.url = "https://x/download/A.pdf"
att.kind = "attachment"
att.parent_url = page.url
att.attachment_name = "A.pdf"
att.review_status = ReviewStatus.APPROVED
(proc / "seite-a-pdf.md").write_text(att.to_markdown(), encoding="utf-8")
opts = {"chunk": "headings", "chunk_kind": "attachment", "chunk_min_chars": 1,
"chunk_context_header": False}
_, chunks, _ = chunk_location(tmp_path, ("intern", "pathos", ""), opts)
assert chunks >= 2
idx = build_index(tmp_path)
att_entry = next(e for e in idx["documents"] if e["kind"] == "attachment")
assert att_entry["chunks"] and att_entry["chunks"]["count"] == chunks
assert att_entry["chunks"]["path"].startswith("chunks/intern/pathos/seite-a-pdf")
assert idx["chunks_total"] == chunks # Anhang-Chunks werden mitgezaehlt
assert idx["by_domain"]["pathos"]["chunks"] == chunks
# Voll-Seite ohne Chunks behaelt chunks=None
page_entry = next(e for e in idx["documents"] if e["kind"] == "document")
assert page_entry["chunks"] is None
def test_attachment_unknown_last_updated_when_no_version_and_no_when(monkeypatch):
"""Ohne version.number UND ohne when wird last_updated = 'unknown' (kein heute-Churn)."""
import src.connectors.pdf_parser as pdfmod
monkeypatch.setattr(pdfmod, "pymupdf_markdown", lambda b: "Inhalt " * 20)
from src.connectors.confluence import ConfluenceConnector
from src.model import Scope, Source, Strategy
class FakeClient:
def get_attachments_from_content(self, pid):
return {"results": [{
"title": "W.pdf", "extensions": {"mediaType": "application/pdf"},
"version": {}, # weder number noch when
"_links": {"download": "/download/W.pdf"},
}]}
def get(self, link, not_json_response=False):
return b"%PDF"
src = Source(url="u", strategy=Strategy.CONFLUENCE_PAGE, scopes=[Scope.INTERN], domain="x")
html = (''
'
')
docs = ConfluenceConnector()._attachment_docs(FakeClient(), "1", "p", "S", html, src, None)
assert docs[0].last_updated == "unknown"