208 lines
8.0 KiB
Python
208 lines
8.0 KiB
Python
"""Tests for the doocus document-extraction pipeline.
|
|
|
|
Covers registry dispatch and one assertion per extractor. Formats whose optional
|
|
dependency is absent are skipped (never failed), matching the graceful-degrade
|
|
design. Everything runs on synthetic fixtures — never real content.
|
|
"""
|
|
import importlib
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from tests import REPO_ROOT # noqa: F401 (puts repo root on sys.path)
|
|
from doocus import registry
|
|
from doocus.workflow import DocConfig, DocWorkflow
|
|
from doocus.tree import build_index, classify
|
|
|
|
|
|
def has(mod: str) -> bool:
|
|
try:
|
|
importlib.import_module(mod)
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
|
|
def run(source: Path, out: Path, **opts) -> dict:
|
|
cfg = DocConfig(source=str(source), output_dir=str(out), **opts)
|
|
return DocWorkflow(cfg).run()
|
|
|
|
|
|
def read_meta(result: dict) -> dict:
|
|
return json.loads(Path(result["meta"]).read_text())
|
|
|
|
|
|
class TestRegistry(unittest.TestCase):
|
|
def test_family_dispatch(self):
|
|
self.assertEqual(registry.family_for(Path("a.md")), "text")
|
|
self.assertEqual(registry.family_for(Path("a.CSV")), "tabular")
|
|
self.assertEqual(registry.family_for(Path("a.docx")), "office")
|
|
self.assertIsNone(registry.family_for(Path("a.xyz")))
|
|
|
|
def test_unhandled_is_not_fatal(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
src = Path(d) / "thing.xyz"
|
|
src.write_text("hi")
|
|
result = run(src, Path(d) / "out")
|
|
self.assertEqual(result["family"], "unhandled")
|
|
self.assertTrue(result["warnings"])
|
|
|
|
|
|
class TestExtractors(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.d = Path(self._tmp.name)
|
|
self.out = self.d / "out"
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def _write(self, name, text):
|
|
p = self.d / name
|
|
p.write_text(text, encoding="utf-8")
|
|
return p
|
|
|
|
def test_markdown(self):
|
|
src = self._write("notes.md", "# Title\n\nbody text\n")
|
|
meta = read_meta(run(src, self.out))
|
|
self.assertEqual(meta["title"], "Title")
|
|
self.assertEqual(meta["family"], "text")
|
|
|
|
def test_json_structure(self):
|
|
src = self._write("data.json", '{"a":1,"b":2}')
|
|
meta = read_meta(run(src, self.out))
|
|
self.assertEqual(meta["structure"]["key_count"], 2)
|
|
|
|
def test_csv_table(self):
|
|
src = self._write("t.csv", "a,b\n1,2\n3,4\n")
|
|
result = run(src, self.out)
|
|
meta = read_meta(result)
|
|
self.assertEqual(meta["rows"], 2)
|
|
self.assertEqual(meta["columns"], 2)
|
|
self.assertIn("| a | b |", Path(result["content"]).read_text())
|
|
|
|
@unittest.skipUnless(has("bs4"), "beautifulsoup4 not installed")
|
|
def test_html(self):
|
|
src = self._write("p.html", "<title>T</title><body><p>hello</p></body>")
|
|
result = run(src, self.out)
|
|
meta = read_meta(result)
|
|
self.assertEqual(meta["title"], "T")
|
|
self.assertIn("hello", Path(result["content"]).read_text())
|
|
|
|
@unittest.skipUnless(has("PIL"), "Pillow not installed")
|
|
def test_image_thumb(self):
|
|
from PIL import Image
|
|
src = self.d / "pic.png"
|
|
Image.new("RGB", (120, 80), "navy").save(src)
|
|
result = run(src, self.out)
|
|
meta = read_meta(result)
|
|
self.assertEqual((meta["width"], meta["height"]), (120, 80))
|
|
self.assertTrue((Path(result["output_dir"]) / "thumb.jpg").exists())
|
|
|
|
@unittest.skipUnless(has("docx"), "python-docx not installed")
|
|
def test_docx(self):
|
|
import docx
|
|
src = self.d / "r.docx"
|
|
doc = docx.Document()
|
|
doc.core_properties.author = "Ada"
|
|
doc.add_paragraph("hello world")
|
|
doc.save(src)
|
|
meta = read_meta(run(src, self.out))
|
|
self.assertEqual(meta["author"], "Ada")
|
|
self.assertEqual(meta["family"], "office")
|
|
|
|
@unittest.skipUnless(has("openpyxl"), "openpyxl not installed")
|
|
def test_xlsx(self):
|
|
from openpyxl import Workbook
|
|
src = self.d / "s.xlsx"
|
|
wb = Workbook()
|
|
wb.active.title = "Sales"
|
|
wb.active.append(["q", "rev"])
|
|
wb.save(src)
|
|
meta = read_meta(run(src, self.out))
|
|
self.assertEqual(meta["sheets"], ["Sales"])
|
|
|
|
@unittest.skipUnless(has("pypdf"), "pypdf not installed")
|
|
def test_pdf(self):
|
|
from pypdf import PdfWriter
|
|
src = self.d / "d.pdf"
|
|
w = PdfWriter()
|
|
w.add_blank_page(width=200, height=200)
|
|
w.add_metadata({"/Title": "PDF Doc"})
|
|
with open(src, "wb") as f:
|
|
w.write(f)
|
|
meta = read_meta(run(src, self.out))
|
|
self.assertEqual(meta["pages"], 1)
|
|
self.assertEqual(meta["title"], "PDF Doc")
|
|
|
|
|
|
class TestTreeIndex(unittest.TestCase):
|
|
def test_classify(self):
|
|
self.assertEqual(classify("docx"), "extracted")
|
|
self.assertEqual(classify("pdf"), "extracted")
|
|
self.assertEqual(classify("mp4"), "meeting")
|
|
self.assertEqual(classify("mkv"), "meeting")
|
|
self.assertEqual(classify("md"), "link")
|
|
self.assertEqual(classify("csv"), "link")
|
|
self.assertEqual(classify("bin"), "link")
|
|
|
|
def test_build_index_replicates_tree(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src = Path(tmp) / "drive"
|
|
(src / "team a" / "sub").mkdir(parents=True)
|
|
(src / "team a" / "notes.md").write_text("# Hi\n\nbody\n")
|
|
(src / "team a" / "sub" / "t.csv").write_text("a,b\n1,2\n")
|
|
(src / "team a" / "clip.mp4").write_bytes(b"\x00\x00") # video → meeting
|
|
(src / "misc.bin").write_bytes(b"\x00") # unknown → link
|
|
out = Path(tmp) / "out"
|
|
|
|
index = build_index(src, out)
|
|
|
|
# Whole tree replicated (no flattening), paths preserved.
|
|
paths = {f["path"] for f in index["files"]}
|
|
self.assertEqual(paths, {
|
|
"team a/notes.md", "team a/sub/t.csv", "team a/clip.mp4", "misc.bin"})
|
|
modes = {f["path"]: f["mode"] for f in index["files"]}
|
|
self.assertEqual(modes["team a/clip.mp4"], "meeting")
|
|
self.assertEqual(modes["team a/notes.md"], "link")
|
|
self.assertEqual(modes["misc.bin"], "link")
|
|
# Every node carries the Drive-url slot and a created date.
|
|
self.assertTrue(all("url" in f for f in index["files"]))
|
|
self.assertTrue(all(f.get("created") for f in index["files"]))
|
|
self.assertTrue((out / "index.json").exists())
|
|
|
|
def test_meeting_node_points_at_meetus_sidecar(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src = Path(tmp) / "drive"
|
|
(src / "team a").mkdir(parents=True)
|
|
(src / "team a" / "standup.mp4").write_bytes(b"\x00\x00")
|
|
index = build_index(src, Path(tmp) / "out")
|
|
node = next(f for f in index["files"] if f["path"] == "team a/standup.mp4")
|
|
# Deterministic pointer that `make batch ... OUT_FORMAT="{name}.meetus"` fills.
|
|
self.assertEqual(node["out"], "team a/standup.mp4.meetus")
|
|
self.assertEqual(node["transcript"], "team a/standup.mp4.meetus/standup_enhanced.txt")
|
|
|
|
@unittest.skipUnless(has("docx"), "python-docx not installed")
|
|
def test_build_index_extracts_sidecar(self):
|
|
import docx
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src = Path(tmp) / "drive"
|
|
src.mkdir()
|
|
doc = docx.Document()
|
|
doc.add_paragraph("hello world")
|
|
doc.save(src / "report.docx")
|
|
out = Path(tmp) / "out"
|
|
|
|
index = build_index(src, out)
|
|
node = next(f for f in index["files"] if f["path"] == "report.docx")
|
|
self.assertEqual(node["mode"], "extracted")
|
|
sidecar = out / node["out"]
|
|
self.assertTrue((sidecar / "content.md").exists())
|
|
self.assertTrue((sidecar / "meta.json").exists())
|
|
self.assertIn("hello", (sidecar / "content.md").read_text())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|