141 lines
4.7 KiB
Python
141 lines
4.7 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
|
|
|
|
|
|
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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|