755 lines
21 KiB
Python
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"])
|