pagepiper/tests/test_shelve_odt.py

# tests/test_shelve_odt.py
"""Unit tests for scripts/shelve_odt.py."""
from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest

from scripts.shelve_odt 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.odt','pending')"
    )
    conn.commit()
    conn.close()
    vec_db_path = str(tmp_path / "vecs.db")
    return db_path, vec_db_path


def _make_odt(path: Path, with_headings: bool = True, with_table: bool = False) -> None:
    from odf.opendocument import OpenDocumentText
    from odf.text import H, P
    from odf.table import Table, TableRow, TableCell

    doc = OpenDocumentText()
    if with_headings:
        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."))
    else:
        doc.text.addElement(P(text="Some unstructured procedure text with no headings at all."))
    if with_table:
        table = Table(name="T1")
        for rowvals in [["Field", "Value"], ["IP Address", "10.0.0.5"]]:
            row = TableRow()
            for v in rowvals:
                cell = TableCell()
                cell.addElement(P(text=v))
                row.addElement(cell)
            table.addElement(row)
        doc.text.addElement(table)
    doc.save(str(path))


def test_shelve_odt_sets_status_ready_on_success(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    odt_path = tmp_path / "test.odt"
    _make_odt(odt_path)

    run(doc_id="d1", file_path=str(odt_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()
    conn.close()
    assert row[0] == "ready"
    assert row[1] == 2  # one chunk per heading section


def test_shelve_odt_splits_by_heading(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    odt_path = tmp_path / "test.odt"
    _make_odt(odt_path)

    run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path)

    conn = sqlite3.connect(db_path)
    rows = conn.execute(
        "SELECT text FROM page_chunks WHERE doc_id='d1' ORDER BY page_number"
    ).fetchall()
    conn.close()
    assert len(rows) == 2
    assert "Setting the IP" in rows[0][0]
    assert "Verifying the Change" in rows[1][0]


def test_shelve_odt_falls_back_to_wordcount_chunks_without_headings(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    odt_path = tmp_path / "test.odt"
    _make_odt(odt_path, with_headings=False)

    run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path)

    conn = sqlite3.connect(db_path)
    rows = conn.execute("SELECT text FROM page_chunks WHERE doc_id='d1'").fetchall()
    conn.close()
    assert len(rows) == 1
    assert "unstructured procedure text" in rows[0][0]


def test_shelve_odt_serializes_tables_inline(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    odt_path = tmp_path / "test.odt"
    _make_odt(odt_path, with_table=True)

    run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path)

    conn = sqlite3.connect(db_path)
    rows = conn.execute("SELECT text FROM page_chunks WHERE doc_id='d1'").fetchall()
    conn.close()
    joined = "\n".join(r[0] for r in rows)
    assert "IP Address | 10.0.0.5" in joined


def test_shelve_odt_sets_error_status_on_failure(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    missing_path = tmp_path / "does-not-exist.odt"

    with pytest.raises(Exception):
        run(doc_id="d1", file_path=str(missing_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 row[1]


def test_shelve_odt_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch):
    db_path, vec_db_path = shelve_db
    monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False)
    odt_path = tmp_path / "test.odt"
    _make_odt(odt_path)

    run(doc_id="d1", file_path=str(odt_path), db_path=db_path, vec_db_path=vec_db_path)

    assert not Path(vec_db_path).exists(), "vec DB should not be created without OLLAMA_URL"