Files
library-desk/tests/test_consolidation.py
T
2025-12-11 17:28:23 +01:00

755 lines
21 KiB
Python

"""
Comprehensive tests for Knowledge Consolidation system.
Tests cover:
- ConsolidationService (unit tests with mocks)
- Consolidation API endpoint (integration tests)
- Model validation
- Error handling
- Dry run mode
Run with: pytest tests/test_consolidation.py -v -s
"""
import pytest
import pytest_asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from typing import AsyncGenerator
from datetime import datetime
import json
from src.services.consolidation_service import ConsolidationService
from src.models.consolidation import (
ConsolidationRequest,
ConsolidationResponse,
ConsolidationResult,
SearchQueryInfo
)
# Test constants
TEST_USER = "consolidation-tester"
TEST_SEARCH_ID = "test-search-123"
# Fixtures
@pytest.fixture
def settings():
"""Get mocked application settings for testing."""
mock_settings = MagicMock()
mock_settings.reranker_model = "mistral-nemo"
mock_settings.ollama_model = "mistral-nemo"
return mock_settings
@pytest.fixture
def mock_neo4j():
"""Mock Neo4j client."""
mock = AsyncMock()
mock.execute_query = AsyncMock()
return mock
@pytest.fixture
def mock_ollama():
"""Mock Ollama client."""
mock = AsyncMock()
mock.generate_text = AsyncMock()
return mock
@pytest.fixture
def mock_wiki():
"""Mock Wiki.js client."""
mock = AsyncMock()
# Default mock for taxonomy structure
mock.get_taxonomy_structure = AsyncMock(return_value={
"companies": [],
"people": [],
"places": ["the-netherlands"],
"reference": ["political-entities", "tech"],
"technology": ["tools", "services"]
})
return mock
@pytest.fixture
def mock_ingestion():
"""Mock Ingestion service."""
mock = AsyncMock()
mock.ingest_page = AsyncMock(return_value=MagicMock(success=True))
return mock
@pytest.fixture
def consolidation_service(mock_neo4j, mock_ollama, mock_wiki, mock_ingestion, settings):
"""Get ConsolidationService with mocked dependencies."""
return ConsolidationService(
neo4j=mock_neo4j,
ollama=mock_ollama,
wiki=mock_wiki,
settings=settings,
ingestion_service=mock_ingestion
)
@pytest.fixture
def consolidation_service_no_ingestion(mock_neo4j, mock_ollama, mock_wiki, settings):
"""Get ConsolidationService without ingestion service (legacy behavior)."""
return ConsolidationService(
neo4j=mock_neo4j,
ollama=mock_ollama,
wiki=mock_wiki,
settings=settings,
ingestion_service=None
)
@pytest.fixture
def sample_unprocessed_searches():
"""Sample unprocessed search queries."""
return [
{
'id': 'search-1',
'query': 'docker orchestration kubernetes',
'user': TEST_USER,
'timestamp': datetime.now().isoformat(),
'total_results': 10,
'web_count': 5,
'keywords': ['docker', 'orchestration', 'kubernetes']
},
{
'id': 'search-2',
'query': 'python async programming',
'user': TEST_USER,
'timestamp': datetime.now().isoformat(),
'total_results': 8,
'web_count': 3,
'keywords': ['python', 'async', 'programming']
}
]
@pytest.fixture
def sample_web_results():
"""Sample web search results."""
return [
{
'url': 'https://kubernetes.io/docs',
'title': 'Kubernetes Documentation',
'content': 'Kubernetes is an orchestration platform for containers...',
'rank': 1,
'rrf_score': 0.05
},
{
'url': 'https://docs.docker.com/swarm',
'title': 'Docker Swarm Documentation',
'content': 'Docker Swarm is a container orchestration tool...',
'rank': 2,
'rrf_score': 0.04
},
{
'url': 'https://example.com/k8s-tutorial',
'title': 'Kubernetes Tutorial',
'content': 'Learn how to use Kubernetes for container orchestration...',
'rank': 3,
'rrf_score': 0.03
}
]
@pytest.fixture
def sample_llm_analysis():
"""Sample LLM analysis response."""
return {
"has_novel_info": True,
"new_pages": [
{
"title": "Kubernetes Container Orchestration",
"path": "infrastructure/kubernetes",
"summary": "Overview of Kubernetes orchestration capabilities"
}
],
"update_pages": [
{
"title": "Docker Infrastructure",
"new_facts": [
"Kubernetes provides automatic bin packing",
"Self-healing capabilities with automatic restarts"
],
"source_url": "https://kubernetes.io/docs"
}
],
"new_entities": [
{
"name": "Kubernetes",
"type": "technology",
"description": "Container orchestration platform"
},
{
"name": "Docker Swarm",
"type": "technology",
"description": "Docker's native orchestration tool"
}
]
}
# Model Tests
def test_consolidation_request_validation():
"""Test ConsolidationRequest model validation."""
# Valid request
request = ConsolidationRequest(
process_limit=10,
lookback_days=7,
min_web_results=2,
dry_run=False
)
assert request.process_limit == 10
assert request.lookback_days == 7
assert request.min_web_results == 2
assert request.dry_run is False
# Default values
request = ConsolidationRequest()
assert request.process_limit == 10
assert request.lookback_days == 7
assert request.min_web_results == 2
assert request.dry_run is False
# Validate limits
with pytest.raises(Exception):
ConsolidationRequest(process_limit=0) # Too low
with pytest.raises(Exception):
ConsolidationRequest(process_limit=101) # Too high
def test_consolidation_response_model():
"""Test ConsolidationResponse model."""
response = ConsolidationResponse(
total_found=5,
processed_count=4,
pages_created=2,
pages_updated=3,
entities_added=5,
errors=["Error 1"],
results=[],
dry_run=False
)
assert response.total_found == 5
assert response.processed_count == 4
assert response.pages_created == 2
assert len(response.errors) == 1
def test_consolidation_result_model():
"""Test ConsolidationResult model."""
result = ConsolidationResult(
search_id="test-123",
query="test query",
pages_created=1,
pages_updated=2,
entities_added=3,
error=None
)
assert result.search_id == "test-123"
assert result.query == "test query"
assert result.pages_created == 1
assert result.error is None
# Service Unit Tests
@pytest.mark.asyncio
async def test_find_unprocessed_searches_empty(consolidation_service, mock_neo4j):
"""Test finding unprocessed searches when none exist."""
# Mock empty result
mock_neo4j.execute_query.return_value = []
searches = await consolidation_service._find_unprocessed_searches(
lookback_days=7,
limit=10
)
assert len(searches) == 0
mock_neo4j.execute_query.assert_called_once()
@pytest.mark.asyncio
async def test_find_unprocessed_searches_with_results(
consolidation_service,
mock_neo4j,
sample_unprocessed_searches
):
"""Test finding unprocessed searches with results."""
# Mock Neo4j response
mock_neo4j.execute_query.return_value = sample_unprocessed_searches
searches = await consolidation_service._find_unprocessed_searches(
lookback_days=7,
limit=10
)
assert len(searches) == 2
assert searches[0]['query'] == 'docker orchestration kubernetes'
assert searches[1]['query'] == 'python async programming'
mock_neo4j.execute_query.assert_called_once()
@pytest.mark.asyncio
async def test_get_web_results(consolidation_service, mock_neo4j, sample_web_results):
"""Test retrieving web results for a search."""
# Mock Neo4j response
mock_neo4j.execute_query.return_value = sample_web_results
results = await consolidation_service._get_web_results(TEST_SEARCH_ID)
assert len(results) == 3
assert results[0]['title'] == 'Kubernetes Documentation'
assert results[1]['url'] == 'https://docs.docker.com/swarm'
mock_neo4j.execute_query.assert_called_once()
@pytest.mark.asyncio
async def test_get_web_results_empty(consolidation_service, mock_neo4j):
"""Test retrieving web results when none exist."""
mock_neo4j.execute_query.return_value = []
results = await consolidation_service._get_web_results(TEST_SEARCH_ID)
assert len(results) == 0
@pytest.mark.asyncio
async def test_analyze_web_results_with_novel_info(
consolidation_service,
mock_ollama,
mock_wiki,
sample_web_results,
sample_llm_analysis
):
"""Test analyzing web results with Ollama - novel info found."""
# Mock Ollama response
mock_ollama.generate_text.return_value = json.dumps(sample_llm_analysis)
analysis = await consolidation_service._analyze_web_results(
query="docker orchestration",
web_results=sample_web_results,
keywords=["docker", "orchestration"],
user=TEST_USER
)
assert analysis is not None
assert analysis['has_novel_info'] is True
assert len(analysis['new_pages']) == 1
assert len(analysis['update_pages']) == 1
assert len(analysis['new_entities']) == 2
mock_ollama.generate_text.assert_called_once()
# Verify taxonomy was fetched
mock_wiki.get_taxonomy_structure.assert_called_once_with(f"users/{TEST_USER}")
@pytest.mark.asyncio
async def test_analyze_web_results_no_novel_info(
consolidation_service,
mock_ollama,
sample_web_results
):
"""Test analyzing web results - no novel info."""
# Mock Ollama response with no novel info
analysis_no_novel = {
"has_novel_info": False,
"new_pages": [],
"update_pages": [],
"new_entities": []
}
mock_ollama.generate_text.return_value = json.dumps(analysis_no_novel)
analysis = await consolidation_service._analyze_web_results(
query="common topic",
web_results=sample_web_results,
keywords=[],
user=TEST_USER
)
assert analysis is not None
assert analysis['has_novel_info'] is False
assert len(analysis['new_pages']) == 0
@pytest.mark.asyncio
async def test_analyze_web_results_invalid_json(
consolidation_service,
mock_ollama,
sample_web_results
):
"""Test analyzing web results with invalid JSON response."""
# Mock Ollama response with invalid JSON
mock_ollama.generate_text.return_value = "This is not JSON"
analysis = await consolidation_service._analyze_web_results(
query="test query",
web_results=sample_web_results,
keywords=[],
user=TEST_USER
)
assert analysis is None
@pytest.mark.asyncio
async def test_analyze_web_results_json_in_markdown(
consolidation_service,
mock_ollama,
sample_web_results,
sample_llm_analysis
):
"""Test extracting JSON from markdown-wrapped response."""
# Mock Ollama response with JSON wrapped in markdown
wrapped_response = f"""Here's the analysis:
```json
{json.dumps(sample_llm_analysis)}
```
Hope this helps!"""
mock_ollama.generate_text.return_value = wrapped_response
analysis = await consolidation_service._analyze_web_results(
query="test",
web_results=sample_web_results,
keywords=[],
user=TEST_USER
)
assert analysis is not None
assert analysis['has_novel_info'] is True
@pytest.mark.asyncio
async def test_analyze_web_results_taxonomy_failure(
consolidation_service,
mock_ollama,
mock_wiki,
sample_web_results,
sample_llm_analysis
):
"""Test that analysis continues even if taxonomy fetch fails."""
# Mock taxonomy fetch failure
mock_wiki.get_taxonomy_structure.side_effect = Exception("Connection error")
mock_ollama.generate_text.return_value = json.dumps(sample_llm_analysis)
analysis = await consolidation_service._analyze_web_results(
query="test",
web_results=sample_web_results,
keywords=[],
user=TEST_USER
)
# Should still succeed, just without taxonomy info in prompt
assert analysis is not None
assert analysis['has_novel_info'] is True
@pytest.mark.asyncio
async def test_format_taxonomy_for_prompt(consolidation_service):
"""Test taxonomy formatting for LLM prompt."""
taxonomy = {
"companies": [],
"places": ["the-netherlands", "rotterdam"],
"reference": ["political-entities"]
}
formatted = consolidation_service._format_taxonomy_for_prompt(taxonomy)
assert "Existing paths" in formatted
assert "companies/" in formatted
assert "places/the-netherlands/" in formatted
assert "places/rotterdam/" in formatted
assert "reference/political-entities/" in formatted
assert "IMPORTANT" in formatted
@pytest.mark.asyncio
async def test_format_taxonomy_empty(consolidation_service):
"""Test taxonomy formatting with empty taxonomy."""
formatted = consolidation_service._format_taxonomy_for_prompt({})
assert formatted == ""
@pytest.mark.asyncio
async def test_mark_search_processed(consolidation_service, mock_neo4j):
"""Test marking search as processed."""
await consolidation_service._mark_search_processed(TEST_SEARCH_ID)
mock_neo4j.execute_query.assert_called_once()
call_args = mock_neo4j.execute_query.call_args
assert TEST_SEARCH_ID in str(call_args)
@pytest.mark.asyncio
async def test_process_search_insufficient_web_results(
consolidation_service,
sample_unprocessed_searches
):
"""Test processing search with insufficient web results."""
search = sample_unprocessed_searches[1].copy()
search['web_count'] = 1 # Below minimum
result = await consolidation_service._process_search(
search=search,
min_web_results=2,
dry_run=False
)
assert result is None
@pytest.mark.asyncio
async def test_process_search_no_web_results_in_db(
consolidation_service,
mock_neo4j,
sample_unprocessed_searches
):
"""Test processing search when web results not found in DB."""
mock_neo4j.execute_query.return_value = []
result = await consolidation_service._process_search(
search=sample_unprocessed_searches[0],
min_web_results=2,
dry_run=False
)
assert result is None
@pytest.mark.asyncio
async def test_process_search_dry_run(
consolidation_service,
mock_neo4j,
mock_ollama,
sample_unprocessed_searches,
sample_web_results,
sample_llm_analysis
):
"""Test processing search in dry run mode."""
# Mock responses
mock_neo4j.execute_query.return_value = sample_web_results
mock_ollama.generate_text.return_value = json.dumps(sample_llm_analysis)
result = await consolidation_service._process_search(
search=sample_unprocessed_searches[0],
min_web_results=2,
dry_run=True
)
assert result is not None
assert result.search_id == 'search-1'
assert result.pages_created == 1
assert result.pages_updated == 1
assert result.entities_added == 2
@pytest.mark.asyncio
async def test_consolidate_knowledge_no_searches(
consolidation_service,
mock_neo4j
):
"""Test consolidation when no unprocessed searches found."""
mock_neo4j.execute_query.return_value = []
response = await consolidation_service.consolidate_knowledge(
process_limit=10,
lookback_days=7,
min_web_results=2,
dry_run=False
)
assert response.total_found == 0
assert response.processed_count == 0
assert response.pages_created == 0
@pytest.mark.asyncio
async def test_consolidate_knowledge_success(
consolidation_service,
mock_neo4j,
mock_ollama,
mock_wiki,
sample_unprocessed_searches,
sample_web_results,
sample_llm_analysis
):
"""Test successful knowledge consolidation."""
# Mock finding searches and entity creation
# Each search processes: get web results, add 2 entities, mark processed
mock_neo4j.execute_query.side_effect = [
sample_unprocessed_searches, # Find searches
sample_web_results, # Get web results for search 1
None, # Add entity 1 (Kubernetes)
None, # Add entity 2 (Docker Swarm)
None, # Mark search 1 processed
sample_web_results, # Get web results for search 2
None, # Add entity 1 (Kubernetes)
None, # Add entity 2 (Docker Swarm)
None, # Mark search 2 processed
]
# Mock wiki operations
mock_wiki.search_pages.return_value = [] # No existing pages
mock_wiki.create_page.return_value = None
mock_wiki.update_page.return_value = None
mock_wiki.get_page.return_value = None
# Mock LLM analysis and WikiPageWriter LLM calls
mock_ollama.generate_text.return_value = json.dumps(sample_llm_analysis)
response = await consolidation_service.consolidate_knowledge(
process_limit=10,
lookback_days=7,
min_web_results=2,
dry_run=False
)
assert response.total_found == 2
assert response.processed_count == 2
assert response.dry_run is False
@pytest.mark.asyncio
async def test_consolidate_knowledge_with_errors(
consolidation_service,
mock_neo4j,
mock_ollama,
sample_unprocessed_searches
):
"""Test consolidation with some searches failing."""
# Mock finding searches - return empty for web results to trigger internal error handling
mock_neo4j.execute_query.side_effect = [
sample_unprocessed_searches, # Find searches
[], # Empty web results for search 1 (causes skip, not error)
[], # Empty web results for search 2 (causes skip, not error)
]
response = await consolidation_service.consolidate_knowledge(
process_limit=10,
lookback_days=7,
min_web_results=2,
dry_run=False
)
assert response.total_found == 2
# Both searches skipped due to no web results (not errors)
assert response.processed_count == 0
# Integration Tests (API Endpoint)
@pytest.mark.asyncio
async def test_consolidation_endpoint_minimal_request(consolidation_service):
"""Test consolidation endpoint with minimal request."""
pytest.importorskip("fastapi") # Skip if fastapi not available
# This would require proper test client setup
# Placeholder for integration test structure
request = ConsolidationRequest()
assert request.process_limit == 10
@pytest.mark.asyncio
async def test_consolidation_endpoint_custom_config(consolidation_service):
"""Test consolidation endpoint with custom configuration."""
request = ConsolidationRequest(
process_limit=5,
lookback_days=14,
min_web_results=3,
dry_run=True
)
assert request.process_limit == 5
assert request.lookback_days == 14
assert request.min_web_results == 3
assert request.dry_run is True
# Edge Cases
@pytest.mark.asyncio
async def test_consolidate_with_max_limits(consolidation_service, mock_neo4j):
"""Test consolidation with maximum limits."""
mock_neo4j.execute_query.return_value = []
response = await consolidation_service.consolidate_knowledge(
process_limit=100, # Max
lookback_days=90, # Max
min_web_results=20, # Max
dry_run=True
)
assert response.total_found == 0
@pytest.mark.asyncio
async def test_analyze_empty_web_results(consolidation_service, mock_ollama):
"""Test analyzing with empty web results list."""
mock_ollama.generate_text.return_value = json.dumps({
"has_novel_info": False,
"new_pages": [],
"update_pages": [],
"new_entities": []
})
analysis = await consolidation_service._analyze_web_results(
query="test",
web_results=[],
keywords=[],
user=TEST_USER
)
# Should still call LLM but return no novel info
assert analysis is not None
# Performance/Load Tests (optional)
@pytest.mark.asyncio
async def test_process_many_searches_dry_run(
consolidation_service,
mock_neo4j,
mock_ollama
):
"""Test processing many searches in dry run mode."""
# Generate many test searches
many_searches = [
{
'id': f'search-{i}',
'query': f'test query {i}',
'user': TEST_USER,
'timestamp': datetime.now().isoformat(),
'total_results': 5,
'web_count': 3,
'keywords': ['test']
}
for i in range(50)
]
mock_neo4j.execute_query.return_value = many_searches[:10] # Limit by config
response = await consolidation_service.consolidate_knowledge(
process_limit=10,
lookback_days=7,
min_web_results=2,
dry_run=True
)
# Should only process up to limit
assert response.total_found == 10
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"])