T

"""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", "T

hello

") 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. self.assertTrue(all("url" in f for f in index["files"])) self.assertTrue((out / "index.json").exists()) @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()