""" Offline regression tests for the consolidation-loop repair (Phase C item 4). Production failure mode being locked in: - The generation LLM was unavailable (OLLAMA_MODEL env collision made every /api/generate call 400), classification returned empty, and the loop STILL marked every SearchQuery processed - permanently draining the queue with zero output. Every subsequent 30-minute run then logged 'No unprocessed searches found'. The repaired behavior: - LLM-infrastructure failure raises ConsolidationLLMUnavailableError, the affected searches stay UNPROCESSED (retried next run), and the run reports searches_deferred. - Successful classification still consumes searches. - Every run logs searches_processed and duration_ms. All clients are mocked - no shared services are contacted. """ import json from unittest.mock import AsyncMock, MagicMock import pytest from src.services.consolidation_service import ( ConsolidationLLMUnavailableError, ConsolidationService, ) TEST_USER = "llm_tester" def _search_row(search_id: str, query: str = "test query", web_count: int = 3): return { "id": search_id, "query": query, "user": TEST_USER, "timestamp": "2026-07-14T00:00:00Z", "total_results": web_count, "web_count": web_count, "keywords": ["test"], } def _web_result_row(url: str = "https://example.com/a"): return { "url": url, "title": "Example", "content": "Example content about the query.", "rank": 1, "rrf_score": 0.5, } def _make_service(searches, ollama_response): """ConsolidationService with a scripted Neo4j and Ollama.""" neo4j = AsyncMock() executed = [] async def fake_query(cypher, params=None): executed.append((cypher, params or {})) if "processed: false" in cypher: return searches if "FOUND]->(wr:WebResult)" in cypher: return [_web_result_row()] return [] neo4j.execute_query = AsyncMock(side_effect=fake_query) neo4j.execute_write = AsyncMock(side_effect=fake_query) ollama = AsyncMock() ollama.generate_text = AsyncMock(return_value=ollama_response) wiki = AsyncMock() wiki.get_taxonomy_structure = AsyncMock(return_value={}) settings = MagicMock() settings.ollama_llm_model = "gemma4:e2b" service = ConsolidationService( neo4j=neo4j, ollama=ollama, wiki=wiki, settings=settings ) return service, executed def _mark_processed_calls(executed): return [(c, p) for c, p in executed if "SET sq.processed = true" in c] class TestLLMUnavailableDoesNotConsumeSearches: @pytest.mark.asyncio async def test_searches_stay_unprocessed_when_llm_returns_none(self): """generate_text -> None (transport/HTTP failure): defer, don't consume.""" searches = [_search_row("aaaa1111"), _search_row("bbbb2222")] service, executed = _make_service(searches, ollama_response=None) response = await service.consolidate_knowledge() assert response.total_found == 2 assert response.processed_count == 0 assert response.searches_deferred == 2 assert response.errors and "unavailable" in response.errors[0].lower() # THE regression guard: no search was marked processed assert _mark_processed_calls(executed) == [] @pytest.mark.asyncio async def test_searches_stay_unprocessed_when_llm_returns_empty(self): """An empty completion is an infra anomaly, not 'nothing to route'.""" service, executed = _make_service([_search_row("cccc3333")], "") response = await service.consolidate_knowledge() assert response.searches_deferred == 1 assert _mark_processed_calls(executed) == [] @pytest.mark.asyncio async def test_classification_raises_on_no_output(self): service, _ = _make_service([], None) with pytest.raises(ConsolidationLLMUnavailableError): await service._classify_web_results_unified( query="q", web_results=[_web_result_row()], keywords=[], user=TEST_USER ) @pytest.mark.asyncio async def test_batch_aborts_after_first_llm_failure(self): """When the LLM is down it is down for all searches: one probe, then stop.""" searches = [_search_row(f"id{i}", web_count=5) for i in range(5)] service, executed = _make_service(searches, ollama_response=None) response = await service.consolidate_knowledge() assert response.searches_deferred == 5 # Only the first search's classification was attempted assert service.ollama.generate_text.await_count == 1 class TestSuccessfulRunsStillConsume: @pytest.mark.asyncio async def test_valid_classification_marks_processed(self): classification = json.dumps([{ "url": "https://example.com/a", "title": "Example", "route_type": "skip", "confidence": 0.9, "reason": "low value", }]) service, executed = _make_service([_search_row("dddd4444")], classification) response = await service.consolidate_knowledge() assert response.total_found == 1 assert response.processed_count == 1 assert response.searches_deferred == 0 marked = _mark_processed_calls(executed) assert len(marked) == 1 assert marked[0][1]["search_id"] == "dddd4444" @pytest.mark.asyncio async def test_unparseable_output_still_consumes_search(self): """Model responded with junk: consume (avoid retrying a bad prompt forever).""" service, executed = _make_service( [_search_row("eeee5555")], "not json at all" ) response = await service.consolidate_knowledge() assert response.searches_deferred == 0 assert len(_mark_processed_calls(executed)) == 1 @pytest.mark.asyncio async def test_skipped_low_web_search_still_consumed(self): """Insufficient web results: intentionally consumed (existing behavior).""" service, executed = _make_service( [_search_row("ffff6666", web_count=0)], None ) response = await service.consolidate_knowledge() assert response.searches_deferred == 0 assert len(_mark_processed_calls(executed)) == 1 # LLM never called for a skipped search service.ollama.generate_text.assert_not_awaited() class TestRunLogging: @pytest.mark.asyncio async def test_logs_searches_processed_and_duration(self, caplog): service, _ = _make_service([], None) with caplog.at_level("INFO"): response = await service.consolidate_knowledge() assert response.duration_ms >= 0 run_logs = [r.message for r in caplog.records if "Consolidation run complete" in r.message] assert run_logs, "every run must emit the run-complete log line" assert "searches_processed=0" in run_logs[0] assert "duration_ms=" in run_logs[0] @pytest.mark.asyncio async def test_logs_on_deferred_run(self, caplog): service, _ = _make_service([_search_row("gggg7777")], None) with caplog.at_level("INFO"): response = await service.consolidate_knowledge() assert response.duration_ms >= 0 run_logs = [r.message for r in caplog.records if "Consolidation run complete" in r.message] assert run_logs assert "searches_deferred=1" in run_logs[0] @pytest.mark.asyncio async def test_lookback_parameter_is_utc_aware(self): """The lookback boundary must be timezone-aware UTC (Neo4j datetime() interprets naive strings as UTC, shifting the window on CET hosts).""" service, executed = _make_service([], None) await service.consolidate_knowledge(lookback_days=7) find_calls = [(c, p) for c, p in executed if "processed: false" in c] assert find_calls lookback = find_calls[0][1]["lookback_date"] assert "+00:00" in lookback or lookback.endswith("Z")