"""Regression tests on the real catalogue. Run: python tests/test_search.py (or: pytest tests) Looks for the MARCXML at $CATALOG_XML or ../opac1.xml (next to app.py). """ import os, sys, time ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") sys.path.insert(0, ROOT) import pandas as pd import search_engine as se import marc_loader as ml import llm XML = os.getenv("CATALOG_XML") or os.path.join(ROOT, "opac1.xml") if not os.path.exists(XML): print(f"SKIP: {XML} not found (set CATALOG_XML)"); sys.exit(0) CAT = ml.load_marcxml(XML, cache_dir=os.path.join(ROOT, ".cache")) BIB, ITEMS = CAT.bib, CAT.items IDX = se.CatalogIndex(BIB, ITEMS) def S(q, **kw): return se.search_books(IDX, q, **kw) def top(q, n=1, **kw): return S(q, **kw).df.head(n) B1 = BIB[BIB.ID == 1].iloc[0] # ---------------------------------------------------------------- loader / data def test_counts_and_ids(): assert len(BIB) == 30542 and len(ITEMS) == 50157 assert BIB.ID.is_unique, "biblionumbers must be unique" assert int(BIB.Copies.sum()) == len(ITEMS) assert CAT.snapshot_date == "2026-04-16" def test_book_1_fields(): assert B1.Title == "Game theory: a nontechnical introduction" assert B1.Author == "Davis, Morton D." and B1.CallNumber == "QA269 .D38 1970" assert B1.ISBN == "0465026265" and B1.Location == "GEN" and B1.Language == "English" assert B1.Year == 1970 and B1.Category == "Game theory" assert se.availability_text(B1) == "1 of 1 copy available to borrow" assert "Shelf 36A-36B" in B1.ShelfLocation def test_titles_cleaned_and_authors_recovered(): assert not BIB.Title.str.contains(r"[/:;,=]\s*$").any(), "ISBD punctuation left on titles" assert (BIB.Author == se.NA).mean() < 0.03 # CSV had ~30% missing; MARC 7xx fields recover them assert (BIB.Category == se.NA).mean() < 0.01 assert ml.tidy_name("KNUTH, Donald E.") == "Knuth, Donald E." and ml.tidy_name("Davis, Morton D.") == "Davis, Morton D." assert ml.clean_isbd("Basic Books,") == "Basic Books" and ml.clean_isbd("Morton D.") == "Morton D." def test_language_from_marc_not_guessed(): bn_title = BIB.Title.str.contains(se.BN_RE) assert BIB[bn_title].Language.str.contains("Bengali").all() assert BIB[BIB.Language == "English"].Title.str.contains(se.BN_RE).sum() == 0 assert 4400 < BIB.Language.str.contains("Bengali").sum() < 4800 def test_item_status_logic_consistent(): assert set(ITEMS.status) <= {"Available", "In-library use only", "Checked out", "Online", "Lost / missing", "Damaged", "Withdrawn"} out = ITEMS[ITEMS.status == "Checked out"] assert out.due.ne("").all() per_bib = ITEMS.groupby("ID").status.apply(lambda s: (s == "Available").sum()) chk = BIB.set_index("ID").AvailableCopies assert (per_bib == chk.loc[per_bib.index]).all() assert BIB.loc[BIB.OnLoanCopies > 0, "NextDue"].ne("").all() def test_multi_copy_titles(): assert BIB[BIB.ID == 6399].Copies.iloc[0] == 29 # Algorithms in C, 29 copies assert BIB.Copies.max() >= 300 # periodicals (Time) keep every issue # ---------------------------------------------------------------- field lookups def test_each_field_finds_book_1(): for label, value in [("ID", 1), ("Title", B1.Title), ("Author", B1.Author), ("Call Number", B1.CallNumber), ("Category", B1.Category), ("Subject", B1.Category), ("Publication Details", B1.Publicationdetails), ("ISBN", B1.ISBN), ("Location", "GEN"), ("Language", "English"), ("Barcode", ITEMS[ITEMS.ID == 1].barcode.iloc[0])]: assert 1 in S(f"{label}: {value}").df.ID.tolist(), f"{label} lookup failed" def test_description_field_uses_real_summary(): row = BIB[BIB.Description != se.NA].iloc[0] assert row.ID in S(f"Description: {row.Description}").df.ID.tolist() def test_isbn10_and_13_find_same_book(): assert 1 in S("ISBN: 9780465026265").df.ID.tolist() and 1 in S("0465026265").df.ID.tolist() assert S("ISBN: 9999999999").total == 0 and S("ID: 999999").total == 0 def test_combined_fields_and_location_language(): r = S("Title: Algorithms Author: Sedgewick") assert r.total >= 3 and r.df.Authors.str.contains("sedgewick", case=False).all() assert S("Location: REF").df.Location.str.contains("REF").all() assert S("Language: Bangla").total == S("Language: Bengali").total > 4000 def test_call_number_matches_any_copy_and_prefix(): assert S("QA269 .D38 1970").df.ID.iloc[0] == 1 r = S("Call Number: QA76.9") assert r.total > 100 assert r.df.CallNumberList.map(lambda L: any(se._clean_key(c).startswith("QA76.9") for c in L)).all() # ---------------------------------------------------------------- label hijack bug def test_everyday_words_not_labels(): for q in ["The call of the wild", "books about facility location theory", "the year 2000 problem", "Man's search for identity"]: assert se.parse_query(q).kind == "natural" and not se.parse_query(q).fields, q assert "London" in top("The call of the wild").Author.iloc[0] # ---------------------------------------------------------------- natural language def test_author_year_language_available(): r = S("books by Harold Laski"); assert r.total >= 1 and r.df.Authors.str.contains("Laski").all() r = S("books published after 2010"); assert r.total > 1000 and (r.df.Year > 2010).all() r = S("recommend a book about accounting published after 2010") assert (r.df.Year >= 2011).all() and r.df.head(5).Title.str.contains("ccounting").any() r = S("books in Bengali"); assert r.total > 4000 and r.df.Language.str.contains("Bengali").all() r = S("books from the 1990s"); assert r.df.Year.between(1990, 1999).all() r = S("available books on algorithms"); assert r.total > 20 and (r.df.AvailableCopies + r.df.InLibraryCopies + r.df.OnlineCopies > 0).all() r = S("e-books on marketing"); assert r.total > 10 and ((r.df.OnlineCopies > 0) | (r.df.URL != "")).all() def test_author_search_uses_added_entries(): # A co-author listed only in a 700 field must still be found (CSV only had the first author). row = BIB[(BIB.Authors.str.count(";") >= 1)].iloc[0] second = row.Authors.split("; ")[1].split(",")[0] assert row.ID in S(f"Author: {second}").df.ID.tolist() def test_ranking_typos_bangla(): assert top("Who wrote A grammar of politics?").Title.iloc[0].lower().startswith("a grammar of politics") assert top("algoritms", 5).Title.str.contains("lgorithm", case=False).any() assert top("Sedgewick algorithms book").Author.iloc[0].lower().startswith("sedgewick") assert S("বাংলা উপন্যাস").total > 0 def test_identifiers_and_speed(): assert se.parse_query("Book ID #2").kind == "id" and se.parse_query("0465026265").kind == "isbn" t = time.time() for q in ["books on poverty in Bangladesh", "machine learning", "Author: Laski", "game theory"]: S(q) assert (time.time() - t) / 4 < 0.5 # ---------------------------------------------------------------- shelves (per copy) def test_shelves_per_copy(): ref = ITEMS[ITEMS.location == "REF"] assert ref.shelf.str.contains("Mezzanine|Branch library").all() assert ITEMS[ITEMS.itype == "EBOOK"].shelf.eq("Online resource — no physical shelf").all() assert ITEMS[ITEMS.holding.isin(["ARC-LIB", "PHR-LIB", "DESM-LIB", "LCR"])].shelf.str.startswith("Branch library").all() assert ITEMS.shelf.str.contains("Check Circulation|Consult Librarian").mean() < 0.01 assert "Shelf 40A-40B" in se.lookup_shelf("QA76.9 .D343 2005") and "Shelf 12A-12B" in se.lookup_shelf("HD28.5 .X1 2000") # ---------------------------------------------------------------- chat helpers / LLM wrapper def test_followups(): rows = [10, 20, 30] assert se.resolve_reference("what's the ISBN of the second one?", rows) == ([20], False) assert se.resolve_reference("is it available?", rows) == ([10], True) assert se.resolve_reference("books about accounting", rows) == (None, False) assert se.MORE_RE.match("show more") and not se.MORE_RE.match("more accounting books") def test_record_line_has_availability_and_shelf(): line = se.record_line(B1) assert "Availability=1 of 1 copy available to borrow" in line and "Shelf=Main Floor" in line and "ISBN=0465026265" in line class FakeResp: def __init__(self, text): self.text = text class FakeModels: """Records every call; can simulate a missing model, a model that rejects thinking settings, or slowness.""" def __init__(self, bad=(), reject_thinking=(), delay=0.0, text='{"author": "Laski", "year_min": 2010, "topic": null}'): self.bad, self.reject_thinking, self.delay, self.text = set(bad), set(reject_thinking), delay, text self.calls, self.configs = [], [] def _check(self, model, config): self.calls.append(model); self.configs.append(dict(config)) if self.delay: time.sleep(self.delay) if model in self.bad: raise RuntimeError("404 NOT_FOUND model not found") if "thinking_config" in config and model in self.reject_thinking: raise RuntimeError("400 INVALID_ARGUMENT: thinking_level is not supported") def generate_content(self, model, contents, config): self._check(model, config); return FakeResp(self.text) def generate_content_stream(self, model, contents, config): self._check(model, config); return iter([FakeResp("Hello "), FakeResp("world")]) class FakeClient: def __init__(self, **kw): self.models = FakeModels(**kw) def _bot(**kw): for k in ("GEMINI_MODEL", "GEMINI_FAST_MODEL"): os.environ.pop(k, None) c = FakeClient(**kw); return llm.LLM("x", client=c, **({"timeout": kw.pop("timeout")} if "timeout" in kw else {})), c def test_llm_model_fallback_and_fast_model_first(): os.environ["GEMINI_MODEL"] = "gemini-3.6-flash" # invalid name from the old app try: c = FakeClient(bad={"gemini-3.6-flash"}); bot = llm.LLM("x", client=c) assert bot.understand("books by Laski after 2010") == {"author": "Laski", "year_min": 2010} assert c.models.calls[0] == "gemini-2.5-flash-lite" # small model used for the interpretation step assert bot.explain("q", ["ID=1 | Title=X"], "", 1, snapshot="2026-04-16") and bot.working == "gemini-3.5-flash" assert c.models.calls[1:3] == ["gemini-3.6-flash", "gemini-3.5-flash"] # invalid model skipped automatically finally: os.environ.pop("GEMINI_MODEL", None) def test_thinking_is_turned_down_and_rejection_is_remembered(): bot, c = _bot(reject_thinking={"gemini-3.5-flash"}) bot.explain("q", [], "", 0) levels = [(cfg.get("thinking_config") or {}).get("thinking_level") for cfg in c.models.configs] assert levels == ["minimal", "low", None], levels # tried fastest first, then fell back assert all("temperature" not in cfg for cfg in c.models.configs) # Gemini 3.x keeps its default temperature n = len(c.models.calls); bot._cache.clear(); bot.explain("another", [], "", 0) assert len(c.models.calls) == n + 1 # rejected settings are not retried bot2, c2 = _bot(bad={"gemini-3.5-flash", "gemini-3-flash-preview"}) bot2.explain("q", [], "", 0) last = c2.models.configs[-1] assert last["thinking_config"] == {"thinking_budget": 0} and last["temperature"] == 0.2 # Gemini 2.5 style def test_timeout_does_not_block(): bot, c = _bot(delay=0.8); bot.timeout = 0.15 t = time.time() try: bot.explain("q", [], "", 0); assert False, "should time out" except TimeoutError: pass assert time.time() - t < 0.6 and "no response" in bot.last_error def test_repeated_questions_are_cached(): bot, c = _bot() a = bot.explain("same question", ["ID=1"], "i", 1); n = len(c.models.calls) b = bot.explain("same question", ["ID=1"], "i", 1) assert a == b and len(c.models.calls) == n and bot.last_cache_hit h1 = bot.understand("zzz"); n = len(c.models.calls); h2 = bot.understand("zzz") assert h1 == h2 and len(c.models.calls) == n def test_streaming_and_cache(): bot, c = _bot() assert list(bot.explain_stream("q", [], "", 0)) == ["Hello ", "world"] n = len(c.models.calls) assert list(bot.explain_stream("q", [], "", 0)) == ["Hello world"] and len(c.models.calls) == n and bot.last_cache_hit def test_prompt_is_short(): lines = [f"ID={i} | Title=T{i}" for i in range(5)] contents = llm.LLM.explain_contents("q", lines, "x", 5, snapshot="2026-04-16") assert contents.count("ID=") == 3 and "SNAPSHOT DATE: 2026-04-16" in contents and "under 60 words" in llm.EXPLAIN_SYSTEM def test_hints_and_json(): r = se.search_books(IDX, "stuff nobody wrote", hints={"author": "Laski"}) assert r.total >= 1 and r.df.Authors.str.contains("Laski").all() assert llm.parse_json('```json\n{"author":"X","year_min":"2005"}\n```') == {"author": "X", "year_min": 2005} if __name__ == "__main__": fails = 0 for name, fn in sorted(globals().items()): if name.startswith("test_") and callable(fn): try: fn(); print("PASS", name) except Exception as e: import traceback; fails += 1; print("FAIL", name, "->", type(e).__name__, e) print(f"\n{'ALL PASSED' if not fails else str(fails) + ' FAILED'}"); sys.exit(1 if fails else 0)