torbenandClaude Code
9c0bb6bfbf
Adds the import-filter registry (tools/chemenu/filters), the raw derive command with folder form, path-budget and name checks, DOCX/PPTX/XLSX and EML filters, the pending and accept hints, the wiki-ingest skill step and the raw/CONTRACT.md section. First of three cuts; containers and PDF/MSG follow. Files changed: - CHANGES.md - VERSION - docs/how-to/ingest-a-source.md - docs/reference/commands/README.md - docs/reference/commands/raw-material-and-uploads.md - instructions/wiki-ingest/SKILL.md - raw/CONTRACT.md - tools/CONTRACT.md - tools/README.md - tools/chemenu/cli_contract.py - tools/chemenu/commands/raw_cmd.py - tools/chemenu/filters/__init__.py - tools/chemenu/filters/html_page.py - tools/chemenu/filters/mail.py - tools/chemenu/filters/ooxml.py - tools/chemenu/tests/test_raw_derive.py - tools/chemenu/web_capture.py Co-Authored-By: Claude Code <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SnAJ7Z3CpVD3PRbN73QtU2
677 lines
26 KiB
Python
677 lines
26 KiB
Python
"""`wikitool raw derive` - reading text from documents that are not text.
|
|
|
|
The documents are built here from `zipfile` and `email`, so every fixture is
|
|
readable in the test that uses it and no binary sits in the repository.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import zipfile
|
|
from email.message import EmailMessage
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import typer
|
|
import yaml
|
|
|
|
from chemenu import filters, web_capture
|
|
from chemenu.commands.raw_cmd import (
|
|
raw_accept_command,
|
|
raw_derive_command,
|
|
raw_pending_command,
|
|
)
|
|
|
|
W = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
|
|
A = "http://schemas.openxmlformats.org/drawingml/2006/main"
|
|
P = "http://schemas.openxmlformats.org/presentationml/2006/main"
|
|
S = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
|
|
R = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
|
|
REL = "http://schemas.openxmlformats.org/package/2006/relationships"
|
|
DOC_REL = R + "/officeDocument"
|
|
|
|
|
|
def _zip(parts: dict[str, str]) -> bytes:
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for name, content in parts.items():
|
|
zf.writestr(name, content)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _rels(target: str, kind: str = DOC_REL, extra: str = "") -> str:
|
|
return (
|
|
f'<Relationships xmlns="{REL}"><Relationship Id="rId1" Type="{kind}" '
|
|
f'Target="{target}"/>{extra}</Relationships>'
|
|
)
|
|
|
|
|
|
def make_docx(body: str, *, comments: str = "", title: str = "") -> bytes:
|
|
parts = {
|
|
"_rels/.rels": _rels("word/document.xml"),
|
|
"word/document.xml": f'<w:document xmlns:w="{W}"><w:body>{body}</w:body></w:document>',
|
|
"word/styles.xml": (
|
|
f'<w:styles xmlns:w="{W}"><w:style w:styleId="Titel"><w:name w:val="heading 1"/></w:style>'
|
|
"</w:styles>"
|
|
),
|
|
"word/numbering.xml": (
|
|
f'<w:numbering xmlns:w="{W}"><w:abstractNum w:abstractNumId="0">'
|
|
'<w:lvl w:ilvl="0"><w:numFmt w:val="decimal"/></w:lvl></w:abstractNum>'
|
|
'<w:abstractNum w:abstractNumId="1"><w:lvl w:ilvl="0"><w:numFmt w:val="bullet"/></w:lvl>'
|
|
'</w:abstractNum><w:num w:numId="1"><w:abstractNumId w:val="0"/></w:num>'
|
|
'<w:num w:numId="2"><w:abstractNumId w:val="1"/></w:num></w:numbering>'
|
|
),
|
|
}
|
|
if comments:
|
|
parts["word/comments.xml"] = f'<w:comments xmlns:w="{W}">{comments}</w:comments>'
|
|
if title:
|
|
parts["docProps/core.xml"] = (
|
|
'<cp:coreProperties xmlns:cp="http://schemas.openxmlformats.org/package/2006/metadata/'
|
|
'core-properties" xmlns:dc="http://purl.org/dc/elements/1.1/">'
|
|
f"<dc:title>{title}</dc:title></cp:coreProperties>"
|
|
)
|
|
return _zip(parts)
|
|
|
|
|
|
def _p(text: str, *, style: str = "", num: tuple[int, int] | None = None) -> str:
|
|
props = ""
|
|
if style:
|
|
props += f'<w:pStyle w:val="{style}"/>'
|
|
if num:
|
|
props += f'<w:numPr><w:ilvl w:val="0"/><w:numId w:val="{num[1]}"/></w:numPr>'
|
|
return f"<w:p><w:pPr>{props}</w:pPr><w:r><w:t>{text}</w:t></w:r></w:p>"
|
|
|
|
|
|
def make_pptx(slides: list[tuple[str, list[str], str]]) -> bytes:
|
|
"""`(title, bullets, notes)` per slide."""
|
|
parts = {"_rels/.rels": _rels("ppt/presentation.xml")}
|
|
ids, rels = "", ""
|
|
for number, (title, bullets, notes) in enumerate(slides, 1):
|
|
ids += f'<p:sldId id="{255 + number}" r:id="rId{number}"/>'
|
|
rels += (
|
|
f'<Relationship Id="rId{number}" Type="{R}/slide" Target="slides/slide{number}.xml"/>'
|
|
)
|
|
body = "".join(f"<a:p><a:r><a:t>{b}</a:t></a:r></a:p>" for b in bullets)
|
|
parts[f"ppt/slides/slide{number}.xml"] = (
|
|
f'<p:sld xmlns:p="{P}" xmlns:a="{A}"><p:cSld><p:spTree>'
|
|
f'<p:sp><p:nvSpPr><p:nvPr><p:ph type="title"/></p:nvPr></p:nvSpPr><p:txBody>'
|
|
f"<a:p><a:r><a:t>{title}</a:t></a:r></a:p></p:txBody></p:sp>"
|
|
f'<p:sp><p:nvSpPr><p:nvPr><p:ph type="body"/></p:nvPr></p:nvSpPr><p:txBody>{body}</p:txBody></p:sp>'
|
|
f'<p:sp><p:nvSpPr><p:nvPr><p:ph type="sldNum"/></p:nvPr></p:nvSpPr><p:txBody>'
|
|
f"<a:p><a:r><a:t>FOOTER-NUMBER</a:t></a:r></a:p></p:txBody></p:sp>"
|
|
"</p:spTree></p:cSld></p:sld>"
|
|
)
|
|
if notes:
|
|
parts[f"ppt/slides/_rels/slide{number}.xml.rels"] = (
|
|
f'<Relationships xmlns="{REL}"><Relationship Id="rId1" Type="{R}/notesSlide" '
|
|
f'Target="../notesSlides/notes{number}.xml"/></Relationships>'
|
|
)
|
|
parts[f"ppt/notesSlides/notes{number}.xml"] = (
|
|
f'<p:notes xmlns:p="{P}" xmlns:a="{A}"><p:cSld><p:spTree><p:sp><p:nvSpPr><p:nvPr>'
|
|
'<p:ph type="body"/></p:nvPr></p:nvSpPr><p:txBody>'
|
|
f"<a:p><a:r><a:t>{notes}</a:t></a:r></a:p></p:txBody></p:sp></p:spTree></p:cSld></p:notes>"
|
|
)
|
|
parts["ppt/presentation.xml"] = (
|
|
f'<p:presentation xmlns:p="{P}" xmlns:r="{R}"><p:sldIdLst>{ids}</p:sldIdLst></p:presentation>'
|
|
)
|
|
parts["ppt/_rels/presentation.xml.rels"] = f'<Relationships xmlns="{REL}">{rels}</Relationships>'
|
|
return _zip(parts)
|
|
|
|
|
|
def make_xlsx() -> bytes:
|
|
sheet_rel = lambda n: f'<Relationship Id="rId{n}" Type="{R}/worksheet" Target="worksheets/sheet{n}.xml"/>' # noqa: E731
|
|
return _zip({
|
|
"_rels/.rels": _rels("xl/workbook.xml"),
|
|
"xl/workbook.xml": (
|
|
f'<workbook xmlns="{S}" xmlns:r="{R}"><sheets>'
|
|
'<sheet name="Budget" sheetId="1" r:id="rId1"/>'
|
|
'<sheet name="Geheim" sheetId="2" state="hidden" r:id="rId2"/></sheets></workbook>'
|
|
),
|
|
"xl/_rels/workbook.xml.rels": (
|
|
f'<Relationships xmlns="{REL}">{sheet_rel(1)}{sheet_rel(2)}'
|
|
f'<Relationship Id="rId3" Type="{R}/sharedStrings" Target="sharedStrings.xml"/>'
|
|
f'<Relationship Id="rId4" Type="{R}/styles" Target="styles.xml"/></Relationships>'
|
|
),
|
|
"xl/sharedStrings.xml": (
|
|
f'<sst xmlns="{S}"><si><t>Posten</t></si><si><t>Miete</t></si>'
|
|
"<si><t>Wert</t></si></sst>"
|
|
),
|
|
"xl/styles.xml": (
|
|
f'<styleSheet xmlns="{S}"><cellXfs><xf numFmtId="0"/><xf numFmtId="14"/></cellXfs></styleSheet>'
|
|
),
|
|
"xl/worksheets/sheet1.xml": (
|
|
f'<worksheet xmlns="{S}"><sheetData>'
|
|
'<row r="2"><c r="B2" t="s"><v>0</v></c><c r="C2" t="s"><v>2</v></c></row>'
|
|
'<row r="3"><c r="B3" t="s"><v>1</v></c><c r="C3"><f>1200+0</f><v>1200</v></c></row>'
|
|
'<row r="4"><c r="B4" t="inlineStr"><is><t>Stichtag</t></is></c><c r="C4" s="1"><v>45292</v></c></row>'
|
|
'<row r="5"><c r="B5" t="inlineStr"><is><t>Fertig</t></is></c><c r="C5" t="b"><v>1</v></c></row>'
|
|
"</sheetData></worksheet>"
|
|
),
|
|
"xl/worksheets/sheet2.xml": (
|
|
f'<worksheet xmlns="{S}"><sheetData><row r="1"><c r="A1" t="inlineStr"><is>'
|
|
"<t>VERSTECKT-INHALT</t></is></c></row></sheetData></worksheet>"
|
|
),
|
|
})
|
|
|
|
|
|
def make_eml(body: str = "Hallo Welt\n", *, html: str = "", attach: bool = False, inline: bool = False,
|
|
subject: str = "Angebot") -> bytes:
|
|
msg = EmailMessage()
|
|
msg["From"] = "Anna Beispiel <anna@example.org>"
|
|
msg["To"] = "bob@example.org, Carla <carla@example.org>"
|
|
msg["Subject"] = subject
|
|
msg["Date"] = "Wed, 07 Oct 2026 09:30:00 +0200"
|
|
msg["Message-ID"] = "<abc@example.org>"
|
|
if html:
|
|
msg.set_content(html, subtype="html")
|
|
else:
|
|
msg.set_content(body)
|
|
if inline:
|
|
msg.add_related(b"\x89PNG", maintype="image", subtype="png", cid="<logo@x>")
|
|
if attach:
|
|
msg.add_attachment(b"data", maintype="application", subtype="octet-stream", filename="a.bin")
|
|
return msg.as_bytes()
|
|
|
|
|
|
@pytest.fixture
|
|
def tree(kb_dir):
|
|
root = kb_dir.parent
|
|
(root / "raw").mkdir()
|
|
(root / "incoming").mkdir(exist_ok=True)
|
|
return root
|
|
|
|
|
|
def _derive(path, name=None):
|
|
return raw_derive_command(path=Path(path), name=name)
|
|
|
|
|
|
def _refused(path, name=None) -> None:
|
|
with pytest.raises(typer.Exit) as excinfo:
|
|
_derive(path, name)
|
|
assert excinfo.value.exit_code == 1
|
|
|
|
|
|
def _snapshot(root: Path) -> dict[str, bytes]:
|
|
return {str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file()}
|
|
|
|
|
|
def _header_and_text(md: Path) -> tuple[dict, str]:
|
|
content = md.read_text(encoding="utf-8")
|
|
assert content.startswith("---\n")
|
|
head, _, text = content[4:].partition("\n---\n")
|
|
return yaml.safe_load(head), text
|
|
|
|
|
|
def _put(tree: Path, name: str, data: bytes) -> Path:
|
|
path = tree / "incoming" / name
|
|
path.write_bytes(data)
|
|
return path
|
|
|
|
|
|
# --- the single-file form ---------------------------------------------------
|
|
|
|
|
|
def test_a_single_file_becomes_a_folder_with_original_and_derived_text(tree):
|
|
data = make_docx(_p("Hello"))
|
|
_put(tree, "Angebot XYZ.docx", data)
|
|
|
|
_derive("incoming/Angebot XYZ.docx")
|
|
|
|
folder = tree / "incoming/Angebot XYZ"
|
|
assert sorted(p.name for p in folder.iterdir()) == ["derived.md", "original.docx"]
|
|
assert (folder / "original.docx").read_bytes() == data
|
|
assert not (tree / "incoming/Angebot XYZ.docx").exists()
|
|
|
|
|
|
def test_the_header_is_valid_yaml_and_records_the_derivation(tree):
|
|
_put(tree, "Angebot XYZ.docx", make_docx(_p("Hello"), title="Offer"))
|
|
_derive("incoming/Angebot XYZ.docx")
|
|
header, text = _header_and_text(tree / "incoming/Angebot XYZ/derived.md")
|
|
|
|
assert header["derived_by"] == "wikitool raw derive"
|
|
assert header["filter"] == "docx"
|
|
assert header["engine"].startswith("wikitool ")
|
|
assert header["derived_from"] == "original.docx"
|
|
assert header["original_name"] == "Angebot XYZ.docx"
|
|
assert header["title"] == "Offer"
|
|
assert header["derived"] is not None
|
|
assert text.strip() == "Hello"
|
|
|
|
|
|
def test_two_derivations_of_the_same_bytes_differ_only_in_stamp_and_file_name(tree):
|
|
data = make_docx(_p("Hello") + _p("World"))
|
|
_put(tree, "a.docx", data)
|
|
_put(tree, "b.docx", data)
|
|
_derive("incoming/a.docx", "same")
|
|
_derive("incoming/b.docx", "same-too")
|
|
|
|
def stripped(name):
|
|
lines = (tree / "incoming" / name / "derived.md").read_text(encoding="utf-8").splitlines()
|
|
return [line for line in lines if not line.startswith(("derived:", "original_name:"))]
|
|
|
|
assert stripped("same") == stripped("same-too")
|
|
|
|
|
|
def test_raw_accept_takes_the_derived_folder_as_one_source(tree):
|
|
data = make_docx(_p("Hello"))
|
|
_put(tree, "report.docx", data)
|
|
_derive("incoming/report.docx")
|
|
|
|
raw_accept_command(
|
|
files=[Path("incoming/report")], fidelity="verbatim", authority="reporting",
|
|
page=None, replaces=None, dry_run=False,
|
|
)
|
|
|
|
[original] = list((tree / "raw").rglob("original.docx"))
|
|
assert original.read_bytes() == data
|
|
assert (original.parent / "derived.md").is_file()
|
|
assert original.parent.name == "report"
|
|
assert not (tree / "incoming/report").exists()
|
|
|
|
|
|
def test_name_picks_the_folder_stem(tree):
|
|
_put(tree, "x.docx", make_docx(_p("Hello")))
|
|
_derive("incoming/x.docx", "kurz")
|
|
assert (tree / "incoming/kurz/original.docx").is_file()
|
|
|
|
|
|
def test_a_120_character_stem_fits_and_an_over_budget_one_is_refused(tree, capsys):
|
|
long_stem = "s" * 120
|
|
_put(tree, f"{long_stem}.docx", make_docx(_p("Hello")))
|
|
_derive(f"incoming/{long_stem}.docx")
|
|
assert (tree / "incoming" / long_stem / "derived.md").is_file()
|
|
capsys.readouterr()
|
|
|
|
too_long = "t" * 170
|
|
_put(tree, f"{too_long}.docx", make_docx(_p("Hello")))
|
|
before = _snapshot(tree / "incoming")
|
|
_refused(f"incoming/{too_long}.docx")
|
|
out = capsys.readouterr().out
|
|
assert "--name" in out and "raw/" in out
|
|
assert _snapshot(tree / "incoming") == before
|
|
_derive(f"incoming/{too_long}.docx", "short")
|
|
assert (tree / "incoming/short/original.docx").is_file()
|
|
|
|
|
|
def test_a_stem_taken_under_raw_is_a_warning_not_a_refusal(tree, capsys):
|
|
(tree / "raw/2026/01").mkdir(parents=True)
|
|
(tree / "raw/2026/01/report.txt").write_text("x", encoding="utf-8")
|
|
_put(tree, "report.docx", make_docx(_p("Hello")))
|
|
_derive("incoming/report.docx")
|
|
out = capsys.readouterr().out
|
|
assert "WARN" in out and "already claims" in out
|
|
assert (tree / "incoming/report/derived.md").is_file()
|
|
|
|
|
|
def test_an_existing_target_folder_is_refused_and_nothing_changes(tree):
|
|
_put(tree, "report.docx", make_docx(_p("Hello")))
|
|
(tree / "incoming/report").mkdir()
|
|
before = _snapshot(tree / "incoming")
|
|
_refused("incoming/report.docx")
|
|
assert _snapshot(tree / "incoming") == before
|
|
|
|
|
|
def test_a_file_outside_incoming_or_in_a_subfolder_is_refused(tree):
|
|
outside = tree / "elsewhere.docx"
|
|
outside.write_bytes(make_docx(_p("Hello")))
|
|
_refused(outside)
|
|
(tree / "incoming/sub").mkdir()
|
|
(tree / "incoming/sub/in.docx").write_bytes(make_docx(_p("Hello")))
|
|
_refused("incoming/sub/in.docx")
|
|
assert (tree / "incoming/sub/in.docx").is_file()
|
|
|
|
|
|
def test_a_file_without_a_filter_is_refused_and_names_what_is_read(tree, capsys):
|
|
_put(tree, "notes.md", b"# notes\n")
|
|
_refused("incoming/notes.md")
|
|
out = capsys.readouterr().out
|
|
assert ".docx" in out and ".xlsx" in out
|
|
assert (tree / "incoming/notes.md").is_file()
|
|
|
|
|
|
def test_a_failure_while_moving_the_original_leaves_incoming_as_it_was(tree, monkeypatch):
|
|
_put(tree, "report.docx", make_docx(_p("Hello")))
|
|
before = _snapshot(tree / "incoming")
|
|
|
|
def broken(self, target):
|
|
raise OSError("disk says no")
|
|
|
|
monkeypatch.setattr(Path, "rename", broken)
|
|
_refused("incoming/report.docx")
|
|
monkeypatch.undo()
|
|
assert _snapshot(tree / "incoming") == before
|
|
assert not (tree / "incoming/report").exists()
|
|
|
|
|
|
def test_a_damaged_document_is_refused_untouched(tree):
|
|
_put(tree, "broken.docx", b"this is not a zip")
|
|
before = _snapshot(tree / "incoming")
|
|
_refused("incoming/broken.docx")
|
|
assert _snapshot(tree / "incoming") == before
|
|
|
|
|
|
def test_a_lone_markdown_file_stays_folderless(tree):
|
|
_put(tree, "notes.md", b"# notes\n")
|
|
raw_accept_command(
|
|
files=[tree / "incoming/notes.md"], fidelity="verbatim", authority="opinion",
|
|
page=None, replaces=None, dry_run=False,
|
|
)
|
|
[moved] = [p for p in (tree / "raw").rglob("*") if p.is_file()]
|
|
assert moved.name == "notes.md" and moved.parent.name.isdigit()
|
|
|
|
|
|
# --- the folder form --------------------------------------------------------
|
|
|
|
|
|
def test_a_folder_gets_a_md_beside_each_file_with_a_filter(tree, capsys):
|
|
folder = tree / "incoming/mappe"
|
|
folder.mkdir()
|
|
(folder / "plan.docx").write_bytes(make_docx(_p("Plan")))
|
|
(folder / "zahlen.xlsx").write_bytes(make_xlsx())
|
|
(folder / "foto.jpg").write_bytes(b"\xff\xd8")
|
|
_derive("incoming/mappe")
|
|
out = capsys.readouterr().out
|
|
|
|
assert sorted(p.name for p in folder.iterdir()) == ["foto.jpg", "plan.docx", "plan.md", "zahlen.md", "zahlen.xlsx"]
|
|
header, _ = _header_and_text(folder / "plan.md")
|
|
assert header["derived_from"] == "plan.docx"
|
|
assert "original_name" not in header
|
|
assert "foto.jpg" in out
|
|
|
|
|
|
def test_a_folder_with_an_existing_md_is_refused_untouched(tree):
|
|
folder = tree / "incoming/mappe"
|
|
folder.mkdir()
|
|
(folder / "plan.docx").write_bytes(make_docx(_p("Plan")))
|
|
(folder / "plan.md").write_text("mine\n", encoding="utf-8")
|
|
before = _snapshot(tree / "incoming")
|
|
_refused("incoming/mappe")
|
|
assert _snapshot(tree / "incoming") == before
|
|
|
|
|
|
def test_a_folder_where_two_files_derive_to_one_md_is_refused(tree):
|
|
folder = tree / "incoming/mappe"
|
|
folder.mkdir()
|
|
(folder / "plan.docx").write_bytes(make_docx(_p("Plan")))
|
|
(folder / "plan.xlsx").write_bytes(make_xlsx())
|
|
before = _snapshot(tree / "incoming")
|
|
_refused("incoming/mappe")
|
|
assert _snapshot(tree / "incoming") == before
|
|
|
|
|
|
def test_a_folder_without_any_filter_file_is_refused(tree):
|
|
folder = tree / "incoming/mappe"
|
|
folder.mkdir()
|
|
(folder / "foto.jpg").write_bytes(b"\xff\xd8")
|
|
_refused("incoming/mappe")
|
|
|
|
|
|
def test_a_folder_with_a_hidden_entry_is_refused(tree):
|
|
folder = tree / "incoming/mappe"
|
|
folder.mkdir()
|
|
(folder / "plan.docx").write_bytes(make_docx(_p("Plan")))
|
|
(folder / ".DS_Store").write_bytes(b"x")
|
|
_refused("incoming/mappe")
|
|
|
|
|
|
def test_name_does_not_apply_to_a_folder(tree):
|
|
(tree / "incoming/mappe").mkdir()
|
|
(tree / "incoming/mappe/plan.docx").write_bytes(make_docx(_p("Plan")))
|
|
_refused("incoming/mappe", "other")
|
|
|
|
|
|
# --- DOCX -------------------------------------------------------------------
|
|
|
|
|
|
def test_docx_headings_lists_tables_and_text(tree):
|
|
body = (
|
|
_p("Titel", style="Titel")
|
|
+ _p("Absatz eins.")
|
|
+ _p("Erster", num=(0, 1)) + _p("Zweiter", num=(0, 1))
|
|
+ _p("Punkt", num=(0, 2))
|
|
+ f'<w:tbl><w:tr><w:tc>{_p("A")}</w:tc><w:tc>{_p("B")}</w:tc></w:tr>'
|
|
f'<w:tr><w:tc>{_p("1")}</w:tc><w:tc>{_p("2")}</w:tc></w:tr></w:tbl>'
|
|
)
|
|
_put(tree, "d.docx", make_docx(body))
|
|
_derive("incoming/d.docx")
|
|
_, text = _header_and_text(tree / "incoming/d/derived.md")
|
|
|
|
assert "# Titel" in text
|
|
assert "1. Erster\n2. Zweiter" in text
|
|
assert "- Punkt" in text
|
|
assert "| A | B |" in text and "| 1 | 2 |" in text
|
|
|
|
|
|
def test_docx_tracked_changes_and_comments(tree, capsys):
|
|
body = (
|
|
"<w:p><w:r><w:t>Alt </w:t></w:r>"
|
|
'<w:del w:id="1"><w:r><w:delText>GELOESCHT </w:delText></w:r></w:del>'
|
|
'<w:ins w:id="2"><w:r><w:t>NEU</w:t></w:r></w:ins></w:p>'
|
|
)
|
|
comments = '<w:comment w:id="0" w:author="Anna" w:date="2026-10-01"><w:p><w:r><w:t>Prüfen</w:t></w:r></w:p></w:comment>'
|
|
_put(tree, "d.docx", make_docx(body, comments=comments))
|
|
_derive("incoming/d.docx")
|
|
out = capsys.readouterr().out
|
|
header, text = _header_and_text(tree / "incoming/d/derived.md")
|
|
|
|
assert "Alt NEU" in text and "GELOESCHT" not in text
|
|
assert "## Comments" in text and "Anna" in text and "Prüfen" in text
|
|
assert header["tracked_changes"] == 2 and header["comments"] == 1
|
|
assert "CHECK" in out and "Tracked changes" in out and "Comments" in out
|
|
|
|
|
|
def test_docx_without_tracked_changes_prints_no_such_check(tree, capsys):
|
|
_put(tree, "d.docx", make_docx(_p("Plain")))
|
|
_derive("incoming/d.docx")
|
|
out = capsys.readouterr().out
|
|
assert "Tracked changes" not in out and "Comments" not in out
|
|
|
|
|
|
# --- PPTX -------------------------------------------------------------------
|
|
|
|
|
|
def test_pptx_slides_notes_and_footer_placeholders(tree):
|
|
_put(tree, "deck.pptx", make_pptx([("Eins", ["a", "b"], "Sprechnotiz"), ("Zwei", ["c"], "")]))
|
|
_derive("incoming/deck.pptx")
|
|
header, text = _header_and_text(tree / "incoming/deck/derived.md")
|
|
|
|
assert header["slides"] == 2
|
|
assert text.index("## Slide 1") < text.index("## Slide 2")
|
|
assert "**Eins**" in text and "- a" in text and "- b" in text
|
|
assert "### Notes" in text and "Sprechnotiz" in text
|
|
assert "FOOTER-NUMBER" not in text
|
|
|
|
|
|
# --- XLSX -------------------------------------------------------------------
|
|
|
|
|
|
def test_xlsx_values_hidden_sheet_and_trimmed_edges(tree, capsys):
|
|
_put(tree, "zahlen.xlsx", make_xlsx())
|
|
_derive("incoming/zahlen.xlsx")
|
|
out = capsys.readouterr().out
|
|
header, text = _header_and_text(tree / "incoming/zahlen/derived.md")
|
|
|
|
assert header["sheets"] == 2 and header["hidden_sheets"] == ["Geheim"]
|
|
assert "## Budget" in text and "## Geheim" in text and "VERSTECKT-INHALT" in text
|
|
assert "| Posten | Wert |" in text
|
|
assert "| Miete | 1200 |" in text
|
|
assert "2024-01-01" in text
|
|
assert "TRUE" in text
|
|
assert text.splitlines()[text.splitlines().index("## Budget") + 2].startswith("| Posten")
|
|
assert "Values are as last saved" in out and "Hidden sheets" in out
|
|
|
|
|
|
def test_xlsx_check_is_printed_only_for_the_formats_derived_in_the_run(tree, capsys):
|
|
_put(tree, "d.docx", make_docx(_p("Plain")))
|
|
_derive("incoming/d.docx")
|
|
assert "last saved" not in capsys.readouterr().out
|
|
|
|
|
|
def test_check_lines_are_at_most_three_per_format_derived(tree, capsys):
|
|
folder = tree / "incoming/mappe"
|
|
folder.mkdir()
|
|
(folder / "z.xlsx").write_bytes(make_xlsx())
|
|
(folder / "m.eml").write_bytes(make_eml("> quoted\nnew\n"))
|
|
_derive("incoming/mappe")
|
|
out = capsys.readouterr().out
|
|
per_filter = filters.MAX_CHECKS
|
|
assert len([line for line in out.splitlines() if line.startswith("CHECK")]) <= 2 * per_filter
|
|
|
|
|
|
# --- EML --------------------------------------------------------------------
|
|
|
|
|
|
def test_eml_headers_and_plain_body(tree):
|
|
_put(tree, "mail.eml", make_eml("Hallo Welt\n"))
|
|
_derive("incoming/mail.eml")
|
|
header, text = _header_and_text(tree / "incoming/mail/derived.md")
|
|
|
|
assert header["filter"] == "eml"
|
|
assert header["from"] == "Anna Beispiel <anna@example.org>"
|
|
assert header["to"] == ["bob@example.org", "Carla <carla@example.org>"]
|
|
assert header["cc"] == []
|
|
assert header["subject"] == "Angebot" and header["title"] == "Angebot"
|
|
assert header["message_id"] == "<abc@example.org>"
|
|
assert str(header["date"]).startswith("2026-10-07")
|
|
assert header["attachments"] == []
|
|
assert text.strip() == "Hallo Welt"
|
|
|
|
|
|
def test_eml_html_only_is_derived_through_the_web_filter(tree):
|
|
_put(tree, "mail.eml", make_eml(html="<html><body><h1>Gruß</h1><p>Text</p></body></html>"))
|
|
_derive("incoming/mail.eml")
|
|
_, text = _header_and_text(tree / "incoming/mail/derived.md")
|
|
assert "# Gruß" in text and "Text" in text
|
|
|
|
|
|
def test_eml_check_lines_cover_quotes_only_when_they_occur(tree, capsys):
|
|
_put(tree, "plain.eml", make_eml("Fresh text\n"))
|
|
_derive("incoming/plain.eml")
|
|
out = capsys.readouterr().out
|
|
assert "Quoted earlier messages" not in out
|
|
assert "commitment" in out and "secrets" in out
|
|
|
|
_put(tree, "reply.eml", make_eml("Yes.\n\n> earlier\n"))
|
|
_derive("incoming/reply.eml")
|
|
out = capsys.readouterr().out
|
|
assert "Quoted earlier messages" in out
|
|
lines = [line for line in out.splitlines() if line.startswith("CHECK")]
|
|
assert len(lines) == filters.MAX_CHECKS
|
|
|
|
|
|
def test_eml_with_inline_image_warns_and_is_not_an_attachment(tree, capsys):
|
|
_put(tree, "mail.eml", make_eml(html="<p>Hi</p>", inline=True))
|
|
_derive("incoming/mail.eml")
|
|
assert "inline image" in capsys.readouterr().out
|
|
|
|
|
|
def test_eml_with_an_attachment_is_refused_untouched(tree, capsys):
|
|
_put(tree, "mail.eml", make_eml(attach=True))
|
|
before = _snapshot(tree / "incoming")
|
|
_refused("incoming/mail.eml")
|
|
assert "a.bin" in capsys.readouterr().out
|
|
assert _snapshot(tree / "incoming") == before
|
|
|
|
|
|
# --- the registry -----------------------------------------------------------
|
|
|
|
|
|
class _Needy(filters.Filter):
|
|
name = "needy"
|
|
extensions = (".needy",)
|
|
requires = ("no_such_package_for_wikitool",)
|
|
|
|
|
|
def test_a_missing_package_fails_that_format_only(tree, monkeypatch, capsys):
|
|
monkeypatch.setattr(filters, "_REGISTRY", list(filters.all_filters()) + [_Needy()])
|
|
_put(tree, "x.needy", b"data")
|
|
before = _snapshot(tree / "incoming")
|
|
_refused("incoming/x.needy")
|
|
out = capsys.readouterr().out
|
|
assert "no_such_package_for_wikitool" in out and "preflight" in out
|
|
assert _snapshot(tree / "incoming") == before
|
|
|
|
_put(tree, "ok.docx", make_docx(_p("Fine")))
|
|
_derive("incoming/ok.docx")
|
|
assert (tree / "incoming/ok/derived.md").is_file()
|
|
|
|
|
|
def test_registering_a_second_reader_for_an_extension_is_refused():
|
|
class Twin(filters.Filter):
|
|
name = "twin"
|
|
extensions = (".docx",)
|
|
|
|
with pytest.raises(ValueError):
|
|
filters.register(Twin())
|
|
|
|
|
|
def test_a_filter_cannot_ask_for_more_than_three_check_lines():
|
|
with pytest.raises(ValueError):
|
|
filters.Derivation(text="", checks=("a", "b", "c", "d"))
|
|
|
|
|
|
def test_extension_matching_is_case_insensitive_and_ignores_leading_dots():
|
|
assert filters.extension_of("Report.DOCX") == ".docx"
|
|
assert filters.extension_of(".hidden") == ""
|
|
assert filters.filter_for("Report.DOCX").name == "docx"
|
|
assert filters.filter_for("noextension") is None
|
|
|
|
|
|
def test_header_renders_a_list_as_a_flow_sequence_and_a_string_as_before():
|
|
rendered = web_capture.header([("a", "plain"), ("b", ["x y", "z"]), ("c", [])])
|
|
assert rendered == '---\na: plain\nb: ["x y", "z"]\nc: []\n---\n'
|
|
assert yaml.safe_load(rendered.strip("-\n").replace("---", "")) == {"a": "plain", "b": ["x y", "z"], "c": []}
|
|
|
|
|
|
# --- the nets in raw pending and raw accept ---------------------------------
|
|
|
|
|
|
def test_pending_names_the_derive_line_for_a_lone_file_with_a_filter(tree, capsys):
|
|
_put(tree, "report.docx", make_docx(_p("Hello")))
|
|
_put(tree, "photo.jpg", b"\xff")
|
|
raw_pending_command(json_out=True)
|
|
by_path = {c["paths"][0]: c for c in json.loads(capsys.readouterr().out)}
|
|
assert by_path["incoming/report.docx"]["hint"] == "tools/wikitool raw derive incoming/report.docx"
|
|
assert by_path["incoming/photo.jpg"]["hint"] is None
|
|
|
|
raw_pending_command(json_out=False)
|
|
assert "raw derive incoming/report.docx" in capsys.readouterr().out
|
|
|
|
|
|
def test_pending_gives_a_fetch_pair_no_hint(tree, capsys):
|
|
_put(tree, "page.html", b"<p>x</p>")
|
|
_put(tree, "page.md", b"x\n")
|
|
raw_pending_command(json_out=True)
|
|
[candidate] = json.loads(capsys.readouterr().out)
|
|
assert candidate["hint"] is None
|
|
|
|
|
|
def test_accept_warns_about_an_underived_file_but_still_promotes_it(tree, capsys):
|
|
path = _put(tree, "report.docx", make_docx(_p("Hello")))
|
|
raw_accept_command(
|
|
files=[path], fidelity="verbatim", authority="reporting",
|
|
page=None, replaces=None, dry_run=False,
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "WARN" in out and "raw derive incoming/report.docx" in out
|
|
assert [p for p in (tree / "raw").rglob("report.docx")]
|
|
|
|
|
|
def test_accept_is_quiet_when_a_derived_md_sits_beside_the_file(tree, capsys):
|
|
page = _put(tree, "page.html", b"<p>x</p>")
|
|
md = _put(tree, "page.md", b"x\n")
|
|
raw_accept_command(
|
|
files=[page, md], fidelity="published", authority="reporting",
|
|
page=None, replaces=None, dry_run=False,
|
|
)
|
|
assert "WARN" not in capsys.readouterr().out
|
|
|
|
|
|
def test_the_web_page_filter_derives_a_saved_html_file(tree):
|
|
_put(tree, "seite.html", b"<html><head><title>T</title></head><body><main><h1>Kopf</h1><p>Text</p></main></body></html>")
|
|
_derive("incoming/seite.html")
|
|
header, text = _header_and_text(tree / "incoming/seite/derived.md")
|
|
assert header["filter"] == "html" and header["title"] == "T"
|
|
assert header["charset"].startswith("utf-8")
|
|
assert "# Kopf" in text
|