"""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
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", "Thello
")
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")
if __name__ == "__main__":
unittest.main()