pagepiper/tests/test_shelve_ods.py

# tests/test_shelve_ods.py
"""Unit tests for scripts/shelve_ods.py."""
from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest

from scripts.shelve_ods import run, ROWS_PER_CHUNK


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


def _make_ods(path: Path, big_sheet: bool = False) -> None:
    from odf.opendocument import OpenDocumentSpreadsheet
    from odf.table import Table, TableRow, TableCell
    from odf.text import P

    def _row(vals):
        row = TableRow()
        for v in vals:
            cell = TableCell()
            cell.addElement(P(text=v))
            row.addElement(cell)
        return row

    doc = OpenDocumentSpreadsheet()

    t1 = Table(name="Config")
    t1.addElement(_row(["Field", "Value"]))
    t1.addElement(_row(["Default IP", "192.168.1.50"]))
    t1.addElement(_row(["AVC-X Model", "AVC-X200"]))
    doc.spreadsheet.addElement(t1)

    t2 = Table(name="Parts List")
    t2.addElement(_row(["Part Number", "Description"]))
    if big_sheet:
        for i in range(ROWS_PER_CHUNK + 50):
            t2.addElement(_row([f"P{i:04d}", f"Widget {i}"]))
    else:
        t2.addElement(_row(["P0001", "Filter cartridge"]))
    doc.spreadsheet.addElement(t2)

    doc.save(str(path))


def test_shelve_ods_sets_status_ready_on_success(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    ods_path = tmp_path / "test.ods"
    _make_ods(ods_path)

    run(doc_id="d1", file_path=str(ods_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


def test_shelve_ods_serializes_rows_with_header(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    ods_path = tmp_path / "test.ods"
    _make_ods(ods_path)

    run(doc_id="d1", file_path=str(ods_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 "Sheet: Config" in rows[0][0]
    assert "Field | Value" in rows[0][0]
    assert "Default IP | 192.168.1.50" in rows[0][0]
    assert "Sheet: Parts List" in rows[1][0]
    assert "P0001 | Filter cartridge" in rows[1][0]


def test_shelve_ods_splits_large_sheet_into_row_windows(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    ods_path = tmp_path / "test.ods"
    _make_ods(ods_path, big_sheet=True)

    run(doc_id="d1", file_path=str(ods_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) >= 3
    parts_list_chunks = [r[0] for r in rows if "Sheet: Parts List" in r[0]]
    assert len(parts_list_chunks) >= 2
    for chunk_text in parts_list_chunks:
        assert "Part Number | Description" in chunk_text


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

    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_ods_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch):
    db_path, vec_db_path = shelve_db
    monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False)
    ods_path = tmp_path / "test.ods"
    _make_ods(ods_path)

    run(doc_id="d1", file_path=str(ods_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"