| """Offline regressions for KB loading, fallback control flow and staff names. |
| |
| Run with stdlib: python -m unittest tests.test_knowledge_regressions -v |
| The BM25 and embedding adapters are fakes: these checks do not validate actual |
| provider SDK behavior, embedding quality or live retrieval relevance. |
| """ |
| import ast |
| import asyncio |
| import html |
| import importlib.util |
| import os |
| from pathlib import Path |
| import sys |
| import tempfile |
| import threading |
| import types |
| import unittest |
| from unittest.mock import patch |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| class FakeBM25: |
| """A score adapter for testing routing/fallback, not BM25 ranking.""" |
| def __init__(self, tokens): |
| self.tokens = tokens |
|
|
| def get_scores(self, query): |
| return [sum(tokens.count(token) for token in query) for tokens in self.tokens] |
|
|
|
|
| def load_rag(settings): |
| config = types.ModuleType("src.config") |
| config.get_settings = lambda: settings |
| spec = importlib.util.spec_from_file_location("isolated_rag", ROOT / "src/services/rag_service.py") |
| module = importlib.util.module_from_spec(spec) |
| with patch.dict(sys.modules, {"src.config": config}): |
| spec.loader.exec_module(module) |
| return module |
|
|
|
|
| def load_library_helpers(*names): |
| """Execute real pure helpers without importing absent external SDKs.""" |
| tree = ast.parse((ROOT / "src/agentcore/intents_library.py").read_text()) |
| selected = [node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names] |
| import re |
| from typing import List |
| scope = { |
| "re": re, "List": List, "ChatMessage": object, |
| "_normalize_whitespace": lambda value: re.sub(r"\s+", " ", value).strip(), |
| "_escape": html.escape, |
| } |
| exec(compile(ast.Module(body=selected, type_ignores=[]), "real_library_helpers", "exec"), scope) |
| return scope |
|
|
|
|
| class KnowledgeRegressions(unittest.TestCase): |
| def setUp(self): |
| self.tmp = tempfile.TemporaryDirectory() |
| self.addCleanup(self.tmp.cleanup) |
| self.path = Path(self.tmp.name) |
| self.settings = types.SimpleNamespace( |
| kb_dir=self.path / "kb", rag_cache_dir=self.path / "cache", origin_token="unit-test-key" |
| ) |
| self.settings.kb_dir.mkdir() |
| self.settings.rag_cache_dir.mkdir() |
| self.module = load_rag(self.settings) |
| self.service = self.module.RAGService() |
| fake_rank = types.ModuleType("rank_bm25") |
| fake_rank.BM25Okapi = FakeBM25 |
| self.adapters = patch.dict(sys.modules, {"rank_bm25": fake_rank}) |
| self.adapters.start() |
| self.addCleanup(self.adapters.stop) |
|
|
| def write_kb(self, text): |
| path = self.settings.kb_dir / "kb_test.txt" |
| path.write_text(text) |
| return path |
|
|
| def test_resource_entries_stay_with_their_access_urls(self): |
| chunks = self.service._chunk_text( |
| "RESOURCE 01: Scopus\nDescription: Citation database.\nAccess URL: https://example.org/scopus\n\n" |
| "RESOURCE 02: PubMed Central\nDescription: Article archive.\nAccess URL: https://pmc.ncbi.nlm.nih.gov/" |
| ) |
| self.assertEqual(len(chunks), 2) |
| self.assertIn("scopus", chunks[0]) |
| self.assertNotIn("PubMed", chunks[0]) |
| self.assertIn("https://pmc.ncbi.nlm.nih.gov/", chunks[1]) |
|
|
| def test_unpunctuated_headings_are_split(self): |
| chunks = self.service._chunk_text( |
| "STUDY ROOMS\nBook a study room at https://example.org/rooms\n\n" |
| "INTERLIBRARY LOAN\nRequest an article at https://example.org/ill" |
| ) |
| self.assertEqual(len(chunks), 2) |
| self.assertNotIn("INTERLIBRARY", chunks[0]) |
|
|
| def test_metadata_does_not_generate_standalone_chunks(self): |
| self.write_kb("SOURCE: https://example.org\nTITLE: Test Library\nLAST_UPDATED: 2026-09-08\n\nSTUDY ROOMS\nReserve a study room through the service desk.") |
| chunks, meta = self.service._load_and_chunk_all() |
| self.assertEqual(len(chunks), 1) |
| self.assertNotIn("SOURCE:", chunks[0]) |
| self.assertEqual(meta[0]["last_updated"], "2026-09-08") |
| self.assertEqual(meta[0]["last_reviewed"], "") |
| self.assertEqual(meta[0]["authority"], "supplied_library_content") |
|
|
| def test_content_hash_ignores_mtime_but_detects_same_length_edits(self): |
| path = self.write_kb("STUDY ROOMS\nLibrary room booking information.") |
| before = self.service._hash_knowledge_base([path]) |
| os.utime(path, (1, 1)) |
| self.assertEqual(before, self.service._hash_knowledge_base([path])) |
| path.write_text(path.read_text().replace("room", "book")) |
| self.assertNotEqual(before, self.service._hash_knowledge_base([path])) |
|
|
| def test_keyless_startup_and_dense_failure_preserve_keyword_search(self): |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| asyncio.run(self.service.initialize()) |
| self.assertTrue(self.service.is_ready()) |
| self.assertEqual(self.service.get_stats()["retrieval_mode"], "keyword") |
| class BrokenDense: |
| async def asimilarity_search_with_score(self, *args, **kwargs): |
| raise RuntimeError("provider unavailable") |
| self.service.vectorstore = BrokenDense() |
| with self.assertLogs("isolated_rag", level="WARNING"): |
| results = asyncio.run(self.service.hybrid_search("study rooms")) |
| self.assertTrue(results) |
| self.assertEqual(results[0]["retrieval_mode"], "keyword") |
|
|
| def test_semantic_build_failure_keeps_local_readiness(self): |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| vectors = types.ModuleType("langchain_community.vectorstores") |
| class BrokenFAISS: |
| @classmethod |
| async def afrom_texts(cls, *args, **kwargs): |
| raise RuntimeError("embedding API unavailable") |
| vectors.FAISS = BrokenFAISS |
| embeddings = types.ModuleType("langchain_openai") |
| embeddings.OpenAIEmbeddings = lambda **kwargs: object() |
| with patch.dict(sys.modules, {"langchain_openai": embeddings, "langchain_community.vectorstores": vectors}): |
| with self.assertLogs("isolated_rag", level="ERROR"): |
| asyncio.run(self.service.initialize("fake-key")) |
| self.assertTrue(self.service.is_ready()) |
| self.assertEqual(self.service.get_stats()["status"], "keyword_ready") |
| self.assertTrue(asyncio.run(self.service.hybrid_search("study rooms"))) |
|
|
| def test_initialization_keeps_the_event_loop_available(self): |
| started, release = threading.Event(), threading.Event() |
| original = self.service._build_keyword_index |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| def slow_build(): |
| started.set() |
| release.wait(timeout=2) |
| return original() |
| self.service._build_keyword_index = slow_build |
| async def scenario(): |
| task = asyncio.create_task(self.service.initialize()) |
| await asyncio.to_thread(started.wait, 1) |
| self.assertFalse(task.done()) |
| |
| release.set() |
| await task |
| asyncio.run(scenario()) |
| self.assertTrue(self.service.is_ready()) |
|
|
| def test_irrelevant_query_does_not_return_zero_overlap_chunks(self): |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| asyncio.run(self.service.initialize()) |
| self.assertEqual(asyncio.run(self.service.hybrid_search("volcanoes")), []) |
| self.assertEqual(asyncio.run(self.service.hybrid_search("the library please")), []) |
|
|
| def test_negative_bm25_scores_can_still_retrieve_matching_terms(self): |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| asyncio.run(self.service.initialize()) |
| self.service.bm25.get_scores = lambda query: [-0.2] |
| self.assertTrue(asyncio.run(self.service.hybrid_search("study rooms"))) |
|
|
| def test_tampered_cache_cannot_reach_pickle_loader(self): |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| asyncio.run(self.service.initialize()) |
| directory, bm25, state = self.service._cache_paths() |
| directory.mkdir() |
| (directory / "index.pkl").write_bytes(b"untrusted pickle") |
| bm25.write_text("{}") |
| import json |
| state.write_text(json.dumps({"schema": self.module.CACHE_SCHEMA, "kb_hash": self.service._kb_hash, "signature": "invalid"})) |
| class ForbiddenLoader: |
| @staticmethod |
| def load_local(*args, **kwargs): |
| raise AssertionError("untrusted cache must never be deserialized") |
| with self.assertLogs("isolated_rag", level="WARNING"): |
| self.assertFalse(self.service._try_load_cached_indexes(ForbiddenLoader)) |
|
|
| def test_force_rebuild_bypasses_a_valid_cache(self): |
| self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| calls = {"cache": 0, "build": 0} |
| def cached(_faiss): |
| calls["cache"] += 1 |
| self.service.vectorstore = object() |
| return True |
| class FakeFAISS: |
| @classmethod |
| async def afrom_texts(cls, *args, **kwargs): |
| calls["build"] += 1 |
| return object() |
| vectors = types.ModuleType("langchain_community.vectorstores") |
| vectors.FAISS = FakeFAISS |
| embeddings = types.ModuleType("langchain_openai") |
| embeddings.OpenAIEmbeddings = lambda **kwargs: object() |
| self.service._try_load_cached_indexes = cached |
| self.service._save_cached_indexes = lambda: None |
| with patch.dict(sys.modules, {"langchain_openai": embeddings, "langchain_community.vectorstores": vectors}): |
| asyncio.run(self.service.initialize("fake-key")) |
| asyncio.run(self.service.initialize("fake-key", force_rebuild=True)) |
| self.assertEqual(calls, {"cache": 1, "build": 1}) |
|
|
| def test_inflight_retrieval_keeps_consistent_corpus_during_rebuild(self): |
| path = self.write_kb("STUDY ROOMS\nReserve group study rooms through the library service desk.") |
| async def scenario(): |
| await self.service.initialize() |
| started, resume = asyncio.Event(), asyncio.Event() |
| old_doc = self.service.bm25_corpus[0] |
| class DelayedDense: |
| async def asimilarity_search_with_score(self, *args, **kwargs): |
| started.set() |
| await resume.wait() |
| return [(types.SimpleNamespace(page_content=old_doc), 0)] |
| self.service.vectorstore = DelayedDense() |
| request = asyncio.create_task(self.service.hybrid_search("study rooms")) |
| await started.wait() |
| path.write_text("PRINTING\nPrinting information is available at the library service desk.") |
| await self.service.initialize() |
| resume.set() |
| return await request |
| results = asyncio.run(scenario()) |
| self.assertIn("STUDY ROOMS", results[0]["content"]) |
| self.assertNotIn("PRINTING", results[0]["content"]) |
|
|
| def test_uploaded_knowledge_has_no_duplicate_titles(self): |
| self.settings.kb_dir = ROOT / "data/knowledge" |
| self.assertEqual(self.service.audit_knowledge_base()["duplicate_titles"], []) |
| chunks, metadata = self.service._load_and_chunk_all() |
| self.assertGreater(len(chunks), 70) |
| self.assertEqual(len(chunks), len(set(chunks))) |
| pmc = [chunk for chunk in chunks if chunk.startswith("RESOURCE 58:")] |
| self.assertEqual(len(pmc), 1) |
| self.assertIn("NIH/NLM", pmc[0]) |
| self.assertNotIn("RefWorks is a powerful", pmc[0]) |
| self.assertTrue(any("FINDING AND BORROWING A BOOK" in chunk for chunk in chunks)) |
| self.assertTrue(any("Title: Research & Access Services Librarian" in chunk for chunk in chunks)) |
|
|
|
|
| class StaffRegressions(unittest.TestCase): |
| def setUp(self): |
| from src.services import staff_service |
| self.staff = staff_service |
|
|
| def test_alternative_name_spellings_and_first_names(self): |
| for question, expected in [ |
| ("Email Nikesh", "Nikesh Narayanan"), |
| ("Contact Rani", "Rani Anand"), |
| ("Contact Walter Hall", "Walter Brian Hall"), |
| ("Email Alia Alharrasi", "Alia Al-Harrasi"), |
| ("Who is Al Harrasi?", "Alia Al-Harrasi"), |
| ]: |
| with self.subTest(question=question): |
| result = self.staff.match_staff_name(question) |
| self.assertIsNotNone(result) |
| self.assertEqual(result["full_name"], expected) |
|
|
| def test_name_typo_matches_name_tokens(self): |
| self.assertEqual(self.staff.match_staff_name("Email Nikkesh")["full_name"], "Nikesh Narayanan") |
|
|
| def test_common_words_and_role_substrings_do_not_match(self): |
| self.assertIsNone(self.staff.match_staff_name("Who handles hall bookings?")) |
| self.assertIsNone(self.staff.match_staff_role("Who is in the directorate?")) |
|
|
| def test_multiple_people_do_not_silently_choose_one(self): |
| self.assertIsNone(self.staff.match_staff_name("Email Nikesh and Rani")) |
|
|
|
|
| class LibraryHelperRegressions(unittest.TestCase): |
| def test_catalogue_request_preserves_machine_learning(self): |
| helper = load_library_helpers("_catalogue_topic")["_catalogue_topic"] |
| for question in [ |
| "Find books available in the KU library on machine learning", |
| "Does the library have books on machine learning?", |
| ]: |
| self.assertEqual(helper(question), "machine learning") |
|
|
| def test_explicit_rag_followup_uses_previous_topic(self): |
| helper = load_library_helpers("_rag_retrieval_query")["_rag_retrieval_query"] |
| history = [types.SimpleNamespace(role="user", content="How do I use Scopus?")] |
| self.assertIn("Scopus", helper("Tell me more about that resource", history)) |
| self.assertEqual(helper("Where are the study rooms?", history), "Where are the study rooms?") |
|
|
| def test_keyless_rag_excerpt_escapes_html(self): |
| scope = load_library_helpers("_rag_answer") |
| scope["get_settings"] = lambda: types.SimpleNamespace(openai_api_key="", anthropic_api_key="") |
| answer = asyncio.run(scope["_rag_answer"]("Help", [{"content": "<script>bad()</script>\nLibrary text"}], [], "gpt")) |
| self.assertNotIn("<script>", answer) |
| self.assertIn("<script>", answer) |
| self.assertIn("<br>Library text", answer) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
