"""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, IdentifyBlock, 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, "ai_blocks": 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,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 model(self) -> str: return "stub-model" @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 model(self) -> str: return "stub-model" @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 model(self) -> str: return "stub-model" @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 model(self) -> str: return "stub-model" @property def max_image_px(self) -> int: return 1600 @property def confidence_threshold(self) -> float: return 0.8 @property def is_vlm(self) -> bool: return False def identify( self, raw_text: str, archive_results: list[CandidateRecord], images: list[tuple[str, str]], ) -> list[IdentifyBlock]: return [IdentifyBlock(title="Found Book", author="Found Author", year="2000", score=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)