Files
bookshelf/tests/test_logic.py
Petr Polezhaev 084d1aebd5 Initial commit
Photo-based book cataloger with AI identification.
Room → Cabinet → Shelf → Book hierarchy; FastAPI + SQLite backend;
vanilla JS SPA; OpenAI-compatible plugin system for boundary
detection, text recognition, and archive search.
2026-03-09 14:17:13 +03:00

586 lines
20 KiB
Python

"""Unit tests for logic modules: boundary helpers, identification helpers, build_query, and all error conditions."""
import asyncio
from pathlib import Path
import pytest
import db as db_module
import logic
from errors import (
BookNotFoundError,
CabinetNotFoundError,
InvalidPluginEntityError,
NoCabinetPhotoError,
NoRawTextError,
NoShelfImageError,
PluginNotFoundError,
PluginTargetMismatchError,
ShelfNotFoundError,
)
from logic.archive import run_archive_searcher
from logic.boundaries import book_spine_source, bounds_for_index, run_boundary_detector, shelf_source
from logic.identification import apply_ai_result, build_query, compute_status, dismiss_field, run_book_identifier
from models import (
AIIdentifyResult,
BoundaryDetectResult,
BookRow,
CandidateRecord,
PluginLookupResult,
TextRecognizeResult,
)
# ── BookRow factory ───────────────────────────────────────────────────────────
def _book(**kwargs: object) -> BookRow:
defaults: dict[str, object] = {
"id": "b1",
"shelf_id": "s1",
"position": 0,
"image_filename": None,
"title": "",
"author": "",
"year": "",
"isbn": "",
"publisher": "",
"notes": "",
"raw_text": "",
"ai_title": "",
"ai_author": "",
"ai_year": "",
"ai_isbn": "",
"ai_publisher": "",
"identification_status": "unidentified",
"title_confidence": 0.0,
"analyzed_at": None,
"created_at": "2024-01-01T00:00:00",
"candidates": None,
}
defaults.update(kwargs)
return BookRow(**defaults) # type: ignore[arg-type]
# ── DB fixture for integration tests ─────────────────────────────────────────
@pytest.fixture
def seeded_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Temporary DB with a single book row (full parent chain)."""
monkeypatch.setattr(db_module, "DB_PATH", tmp_path / "test.db")
db_module.init_db()
ts = "2024-01-01T00:00:00"
c = db_module.conn()
c.execute("INSERT INTO rooms VALUES (?,?,?,?)", ["r1", "Room", 1, ts])
c.execute("INSERT INTO cabinets VALUES (?,?,?,?,?,?,?,?)", ["c1", "r1", "Cabinet", None, None, None, 1, ts])
c.execute("INSERT INTO shelves VALUES (?,?,?,?,?,?,?,?)", ["s1", "c1", "Shelf", None, None, None, 1, ts])
c.execute(
"INSERT INTO books VALUES (?,?,0,NULL,'','','','','','','','','','','','','unidentified',0,NULL,?,NULL)",
["b1", "s1", ts],
)
c.commit()
c.close()
# ── Stub plugins ──────────────────────────────────────────────────────────────
class _BoundaryDetectorStub:
"""Stub boundary detector that returns empty boundaries."""
plugin_id = "bd_stub"
name = "Stub BD"
auto_queue = False
target = "books"
@property
def max_image_px(self) -> int:
return 1600
def detect(self, image_b64: str, image_mime: str) -> BoundaryDetectResult:
return {"boundaries": [0.5]}
class _BoundaryDetectorShelvesStub:
"""Stub boundary detector targeting shelves (for cabinet entity_type)."""
plugin_id = "bd_shelves_stub"
name = "Stub BD Shelves"
auto_queue = False
target = "shelves"
@property
def max_image_px(self) -> int:
return 1600
def detect(self, image_b64: str, image_mime: str) -> BoundaryDetectResult:
return {"boundaries": []}
class _TextRecognizerStub:
"""Stub text recognizer that returns fixed text."""
plugin_id = "tr_stub"
name = "Stub TR"
auto_queue = False
@property
def max_image_px(self) -> int:
return 1600
def recognize(self, image_b64: str, image_mime: str) -> TextRecognizeResult:
return {"raw_text": "Stub Title", "title": "Stub Title", "author": "Stub Author"}
class _BookIdentifierStub:
"""Stub book identifier that returns a high-confidence result."""
plugin_id = "bi_stub"
name = "Stub BI"
auto_queue = False
@property
def confidence_threshold(self) -> float:
return 0.8
def identify(self, raw_text: str) -> AIIdentifyResult:
return {
"title": "Found Book",
"author": "Found Author",
"year": "2000",
"isbn": "",
"publisher": "",
"confidence": 0.9,
}
class _ArchiveSearcherStub:
"""Stub archive searcher that returns an empty result list."""
plugin_id = "as_stub"
name = "Stub AS"
auto_queue = False
def search(self, query: str) -> list[CandidateRecord]:
return []
# ── bounds_for_index ──────────────────────────────────────────────────────────
def test_bounds_empty_boundaries() -> None:
assert bounds_for_index(None, 0) == (0.0, 1.0)
def test_bounds_empty_json() -> None:
assert bounds_for_index("[]", 0) == (0.0, 1.0)
def test_bounds_single_boundary_first() -> None:
assert bounds_for_index("[0.5]", 0) == (0.0, 0.5)
def test_bounds_single_boundary_second() -> None:
assert bounds_for_index("[0.5]", 1) == (0.5, 1.0)
def test_bounds_multiple_boundaries() -> None:
b = "[0.25, 0.5, 0.75]"
assert bounds_for_index(b, 0) == (0.0, 0.25)
assert bounds_for_index(b, 1) == (0.25, 0.5)
assert bounds_for_index(b, 2) == (0.5, 0.75)
assert bounds_for_index(b, 3) == (0.75, 1.0)
def test_bounds_out_of_range_returns_last_segment() -> None:
_, end = bounds_for_index("[0.5]", 99)
assert end == 1.0
# ── compute_status ────────────────────────────────────────────────────────────
def test_compute_status_unidentified_no_ai_title() -> None:
assert compute_status(_book(ai_title="", title="", author="", year="")) == "unidentified"
def test_compute_status_unidentified_empty() -> None:
assert compute_status(_book()) == "unidentified"
def test_compute_status_ai_identified() -> None:
book = _book(ai_title="Some Book", ai_author="Author", ai_year="2000", ai_isbn="", ai_publisher="")
assert compute_status(book) == "ai_identified"
def test_compute_status_user_approved() -> None:
book = _book(
ai_title="Some Book",
ai_author="Author",
ai_year="2000",
ai_isbn="",
ai_publisher="",
title="Some Book",
author="Author",
year="2000",
isbn="",
publisher="",
)
assert compute_status(book) == "user_approved"
def test_compute_status_ai_identified_when_fields_differ() -> None:
book = _book(
ai_title="Some Book",
ai_author="Original Author",
ai_year="2000",
title="Some Book",
author="Different Author",
year="2000",
)
assert compute_status(book) == "ai_identified"
# ── build_query ───────────────────────────────────────────────────────────────
def test_build_query_from_candidates() -> None:
book = _book(candidates='[{"source": "x", "author": "Tolkien", "title": "LOTR"}]')
assert build_query(book) == "Tolkien LOTR"
def test_build_query_from_ai_fields() -> None:
book = _book(candidates="[]", ai_author="Pushkin", ai_title="Evgeny Onegin", raw_text="")
assert build_query(book) == "Pushkin Evgeny Onegin"
def test_build_query_from_raw_text() -> None:
book = _book(candidates="[]", ai_author="", ai_title="", raw_text="some spine text")
assert build_query(book) == "some spine text"
def test_build_query_empty() -> None:
book = _book(candidates="[]", ai_author="", ai_title="", raw_text="")
assert build_query(book) == ""
def test_build_query_candidates_prefer_first_nonempty() -> None:
book = _book(
candidates='[{"source":"a","author":"","title":""}, {"source":"b","author":"Auth","title":"Title"}]',
ai_author="other",
ai_title="other",
)
assert build_query(book) == "Auth Title"
# ── apply_ai_result ───────────────────────────────────────────────────────────
def test_apply_ai_result_high_confidence(seeded_db: None) -> None:
result: AIIdentifyResult = {
"title": "My Book",
"author": "J. Doe",
"year": "1999",
"isbn": "123",
"publisher": "Pub",
"confidence": 0.9,
}
apply_ai_result("b1", result, confidence_threshold=0.8)
with db_module.connection() as c:
book = db_module.get_book(c, "b1")
assert book is not None
assert book.ai_title == "My Book"
assert book.ai_author == "J. Doe"
assert abs(book.title_confidence - 0.9) < 1e-9
assert book.identification_status == "ai_identified"
def test_apply_ai_result_low_confidence_skips_fields(seeded_db: None) -> None:
result: AIIdentifyResult = {
"title": "My Book",
"author": "J. Doe",
"year": "1999",
"isbn": "",
"publisher": "",
"confidence": 0.5,
}
apply_ai_result("b1", result, confidence_threshold=0.8)
with db_module.connection() as c:
book = db_module.get_book(c, "b1")
assert book is not None
assert book.ai_title == "" # not updated
assert abs(book.title_confidence - 0.5) < 1e-9 # confidence stored regardless
assert book.identification_status == "unidentified"
def test_apply_ai_result_exact_threshold(seeded_db: None) -> None:
result: AIIdentifyResult = {
"title": "Book",
"author": "",
"year": "",
"isbn": "",
"publisher": "",
"confidence": 0.8,
}
apply_ai_result("b1", result, confidence_threshold=0.8)
with db_module.connection() as c:
book = db_module.get_book(c, "b1")
assert book is not None
assert book.ai_title == "Book"
# ── shelf_source error conditions ─────────────────────────────────────────────
def test_shelf_source_not_found(seeded_db: None) -> None:
with db_module.connection() as c:
with pytest.raises(ShelfNotFoundError) as exc_info:
shelf_source(c, "nonexistent")
assert exc_info.value.shelf_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
def test_shelf_source_no_image(seeded_db: None) -> None:
# s1 has no photo_filename and c1 has no photo_filename → NoShelfImageError
with db_module.connection() as c:
with pytest.raises(NoShelfImageError) as exc_info:
shelf_source(c, "s1")
assert exc_info.value.shelf_id == "s1"
assert exc_info.value.cabinet_id == "c1"
assert "s1" in str(exc_info.value)
assert "c1" in str(exc_info.value)
# ── book_spine_source error conditions ────────────────────────────────────────
def test_book_spine_source_book_not_found(seeded_db: None) -> None:
with db_module.connection() as c:
with pytest.raises(BookNotFoundError) as exc_info:
book_spine_source(c, "nonexistent")
assert exc_info.value.book_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
def test_book_spine_source_propagates_no_shelf_image(seeded_db: None) -> None:
# b1 exists but s1 has no image → NoShelfImageError propagates through book_spine_source
with db_module.connection() as c:
with pytest.raises(NoShelfImageError) as exc_info:
book_spine_source(c, "b1")
assert exc_info.value.shelf_id == "s1"
assert exc_info.value.cabinet_id == "c1"
# ── run_boundary_detector error conditions ────────────────────────────────────
def test_run_boundary_detector_cabinet_not_found(seeded_db: None) -> None:
plugin = _BoundaryDetectorShelvesStub()
with pytest.raises(CabinetNotFoundError) as exc_info:
run_boundary_detector(plugin, "cabinets", "nonexistent")
assert exc_info.value.cabinet_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
def test_run_boundary_detector_no_cabinet_photo(seeded_db: None) -> None:
# c1 exists but has no photo_filename
plugin = _BoundaryDetectorShelvesStub()
with pytest.raises(NoCabinetPhotoError) as exc_info:
run_boundary_detector(plugin, "cabinets", "c1")
assert exc_info.value.cabinet_id == "c1"
assert "c1" in str(exc_info.value)
def test_run_boundary_detector_shelf_not_found(seeded_db: None) -> None:
plugin = _BoundaryDetectorStub()
with pytest.raises(ShelfNotFoundError) as exc_info:
run_boundary_detector(plugin, "shelves", "nonexistent")
assert exc_info.value.shelf_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
def test_run_boundary_detector_shelf_no_image(seeded_db: None) -> None:
# s1 exists but has no image (neither override nor cabinet photo)
plugin = _BoundaryDetectorStub()
with pytest.raises(NoShelfImageError) as exc_info:
run_boundary_detector(plugin, "shelves", "s1")
assert exc_info.value.shelf_id == "s1"
assert exc_info.value.cabinet_id == "c1"
# ── run_book_identifier error conditions ──────────────────────────────────────
def test_run_book_identifier_not_found(seeded_db: None) -> None:
plugin = _BookIdentifierStub()
with pytest.raises(BookNotFoundError) as exc_info:
run_book_identifier(plugin, "nonexistent")
assert exc_info.value.book_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
def test_run_book_identifier_no_raw_text(seeded_db: None) -> None:
# b1 has raw_text='' (default)
plugin = _BookIdentifierStub()
with pytest.raises(NoRawTextError) as exc_info:
run_book_identifier(plugin, "b1")
assert exc_info.value.book_id == "b1"
assert "b1" in str(exc_info.value)
# ── run_archive_searcher error conditions ─────────────────────────────────────
def test_run_archive_searcher_not_found(seeded_db: None) -> None:
plugin = _ArchiveSearcherStub()
with pytest.raises(BookNotFoundError) as exc_info:
run_archive_searcher(plugin, "nonexistent")
assert exc_info.value.book_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
# ── dismiss_field error conditions ────────────────────────────────────────────
def test_dismiss_field_not_found(seeded_db: None) -> None:
with pytest.raises(BookNotFoundError) as exc_info:
dismiss_field("nonexistent", "title", "some value")
assert exc_info.value.book_id == "nonexistent"
assert "nonexistent" in str(exc_info.value)
# ── dispatch_plugin error conditions ──────────────────────────────────────────
def _run_dispatch(plugin_id: str, lookup: PluginLookupResult, entity_type: str, entity_id: str) -> None:
"""Helper to synchronously drive the async dispatch_plugin."""
async def _inner() -> None:
loop = asyncio.get_event_loop()
await logic.dispatch_plugin(plugin_id, lookup, entity_type, entity_id, loop)
asyncio.run(_inner())
def test_dispatch_plugin_not_found() -> None:
with pytest.raises(PluginNotFoundError) as exc_info:
_run_dispatch("no_such_plugin", (None, None), "books", "b1")
assert exc_info.value.plugin_id == "no_such_plugin"
assert "no_such_plugin" in str(exc_info.value)
def test_dispatch_plugin_boundary_wrong_entity_type() -> None:
lookup = ("boundary_detector", _BoundaryDetectorStub())
with pytest.raises(InvalidPluginEntityError) as exc_info:
_run_dispatch("bd_stub", lookup, "books", "b1")
assert exc_info.value.plugin_category == "boundary_detector"
assert exc_info.value.entity_type == "books"
assert "boundary_detector" in str(exc_info.value)
assert "books" in str(exc_info.value)
def test_dispatch_plugin_target_mismatch_cabinets(seeded_db: None) -> None:
# Plugin targets "books" but entity_type is "cabinets" (expects target="shelves")
plugin = _BoundaryDetectorStub() # target = "books"
lookup = ("boundary_detector", plugin)
with pytest.raises(PluginTargetMismatchError) as exc_info:
_run_dispatch("bd_stub", lookup, "cabinets", "c1")
assert exc_info.value.plugin_id == "bd_stub"
assert exc_info.value.expected_target == "shelves"
assert exc_info.value.actual_target == "books"
assert "bd_stub" in str(exc_info.value)
def test_dispatch_plugin_target_mismatch_shelves(seeded_db: None) -> None:
# Plugin targets "shelves" but entity_type is "shelves" (expects target="books")
plugin = _BoundaryDetectorShelvesStub() # target = "shelves"
lookup = ("boundary_detector", plugin)
with pytest.raises(PluginTargetMismatchError) as exc_info:
_run_dispatch("bd_shelves_stub", lookup, "shelves", "s1")
assert exc_info.value.plugin_id == "bd_shelves_stub"
assert exc_info.value.expected_target == "books"
assert exc_info.value.actual_target == "shelves"
def test_dispatch_plugin_text_recognizer_wrong_entity_type() -> None:
lookup = ("text_recognizer", _TextRecognizerStub())
with pytest.raises(InvalidPluginEntityError) as exc_info:
_run_dispatch("tr_stub", lookup, "cabinets", "c1")
assert exc_info.value.plugin_category == "text_recognizer"
assert exc_info.value.entity_type == "cabinets"
def test_dispatch_plugin_book_identifier_wrong_entity_type() -> None:
lookup = ("book_identifier", _BookIdentifierStub())
with pytest.raises(InvalidPluginEntityError) as exc_info:
_run_dispatch("bi_stub", lookup, "shelves", "s1")
assert exc_info.value.plugin_category == "book_identifier"
assert exc_info.value.entity_type == "shelves"
def test_dispatch_plugin_archive_searcher_wrong_entity_type() -> None:
lookup = ("archive_searcher", _ArchiveSearcherStub())
with pytest.raises(InvalidPluginEntityError) as exc_info:
_run_dispatch("as_stub", lookup, "cabinets", "c1")
assert exc_info.value.plugin_category == "archive_searcher"
assert exc_info.value.entity_type == "cabinets"
# ── Exception string representation ───────────────────────────────────────────
def test_exception_str_cabinet_not_found() -> None:
exc = CabinetNotFoundError("cab-123")
assert exc.cabinet_id == "cab-123"
assert "cab-123" in str(exc)
def test_exception_str_shelf_not_found() -> None:
exc = ShelfNotFoundError("shelf-456")
assert exc.shelf_id == "shelf-456"
assert "shelf-456" in str(exc)
def test_exception_str_plugin_not_found() -> None:
exc = PluginNotFoundError("myplugin")
assert exc.plugin_id == "myplugin"
assert "myplugin" in str(exc)
def test_exception_str_no_shelf_image() -> None:
exc = NoShelfImageError("s1", "c1")
assert exc.shelf_id == "s1"
assert exc.cabinet_id == "c1"
assert "s1" in str(exc)
assert "c1" in str(exc)
def test_exception_str_no_cabinet_photo() -> None:
exc = NoCabinetPhotoError("c1")
assert exc.cabinet_id == "c1"
assert "c1" in str(exc)
def test_exception_str_no_raw_text() -> None:
exc = NoRawTextError("b1")
assert exc.book_id == "b1"
assert "b1" in str(exc)
def test_exception_str_invalid_plugin_entity() -> None:
exc = InvalidPluginEntityError("text_recognizer", "cabinets")
assert exc.plugin_category == "text_recognizer"
assert exc.entity_type == "cabinets"
assert "text_recognizer" in str(exc)
assert "cabinets" in str(exc)
def test_exception_str_plugin_target_mismatch() -> None:
exc = PluginTargetMismatchError("my_bd", "shelves", "books")
assert exc.plugin_id == "my_bd"
assert exc.expected_target == "shelves"
assert exc.actual_target == "books"
assert "my_bd" in str(exc)
assert "shelves" in str(exc)
assert "books" in str(exc)