Spaces:
Running
Running
| """Unit tests for SQLite persistence and CSV export (build-plan task 4.1). | |
| All tests are fully offline: they use pytest's ``tmp_path`` fixture for | |
| isolated, throwaway databases and CSV files -- no real documents, no network. | |
| The ``ExtractionResult`` objects are built from the offline ``StubBackend`` | |
| so the Document shape is correct without touching the pipeline. | |
| Acceptance criteria (T7): | |
| - An accepted record persists and can be retrieved (record_count increases). | |
| - The same record inserted twice is a no-op (idempotency via content_hash). | |
| - export_csv writes a header + one data row per record. | |
| - A freshly-exported row contains the original field values. | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import json | |
| from datetime import date | |
| from pathlib import Path | |
| import pytest | |
| from doc_agent.backends.base import DocumentPayload | |
| from doc_agent.backends.stub import StubBackend | |
| from doc_agent.config import load_config | |
| from doc_agent.core import ExtractionResult, process_document | |
| from doc_agent.store.db import append_record, record_count | |
| from doc_agent.store.export import export_csv | |
| # Fixed timestamp injected so tests are deterministic. | |
| _TS = "2024-06-01T00:00:00+00:00" | |
| _HASH = "abc123def456" * 4 # 48-char fake SHA-256 | |
| def _offline_settings(): | |
| return load_config(extraction_backend="ollama", image_strategy="ocr_then_text") | |
| def _acquire_fixed(path: Path, modality): | |
| return DocumentPayload(modality=modality, source_path=path, text="stub") | |
| def _accepted_result(source: str = "invoice.pdf") -> ExtractionResult: | |
| """Build a clean, auto-accepted ExtractionResult via the stub pipeline.""" | |
| return process_document( | |
| source, | |
| settings=_offline_settings(), | |
| backend=StubBackend(), | |
| acquire=_acquire_fixed, | |
| today=date(2024, 6, 1), | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # append_record | |
| # --------------------------------------------------------------------------- | |
| def test_append_inserts_record(tmp_path: Path) -> None: | |
| """An accepted result is persisted; record_count increases by 1 (AC).""" | |
| db = tmp_path / "test.db" | |
| result = _accepted_result() | |
| inserted = append_record(result, db, _HASH, processed_at=_TS) | |
| assert inserted is True | |
| assert record_count(db) == 1 | |
| def test_append_idempotent_on_duplicate_hash(tmp_path: Path) -> None: | |
| """Inserting the same content_hash twice is a silent no-op (AC).""" | |
| db = tmp_path / "test.db" | |
| result = _accepted_result() | |
| first = append_record(result, db, _HASH, processed_at=_TS) | |
| second = append_record(result, db, _HASH, processed_at=_TS) | |
| assert first is True | |
| assert second is False | |
| assert record_count(db) == 1 | |
| def test_append_different_hashes_both_persist(tmp_path: Path) -> None: | |
| """Two records with distinct hashes both persist.""" | |
| db = tmp_path / "test.db" | |
| result = _accepted_result() | |
| hash_a = "a" * 64 | |
| hash_b = "b" * 64 | |
| append_record(result, db, hash_a, processed_at=_TS) | |
| append_record(result, db, hash_b, processed_at=_TS) | |
| assert record_count(db) == 2 | |
| def test_record_count_zero_when_db_absent(tmp_path: Path) -> None: | |
| """record_count returns 0 when the database file does not yet exist.""" | |
| assert record_count(tmp_path / "nonexistent.db") == 0 | |
| def test_append_creates_parent_dirs(tmp_path: Path) -> None: | |
| """append_record creates parent directories if they do not exist.""" | |
| db = tmp_path / "deep" / "nested" / "store.db" | |
| result = _accepted_result() | |
| append_record(result, db, _HASH, processed_at=_TS) | |
| assert db.exists() | |
| assert record_count(db) == 1 | |
| def test_persisted_fields_match_document(tmp_path: Path) -> None: | |
| """The persisted row reflects the document's extracted field values.""" | |
| import sqlite3 | |
| db = tmp_path / "test.db" | |
| result = _accepted_result() | |
| append_record(result, db, _HASH, processed_at=_TS) | |
| conn = sqlite3.connect(db) | |
| row = conn.execute( | |
| "SELECT vendor_name, total, invoice_number, decision, confidence, processed_at " | |
| "FROM documents WHERE content_hash = ?", | |
| (_HASH,), | |
| ).fetchone() | |
| conn.close() | |
| assert row is not None | |
| vendor, total, inv_num, decision, confidence, ts = row | |
| assert vendor == result.document.vendor_name | |
| assert total == pytest.approx(result.document.total) | |
| assert inv_num == result.document.invoice_number | |
| assert decision == "accept" | |
| assert 0.0 <= confidence <= 1.0 | |
| assert ts == _TS | |
| def test_line_items_stored_as_json(tmp_path: Path) -> None: | |
| """Line items are serialised as a JSON array in the database.""" | |
| import sqlite3 | |
| db = tmp_path / "test.db" | |
| result = _accepted_result() | |
| append_record(result, db, _HASH, processed_at=_TS) | |
| conn = sqlite3.connect(db) | |
| (raw,) = conn.execute("SELECT line_items FROM documents").fetchone() | |
| conn.close() | |
| items = json.loads(raw) | |
| assert isinstance(items, list) | |
| assert len(items) == len(result.document.line_items) | |
| # --------------------------------------------------------------------------- | |
| # export_csv | |
| # --------------------------------------------------------------------------- | |
| def test_export_csv_writes_header_and_row(tmp_path: Path) -> None: | |
| """export_csv produces a CSV with a header and one data row (AC).""" | |
| db = tmp_path / "test.db" | |
| csv_path = tmp_path / "out.csv" | |
| result = _accepted_result() | |
| append_record(result, db, _HASH, processed_at=_TS) | |
| count = export_csv(db, csv_path) | |
| assert count == 1 | |
| assert csv_path.exists() | |
| rows = list(csv.DictReader(csv_path.open(encoding="utf-8"))) | |
| assert len(rows) == 1 | |
| def test_export_csv_row_values_match_record(tmp_path: Path) -> None: | |
| """Exported CSV row values match the persisted document fields (AC).""" | |
| db = tmp_path / "test.db" | |
| csv_path = tmp_path / "out.csv" | |
| result = _accepted_result() | |
| append_record(result, db, _HASH, processed_at=_TS) | |
| export_csv(db, csv_path) | |
| rows = list(csv.DictReader(csv_path.open(encoding="utf-8"))) | |
| row = rows[0] | |
| assert row["vendor_name"] == (result.document.vendor_name or "") | |
| assert float(row["total"]) == pytest.approx(result.document.total) | |
| assert row["decision"] == "accept" | |
| assert row["content_hash"] == _HASH | |
| assert int(row["line_item_count"]) == len(result.document.line_items) | |
| def test_export_csv_empty_db_writes_header_only(tmp_path: Path) -> None: | |
| """export_csv on an absent DB writes only the header row.""" | |
| csv_path = tmp_path / "empty.csv" | |
| count = export_csv(tmp_path / "nonexistent.db", csv_path) | |
| assert count == 0 | |
| rows = list(csv.DictReader(csv_path.open(encoding="utf-8"))) | |
| assert rows == [] | |
| def test_export_csv_multiple_records(tmp_path: Path) -> None: | |
| """All records appear in the exported CSV, one row each.""" | |
| db = tmp_path / "test.db" | |
| csv_path = tmp_path / "out.csv" | |
| result = _accepted_result() | |
| for i in range(3): | |
| append_record(result, db, f"hash{i}" + "0" * 61, processed_at=_TS) | |
| count = export_csv(db, csv_path) | |
| assert count == 3 | |
| rows = list(csv.DictReader(csv_path.open(encoding="utf-8"))) | |
| assert len(rows) == 3 | |
| def test_export_csv_creates_parent_dirs(tmp_path: Path) -> None: | |
| """export_csv creates parent directories of the output path.""" | |
| db = tmp_path / "test.db" | |
| csv_path = tmp_path / "exports" / "sub" / "out.csv" | |
| result = _accepted_result() | |
| append_record(result, db, _HASH, processed_at=_TS) | |
| export_csv(db, csv_path) | |
| assert csv_path.exists() | |