pagepiper/tests/test_shelve_xlsx.py

# tests/test_shelve_xlsx.py
"""Unit tests for scripts/shelve_xlsx.py."""
from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest

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


def _make_xlsx(path: Path, big_sheet: bool = False) -> None:
    import openpyxl

    wb = openpyxl.Workbook()
    ws1 = wb.active
    ws1.title = "Config"
    ws1.append(["Field", "Value"])
    ws1.append(["Default IP", "192.168.1.50"])
    ws1.append(["AVC-X Model", "AVC-X200"])

    ws2 = wb.create_sheet("Parts List")
    ws2.append(["Part Number", "Description"])
    if big_sheet:
        for i in range(ROWS_PER_CHUNK + 50):
            ws2.append([f"P{i:04d}", f"Widget {i}"])
    else:
        ws2.append(["P0001", "Filter cartridge"])

    wb.save(str(path))


def test_shelve_xlsx_sets_status_ready_on_success(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    xlsx_path = tmp_path / "test.xlsx"
    _make_xlsx(xlsx_path)

    run(doc_id="d1", file_path=str(xlsx_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 sheet, both small


def test_shelve_xlsx_serializes_rows_with_header(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    xlsx_path = tmp_path / "test.xlsx"
    _make_xlsx(xlsx_path)

    run(doc_id="d1", file_path=str(xlsx_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_xlsx_splits_large_sheet_into_row_windows(shelve_db, tmp_path):
    db_path, vec_db_path = shelve_db
    xlsx_path = tmp_path / "test.xlsx"
    _make_xlsx(xlsx_path, big_sheet=True)

    run(doc_id="d1", file_path=str(xlsx_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()
    # Config sheet (1 chunk) + Parts List sheet split across >=2 row-window chunks
    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
    # header row repeated in every window
    for chunk_text in parts_list_chunks:
        assert "Part Number | Description" in chunk_text


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

    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_xlsx_skips_embeddings_without_ollama_url(shelve_db, tmp_path, monkeypatch):
    db_path, vec_db_path = shelve_db
    monkeypatch.delenv("PAGEPIPER_OLLAMA_URL", raising=False)
    xlsx_path = tmp_path / "test.xlsx"
    _make_xlsx(xlsx_path)

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