pyr0ball
f941ebdeeb
Extends Pagepiper's document shelving pipeline (renamed from "ingest" — see below) to cover the formats most likely to appear in a real-world engineering document corpus, prompted by scoping a STERIS licensing pitch that needs DOCX/ODT coverage. - Rename the ingest pipeline to "shelve" throughout (scripts/, app/api, tests, docs, frontend). "Glean" (Turnstone's term) was considered and rejected — that's a harvest metaphor for log/knowledge extraction, not a fit for documents entering a library. Documented as a general CF naming principle in the org-level CLAUDE.md. - Wire DOCX into the upload/scan UI, README, and docs — the extraction logic (heading-based chunking, table serialization) already existed but wasn't exposed to users or covered by tests. - Add ODT support via odfpy, mirroring DOCX's chunking strategy. - Add Apple Pages support via headless LibreOffice conversion to ODT. No maintained Python library parses the IWA format directly; libreoffice bundles libetonyek, the only real open-source Pages parser. Adds libreoffice-writer to the Docker image (~300-400MB) for this. - 24 new/updated tests across shelve_docx, shelve_odt, and shelve_pages; full suite (72 tests) passing. Known gaps not addressed here: no Windchill/DocPortal connector exists yet (metadata-only PowerShell recon only), Excel/.xlsx is unsupported, and circuitforge_core.tasks.dispatch_task does not currently exist in circuitforge-core — cf-orch dispatch is dead code, always falling through to local BackgroundTasks. See circuitforge-plans/pagepiper/superpowers/plans/2026-07-10-steris-licensing-pitch.md for the full writeup.
122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
# tests/test_shelve_pages.py
|
|
"""Unit tests for scripts/shelve_pages.py.
|
|
|
|
soffice isn't available in the test environment, so _convert_to_odt is
|
|
mocked to copy a pre-built ODT fixture into the expected output path —
|
|
everything downstream (extraction/chunking/storage) runs for real via
|
|
scripts.shelve_odt.run.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from scripts.shelve_pages import run
|
|
|
|
|
|
@pytest.fixture
|
|
def shelve_db(tmp_path) -> tuple[str, str]:
|
|
db_path = str(tmp_path / "test.db")
|
|
schema = Path("migrations/001_initial_schema.sql").read_text()
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(schema)
|
|
conn.execute(
|
|
"INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Test','test.pages','pending')"
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
vec_db_path = str(tmp_path / "vecs.db")
|
|
return db_path, vec_db_path
|
|
|
|
|
|
def _make_fixture_odt(path: Path) -> None:
|
|
from odf.opendocument import OpenDocumentText
|
|
from odf.text import H, P
|
|
|
|
doc = OpenDocumentText()
|
|
doc.text.addElement(H(outlinelevel=1, text="Setting the IP"))
|
|
doc.text.addElement(P(text="Connect to the device over the service port."))
|
|
doc.text.addElement(H(outlinelevel=1, text="Verifying the Change"))
|
|
doc.text.addElement(P(text="Ping the new address to confirm."))
|
|
doc.save(str(path))
|
|
|
|
|
|
def _fake_soffice_convert(fixture_odt: Path):
|
|
"""Return a fake subprocess.run that copies fixture_odt into --outdir."""
|
|
|
|
def _fake_run(cmd, **kwargs):
|
|
out_dir = Path(cmd[cmd.index("--outdir") + 1])
|
|
pages_path = Path(cmd[-1])
|
|
shutil.copy(fixture_odt, out_dir / f"{pages_path.stem}.odt")
|
|
return subprocess.CompletedProcess(cmd, returncode=0, stdout="", stderr="")
|
|
|
|
return _fake_run
|
|
|
|
|
|
def test_shelve_pages_converts_and_shelves_successfully(shelve_db, tmp_path):
|
|
db_path, vec_db_path = shelve_db
|
|
pages_path = tmp_path / "test.pages"
|
|
pages_path.write_bytes(b"not a real pages bundle, just needs to exist")
|
|
|
|
fixture_odt = tmp_path / "fixture.odt"
|
|
_make_fixture_odt(fixture_odt)
|
|
|
|
with patch("scripts.shelve_pages.subprocess.run", side_effect=_fake_soffice_convert(fixture_odt)):
|
|
run(doc_id="d1", file_path=str(pages_path), db_path=db_path, vec_db_path=vec_db_path)
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
row = conn.execute("SELECT status, page_count FROM documents WHERE id='d1'").fetchone()
|
|
rows = conn.execute(
|
|
"SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number"
|
|
).fetchall()
|
|
conn.close()
|
|
assert row[0] == "ready"
|
|
assert row[1] == 2
|
|
assert "Setting the IP" in rows[0][0]
|
|
assert "Verifying the Change" in rows[1][0]
|
|
|
|
|
|
def test_shelve_pages_sets_error_status_on_conversion_failure(shelve_db, tmp_path):
|
|
db_path, vec_db_path = shelve_db
|
|
pages_path = tmp_path / "test.pages"
|
|
pages_path.write_bytes(b"not a real pages bundle")
|
|
|
|
fake_result = subprocess.CompletedProcess(
|
|
args=[], returncode=1, stdout="", stderr="soffice: unsupported document format"
|
|
)
|
|
with patch("scripts.shelve_pages.subprocess.run", return_value=fake_result):
|
|
with pytest.raises(RuntimeError, match="LibreOffice conversion failed"):
|
|
run(doc_id="d1", file_path=str(pages_path), db_path=db_path, vec_db_path=vec_db_path)
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
row = conn.execute("SELECT status, error_msg FROM documents WHERE id='d1'").fetchone()
|
|
conn.close()
|
|
assert row[0] == "error"
|
|
assert "unsupported document format" in row[1]
|
|
|
|
|
|
def test_shelve_pages_cleans_up_temp_dir(shelve_db, tmp_path):
|
|
db_path, vec_db_path = shelve_db
|
|
pages_path = tmp_path / "test.pages"
|
|
pages_path.write_bytes(b"not a real pages bundle")
|
|
|
|
fixture_odt = tmp_path / "fixture.odt"
|
|
_make_fixture_odt(fixture_odt)
|
|
|
|
captured_tmp_dirs: list[str] = []
|
|
real_convert = _fake_soffice_convert(fixture_odt)
|
|
|
|
def _tracking_run(cmd, **kwargs):
|
|
captured_tmp_dirs.append(cmd[cmd.index("--outdir") + 1])
|
|
return real_convert(cmd, **kwargs)
|
|
|
|
with patch("scripts.shelve_pages.subprocess.run", side_effect=_tracking_run):
|
|
run(doc_id="d1", file_path=str(pages_path), db_path=db_path, vec_db_path=vec_db_path)
|
|
|
|
assert captured_tmp_dirs
|
|
assert not Path(captured_tmp_dirs[0]).exists(), "temp conversion dir should be cleaned up"
|