diff --git a/services/library-desk/tests/test_consolidation.py b/services/library-desk/tests/test_consolidation.py index a75b7cb..b66ba3e 100644 --- a/services/library-desk/tests/test_consolidation.py +++ b/services/library-desk/tests/test_consolidation.py @@ -25,7 +25,6 @@ from src.models.consolidation import ( ConsolidationResult, SearchQueryInfo ) -from src.config import get_settings # Test constants TEST_USER = "consolidation-tester" @@ -36,8 +35,11 @@ TEST_SEARCH_ID = "test-search-123" @pytest.fixture def settings(): - """Get application settings.""" - return get_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 @@ -60,17 +62,46 @@ def mock_ollama(): 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 consolidation_service(mock_neo4j, mock_ollama, mock_wiki, settings): +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 + 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 ) @@ -297,6 +328,7 @@ async def test_get_web_results_empty(consolidation_service, mock_neo4j): async def test_analyze_web_results_with_novel_info( consolidation_service, mock_ollama, + mock_wiki, sample_web_results, sample_llm_analysis ): @@ -307,7 +339,8 @@ async def test_analyze_web_results_with_novel_info( analysis = await consolidation_service._analyze_web_results( query="docker orchestration", web_results=sample_web_results, - keywords=["docker", "orchestration"] + keywords=["docker", "orchestration"], + user=TEST_USER ) assert analysis is not None @@ -316,6 +349,8 @@ async def test_analyze_web_results_with_novel_info( 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 @@ -337,7 +372,8 @@ async def test_analyze_web_results_no_novel_info( analysis = await consolidation_service._analyze_web_results( query="common topic", web_results=sample_web_results, - keywords=[] + keywords=[], + user=TEST_USER ) assert analysis is not None @@ -358,7 +394,8 @@ async def test_analyze_web_results_invalid_json( analysis = await consolidation_service._analyze_web_results( query="test query", web_results=sample_web_results, - keywords=[] + keywords=[], + user=TEST_USER ) assert analysis is None @@ -385,13 +422,65 @@ Hope this helps!""" analysis = await consolidation_service._analyze_web_results( query="test", web_results=sample_web_results, - keywords=[] + 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.""" @@ -563,8 +652,7 @@ async def test_consolidate_knowledge_with_errors( @pytest.mark.asyncio async def test_consolidation_endpoint_minimal_request(consolidation_service): """Test consolidation endpoint with minimal request.""" - from fastapi.testclient import TestClient - from src.main import app + pytest.importorskip("fastapi") # Skip if fastapi not available # This would require proper test client setup # Placeholder for integration test structure @@ -618,7 +706,8 @@ async def test_analyze_empty_web_results(consolidation_service, mock_ollama): analysis = await consolidation_service._analyze_web_results( query="test", web_results=[], - keywords=[] + keywords=[], + user=TEST_USER ) # Should still call LLM but return no novel info diff --git a/services/library-desk/tests/test_graph_service.py b/services/library-desk/tests/test_graph_service.py new file mode 100644 index 0000000..cd20f33 --- /dev/null +++ b/services/library-desk/tests/test_graph_service.py @@ -0,0 +1,249 @@ +""" +Tests for GraphService - knowledge graph operations. + +Tests cover: +- Document node creation with tags +- Entity-stub page skipping +- Entity extraction + +Run with: pytest tests/test_graph_service.py -v -s +""" + +import pytest +import pytest_asyncio +from unittest.mock import AsyncMock, MagicMock, patch +from typing import AsyncGenerator + +from src.services.graph_service import GraphService + + +# Test constants +TEST_USER = "graph-tester" +TEST_PAGE_ID = 123 + + +@pytest.fixture +def mock_neo4j(): + """Mock Neo4j client.""" + mock = AsyncMock() + mock.execute_query = AsyncMock(return_value=[{"d": {"page_id": TEST_PAGE_ID}}]) + return mock + + +@pytest.fixture +def mock_wiki(): + """Mock Wiki.js client.""" + mock = AsyncMock() + return mock + + +@pytest.fixture +def graph_service(mock_neo4j, mock_wiki): + """Get GraphService with mocked dependencies.""" + return GraphService( + neo4j_client=mock_neo4j, + wikijs_client=mock_wiki + ) + + +@pytest.fixture +def sample_page(): + """Sample wiki page data.""" + return { + "id": TEST_PAGE_ID, + "title": "Test Page", + "path": f"users/{TEST_USER}/technology/docker", + "content": "Docker is a containerization platform. It uses containers to run applications.", + "tags": ["technology", "docker", "containers"] + } + + +@pytest.fixture +def sample_page_without_tags(): + """Sample wiki page without tags.""" + return { + "id": TEST_PAGE_ID, + "title": "Test Page No Tags", + "path": f"users/{TEST_USER}/misc/test", + "content": "This is a test page with no tags.", + "tags": [] + } + + +class TestDocumentNodeCreation: + """Test Document node creation in Neo4j.""" + + @pytest.mark.asyncio + async def test_document_node_includes_tags( + self, + graph_service, + mock_neo4j, + mock_wiki, + sample_page + ): + """Test that Document node is created with tags property.""" + mock_wiki.get_page = AsyncMock(return_value=sample_page) + + result = await graph_service.update_from_page( + page_id=TEST_PAGE_ID, + user=TEST_USER + ) + + # Verify execute_query was called + assert mock_neo4j.execute_query.called + assert result.success is True + + # Find the document creation query + calls = mock_neo4j.execute_query.call_args_list + doc_creation_call = None + for call in calls: + query = call[0][0] if call[0] else "" + if "MERGE" in query and "Document" in query and "tags" in query: + doc_creation_call = call + break + + assert doc_creation_call is not None, "Document creation query with tags not found" + + # Verify tags are in the query parameters + params = doc_creation_call[0][1] if len(doc_creation_call[0]) > 1 else {} + assert "tags" in params + assert params["tags"] == ["technology", "docker", "containers"] + + @pytest.mark.asyncio + async def test_document_node_with_empty_tags( + self, + graph_service, + mock_neo4j, + mock_wiki, + sample_page_without_tags + ): + """Test Document node creation with empty tags list.""" + mock_wiki.get_page = AsyncMock(return_value=sample_page_without_tags) + + result = await graph_service.update_from_page( + page_id=TEST_PAGE_ID, + user=TEST_USER + ) + + assert result.success is True + + # Find the document creation query + calls = mock_neo4j.execute_query.call_args_list + doc_creation_call = None + for call in calls: + query = call[0][0] if call[0] else "" + if "MERGE" in query and "Document" in query: + doc_creation_call = call + break + + assert doc_creation_call is not None + params = doc_creation_call[0][1] if len(doc_creation_call[0]) > 1 else {} + assert "tags" in params + assert params["tags"] == [] + + +class TestEntityStubSkipping: + """Test that entity-stub pages are skipped.""" + + @pytest.mark.asyncio + async def test_skip_entity_stub_pages( + self, + graph_service, + mock_neo4j, + mock_wiki + ): + """Test that entity-stub tagged pages skip entity extraction.""" + stub_page = { + "id": TEST_PAGE_ID, + "title": "Auto Entity", + "path": f"users/{TEST_USER}/entities/test", + "content": "Auto-generated content.", + "tags": ["entity-stub", "auto-generated"] + } + mock_wiki.get_page = AsyncMock(return_value=stub_page) + + result = await graph_service.update_from_page( + page_id=TEST_PAGE_ID, + user=TEST_USER + ) + + # Should return success but skip processing + assert result.success is True + # Neo4j should NOT be called for entity-stub pages + assert mock_neo4j.execute_query.call_count == 0 + + @pytest.mark.asyncio + async def test_skip_auto_generated_pages( + self, + graph_service, + mock_neo4j, + mock_wiki + ): + """Test that auto-generated tagged pages skip entity extraction.""" + auto_page = { + "id": TEST_PAGE_ID, + "title": "Auto Page", + "path": f"users/{TEST_USER}/auto/test", + "content": "Auto-generated content.", + "tags": ["auto-generated"] + } + mock_wiki.get_page = AsyncMock(return_value=auto_page) + + result = await graph_service.update_from_page( + page_id=TEST_PAGE_ID, + user=TEST_USER + ) + + assert result.success is True + assert mock_neo4j.execute_query.call_count == 0 + + +class TestPageNotFound: + """Test handling of missing pages.""" + + @pytest.mark.asyncio + async def test_page_not_found_returns_failure( + self, + graph_service, + mock_wiki + ): + """Test that missing page returns failure result.""" + mock_wiki.get_page = AsyncMock(return_value=None) + + result = await graph_service.update_from_page( + page_id=999, + user=TEST_USER + ) + + # The service catches the exception and returns a failed result + assert result.success is False + assert result.error_message is not None + assert "not found" in result.error_message.lower() + + +class TestEntityExtraction: + """Test entity extraction from page content.""" + + @pytest.mark.asyncio + async def test_creates_document_and_entities( + self, + graph_service, + mock_neo4j, + mock_wiki, + sample_page + ): + """Test that document and entity nodes are created.""" + mock_wiki.get_page = AsyncMock(return_value=sample_page) + + result = await graph_service.update_from_page( + page_id=TEST_PAGE_ID, + user=TEST_USER + ) + + assert result.success is True + # Should have called neo4j at least once (for document node) + assert mock_neo4j.execute_query.called + + +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"]) diff --git a/services/library-desk/tests/test_ingestion.py b/services/library-desk/tests/test_ingestion.py new file mode 100644 index 0000000..d70f1b3 --- /dev/null +++ b/services/library-desk/tests/test_ingestion.py @@ -0,0 +1,314 @@ +""" +Tests for IngestionService - document ingestion operations. + +Tests cover: +- Single page ingestion +- Batch ingestion +- Full re-index (ingest_all_pages) +- list_all_pages usage + +Run with: pytest tests/test_ingestion.py -v -s +""" + +import pytest +import pytest_asyncio +from unittest.mock import AsyncMock, MagicMock, patch +from typing import AsyncGenerator + +from src.services.ingestion_service import IngestionService +from src.models.ingestion import ( + IngestionRequest, + IngestionResult, + BatchIngestionRequest, + BatchIngestionResult +) + + +# Test constants +TEST_USER = "ingestion-tester" +TEST_PAGE_ID = 456 + + +@pytest.fixture +def mock_vector_service(): + """Mock Vector service.""" + mock = AsyncMock() + mock.update_from_page = AsyncMock(return_value=MagicMock( + chunks_created=2, + chunks_deleted=0, + success=True + )) + return mock + + +@pytest.fixture +def mock_graph_service(): + """Mock Graph service.""" + mock = AsyncMock() + mock.update_from_page = AsyncMock(return_value=MagicMock( + entities_extracted=5, + relationships_created=5, + success=True + )) + mock.create_entity_mention_links = AsyncMock(return_value=3) + mock.get_all_entities = AsyncMock(return_value=[]) + return mock + + +@pytest.fixture +def mock_wiki_client(): + """Mock Wiki.js client.""" + mock = AsyncMock() + mock.get_page = AsyncMock(return_value={ + "id": TEST_PAGE_ID, + "title": "Test Page", + "path": f"users/{TEST_USER}/test", + "content": "Test content here.", + "tags": ["test"] + }) + mock.list_all_pages = AsyncMock(return_value=[ + {"id": 1, "path": f"users/{TEST_USER}/page1", "title": "Page 1"}, + {"id": 2, "path": f"users/{TEST_USER}/page2", "title": "Page 2"}, + {"id": 3, "path": f"users/{TEST_USER}/page3", "title": "Page 3"}, + ]) + return mock + + +@pytest.fixture +def ingestion_service(mock_vector_service, mock_graph_service, mock_wiki_client): + """Get IngestionService with mocked dependencies.""" + return IngestionService( + vector_service=mock_vector_service, + graph_service=mock_graph_service, + wiki_client=mock_wiki_client + ) + + +class TestSinglePageIngestion: + """Test single page ingestion.""" + + @pytest.mark.asyncio + async def test_ingest_page_success( + self, + ingestion_service, + mock_wiki_client + ): + """Test successful page ingestion.""" + result = await ingestion_service.ingest_page( + page_id=TEST_PAGE_ID, + user=TEST_USER + ) + + assert result.success is True + assert result.page_id == TEST_PAGE_ID + mock_wiki_client.get_page.assert_called_once_with(TEST_PAGE_ID) + + @pytest.mark.asyncio + async def test_ingest_page_not_found( + self, + ingestion_service, + mock_wiki_client + ): + """Test ingestion when page not found.""" + mock_wiki_client.get_page.return_value = None + + result = await ingestion_service.ingest_page( + page_id=999, + user=TEST_USER + ) + + assert result.success is False + assert "not found" in result.error.lower() + + @pytest.mark.asyncio + async def test_ingest_page_skip_vectors( + self, + ingestion_service, + mock_vector_service, + mock_graph_service + ): + """Test ingestion with vectors skipped.""" + result = await ingestion_service.ingest_page( + page_id=TEST_PAGE_ID, + user=TEST_USER, + skip_vectors=True + ) + + assert result.success is True + # Vector service should not be called + mock_vector_service.update_from_page.assert_not_called() + # Graph service should still be called + mock_graph_service.update_from_page.assert_called_once() + + @pytest.mark.asyncio + async def test_ingest_page_skip_graph( + self, + ingestion_service, + mock_vector_service, + mock_graph_service + ): + """Test ingestion with graph skipped.""" + result = await ingestion_service.ingest_page( + page_id=TEST_PAGE_ID, + user=TEST_USER, + skip_graph=True + ) + + assert result.success is True + # Vector service should be called + mock_vector_service.update_from_page.assert_called_once() + # Graph service should not be called + mock_graph_service.update_from_page.assert_not_called() + + +class TestBatchIngestion: + """Test batch page ingestion.""" + + @pytest.mark.asyncio + async def test_ingest_batch_success( + self, + ingestion_service, + mock_wiki_client + ): + """Test successful batch ingestion.""" + result = await ingestion_service.ingest_batch( + page_ids=[1, 2, 3], + user=TEST_USER, + max_concurrent=2 + ) + + assert result.total_pages == 3 + assert result.successful == 3 + assert result.failed == 0 + + @pytest.mark.asyncio + async def test_ingest_batch_with_failures( + self, + ingestion_service, + mock_wiki_client + ): + """Test batch ingestion with some failures.""" + # Make page 2 not found + def get_page_side_effect(page_id): + if page_id == 2: + return None + return { + "id": page_id, + "title": f"Page {page_id}", + "path": f"users/{TEST_USER}/page{page_id}", + "content": "Content", + "tags": [] + } + + mock_wiki_client.get_page.side_effect = get_page_side_effect + + result = await ingestion_service.ingest_batch( + page_ids=[1, 2, 3], + user=TEST_USER + ) + + assert result.total_pages == 3 + assert result.successful == 2 + assert result.failed == 1 + + +class TestIngestAllPages: + """Test full re-index (ingest_all_pages).""" + + @pytest.mark.asyncio + async def test_ingest_all_uses_list_all_pages( + self, + ingestion_service, + mock_wiki_client + ): + """Test that ingest_all_pages uses list_all_pages (not search).""" + result = await ingestion_service.ingest_all_pages( + user=TEST_USER + ) + + # Should use list_all_pages, not search_pages + mock_wiki_client.list_all_pages.assert_called_once() + # Should have processed 3 pages from the mock + assert result.total_pages == 3 + + @pytest.mark.asyncio + async def test_ingest_all_with_path_prefix( + self, + ingestion_service, + mock_wiki_client + ): + """Test ingest_all_pages with path prefix filter.""" + await ingestion_service.ingest_all_pages( + user=TEST_USER, + path_prefix=f"users/{TEST_USER}/technology" + ) + + mock_wiki_client.list_all_pages.assert_called_once_with( + path_prefix=f"users/{TEST_USER}/technology" + ) + + @pytest.mark.asyncio + async def test_ingest_all_empty_wiki( + self, + ingestion_service, + mock_wiki_client + ): + """Test ingest_all_pages when no pages found.""" + mock_wiki_client.list_all_pages.return_value = [] + + result = await ingestion_service.ingest_all_pages( + user=TEST_USER + ) + + assert result.total_pages == 0 + assert result.successful == 0 + + @pytest.mark.asyncio + async def test_ingest_all_respects_max_concurrent( + self, + ingestion_service, + mock_wiki_client + ): + """Test that max_concurrent parameter is passed through.""" + # Create many pages + mock_wiki_client.list_all_pages.return_value = [ + {"id": i, "path": f"users/{TEST_USER}/page{i}", "title": f"Page {i}"} + for i in range(20) + ] + + result = await ingestion_service.ingest_all_pages( + user=TEST_USER, + max_concurrent=5 + ) + + assert result.total_pages == 20 + + +class TestIngestionModels: + """Test ingestion request/response models.""" + + def test_ingestion_request_defaults(self): + """Test IngestionRequest default values.""" + request = IngestionRequest( + page_id=123, + user="testuser" + ) + assert request.page_id == 123 + assert request.user == "testuser" + assert request.force_refresh is False + assert request.skip_vectors is False + assert request.skip_graph is False + + def test_batch_ingestion_request(self): + """Test BatchIngestionRequest.""" + request = BatchIngestionRequest( + page_ids=[1, 2, 3], + user="testuser", + max_concurrent=5 + ) + assert len(request.page_ids) == 3 + assert request.max_concurrent == 5 + + +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"]) diff --git a/services/library-desk/tests/test_integration.py b/services/library-desk/tests/test_integration.py index e02c83a..34f3937 100644 --- a/services/library-desk/tests/test_integration.py +++ b/services/library-desk/tests/test_integration.py @@ -187,6 +187,27 @@ class TestWikiJSIntegration: assert isinstance(pages, list) # May be empty if wiki is new + @pytest.mark.asyncio + async def test_list_all_pages(self, wikijs_client): + """Test listing all pages with pagination support.""" + pages = await wikijs_client.list_all_pages(path_prefix="users/") + assert isinstance(pages, list) + # Verify each page has expected fields + for page in pages[:5]: # Check first 5 + assert "id" in page + assert "path" in page + assert "title" in page + + @pytest.mark.asyncio + async def test_get_taxonomy_structure(self, wikijs_client): + """Test getting taxonomy structure for a user.""" + taxonomy = await wikijs_client.get_taxonomy_structure("users/jpmschweitzer") + assert isinstance(taxonomy, dict) + # Each key should be a category, value should be list of subcategories + for category, subcategories in taxonomy.items(): + assert isinstance(category, str) + assert isinstance(subcategories, list) + @pytest.mark.asyncio async def test_search_pages(self, wikijs_client): """Test searching pages.""" diff --git a/services/library-desk/tests/test_wiki_change_listener.py b/services/library-desk/tests/test_wiki_change_listener.py new file mode 100644 index 0000000..b04d3fd --- /dev/null +++ b/services/library-desk/tests/test_wiki_change_listener.py @@ -0,0 +1,435 @@ +""" +Tests for Wiki.js Change Listener + +Tests the PostgreSQL NOTIFY/LISTEN change detection system including: +- Database connection and listener startup +- Notification handling (INSERT/UPDATE/DELETE) +- Loop prevention (automated user filtering) +- Debouncing (duplicate notification filtering) +- Page processing +""" + +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +from datetime import datetime, timedelta + +from src.services.wiki_change_listener import WikiChangeListener + + +@pytest.fixture +def mock_settings(): + """Mock settings for testing.""" + settings = MagicMock() + settings.wikijs_db_host = "postgres-shared" + settings.wikijs_db_port = 5432 + settings.wikijs_db_user = "library_desk_listener" + settings.wikijs_db_password = "test_password" + settings.wikijs_db_name = "library" + settings.wikijs_username = "librarian@schweitz.net" + settings.wikijs_change_listener_debounce_seconds = 5 + return settings + + +@pytest.fixture +def listener(mock_settings): + """Create a WikiChangeListener instance with mocked settings.""" + with patch('src.services.wiki_change_listener.get_settings', return_value=mock_settings): + return WikiChangeListener() + + +class TestWikiChangeListener: + """Test suite for WikiChangeListener.""" + + @pytest.mark.asyncio + async def test_listener_initialization(self, listener, mock_settings): + """Test that listener initializes with correct settings.""" + assert listener.settings == mock_settings + assert listener.connection is None + assert listener.running is False + assert listener._debounce_seconds == 5 + assert len(listener._recent_notifications) == 0 + + @pytest.mark.asyncio + async def test_automated_user_filtering(self, listener): + """Test that automated users are correctly identified.""" + # Automated users should be filtered + assert listener._is_automated_user("librarian@schweitz.net") is True + assert listener._is_automated_user("library-desk@system") is True + assert listener._is_automated_user("automation@system") is True + assert listener._is_automated_user("bot@system") is True + + # Case insensitive + assert listener._is_automated_user("LIBRARIAN@SCHWEITZ.NET") is True + + # Regular users should not be filtered + assert listener._is_automated_user("user@example.com") is False + assert listener._is_automated_user("john@example.com") is False + + @pytest.mark.asyncio + async def test_debouncing_prevents_duplicates(self, listener): + """Test that debouncing prevents duplicate processing.""" + page_id = 123 + + # First notification - should not be filtered + assert listener._is_recently_processed(page_id) is False + + # Mark as processed + listener._mark_as_processed(page_id) + + # Immediate second notification - should be filtered (within debounce window) + assert listener._is_recently_processed(page_id) is True + + # Different page - should not be filtered + assert listener._is_recently_processed(456) is False + + @pytest.mark.asyncio + async def test_debouncing_expires_after_window(self, listener): + """Test that debouncing expires after the configured time window.""" + page_id = 123 + + # Mark as processed with old timestamp (outside debounce window) + listener._recent_notifications[page_id] = datetime.now() - timedelta(seconds=10) + + # Should not be filtered anymore (10 seconds > 5 second debounce) + assert listener._is_recently_processed(page_id) is False + + @pytest.mark.asyncio + async def test_mark_as_processed_cleanup(self, listener): + """Test that old entries are cleaned up to prevent memory growth.""" + # Add 101 entries to trigger cleanup (threshold is 100) + for i in range(101): + listener._mark_as_processed(i) + + # Should only keep last 100 entries + assert len(listener._recent_notifications) == 100 + + # Oldest entry (0) should be removed + assert 0 not in listener._recent_notifications + + # Newest entries should be kept + assert 100 in listener._recent_notifications + + @pytest.mark.asyncio + async def test_notification_payload_parsing(self, listener): + """Test that notification payloads are correctly parsed.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # Test UPDATE notification + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:123:user@example.com' + ) + + mock_process.assert_called_once() + call_args = mock_process.call_args[1] + assert call_args['page_id'] == 123 + assert call_args['event'] == 'page.update' + assert call_args['user'] == 'user' + + @pytest.mark.asyncio + async def test_notification_operations_mapping(self, listener): + """Test that database operations map to correct webhook events.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # INSERT -> page.create (use page 100) + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'INSERT:100:user@example.com' + ) + mock_process.assert_called_once() + assert mock_process.call_args[1]['event'] == 'page.create' + + mock_process.reset_mock() + + # UPDATE -> page.update (use different page 200 to avoid debouncing) + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:200:user@example.com' + ) + mock_process.assert_called_once() + assert mock_process.call_args[1]['event'] == 'page.update' + + mock_process.reset_mock() + + # DELETE -> page.delete (use different page 300 to avoid debouncing) + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'DELETE:300:user@example.com' + ) + mock_process.assert_called_once() + assert mock_process.call_args[1]['event'] == 'page.delete' + + @pytest.mark.asyncio + async def test_automated_user_notification_filtered(self, listener): + """Test that notifications from automated users are filtered out.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # Notification from automated user should be skipped + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:123:librarian@schweitz.net' + ) + + # Process should NOT be called + mock_process.assert_not_called() + + @pytest.mark.asyncio + async def test_duplicate_notification_filtered(self, listener): + """Test that duplicate notifications within debounce window are filtered.""" + mock_connection = AsyncMock() + page_id = 123 + + # Mark page as recently processed + listener._mark_as_processed(page_id) + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # Duplicate notification should be skipped + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + f'UPDATE:{page_id}:user@example.com' + ) + + # Process should NOT be called + mock_process.assert_not_called() + + @pytest.mark.asyncio + async def test_invalid_notification_payload_handled(self, listener): + """Test that invalid notification payloads are handled gracefully.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # Invalid payload (too few parts) + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'INVALID:123' # Missing user email + ) + + # Should not crash and should not process + mock_process.assert_not_called() + + @pytest.mark.asyncio + async def test_email_to_username_extraction(self, listener): + """Test that user email is correctly extracted to username.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:123:john.doe@example.com' + ) + + # Should extract 'john.doe' from email + call_args = mock_process.call_args[1] + assert call_args['user'] == 'john.doe' + + @pytest.mark.asyncio + async def test_email_without_at_sign_fallback(self, listener): + """Test fallback when email doesn't contain @ sign.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:123:invaliduser' + ) + + # Should use default user + call_args = mock_process.call_args[1] + assert call_args['user'] == 'jpmschweitzer' + + @pytest.mark.asyncio + async def test_process_page_delete_calls_cleanup(self, listener): + """Test that DELETE events call cleanup_deleted_page.""" + with patch('src.routers.webhooks.cleanup_deleted_page', new_callable=AsyncMock) as mock_cleanup, \ + patch('src.services.wiki_change_listener.get_ingestion_service') as mock_service: + + await listener._process_page_change( + page_id=123, + event='page.delete', + user='testuser' + ) + + mock_cleanup.assert_called_once() + call_args = mock_cleanup.call_args[1] + assert call_args['page_id'] == 123 + assert call_args['user'] == 'testuser' + + @pytest.mark.asyncio + async def test_process_page_create_calls_process_wiki_page_change(self, listener): + """Test that CREATE events call process_wiki_page_change.""" + mock_page = MagicMock() + mock_page.title = "Test Page" + + with patch('src.routers.webhooks.process_wiki_page_change', new_callable=AsyncMock) as mock_process, \ + patch('src.core.dependencies.get_wiki_service') as mock_wiki_service_factory, \ + patch('src.services.wiki_change_listener.get_ingestion_service'): + + mock_wiki_service = AsyncMock() + mock_wiki_service.get_page.return_value = mock_page + mock_wiki_service_factory.return_value = mock_wiki_service + + await listener._process_page_change( + page_id=123, + event='page.create', + user='testuser' + ) + + mock_process.assert_called_once() + call_args = mock_process.call_args[1] + assert call_args['page_id'] == 123 + assert call_args['page_title'] == "Test Page" + assert call_args['event'] == 'page.create' + assert call_args['user'] == 'testuser' + + @pytest.mark.asyncio + async def test_process_page_update_calls_process_wiki_page_change(self, listener): + """Test that UPDATE events call process_wiki_page_change.""" + mock_page = MagicMock() + mock_page.title = "Updated Page" + + with patch('src.routers.webhooks.process_wiki_page_change', new_callable=AsyncMock) as mock_process, \ + patch('src.core.dependencies.get_wiki_service') as mock_wiki_service_factory, \ + patch('src.services.wiki_change_listener.get_ingestion_service'): + + mock_wiki_service = AsyncMock() + mock_wiki_service.get_page.return_value = mock_page + mock_wiki_service_factory.return_value = mock_wiki_service + + await listener._process_page_change( + page_id=456, + event='page.update', + user='testuser' + ) + + mock_process.assert_called_once() + call_args = mock_process.call_args[1] + assert call_args['page_id'] == 456 + assert call_args['page_title'] == "Updated Page" + assert call_args['event'] == 'page.update' + + @pytest.mark.asyncio + async def test_connection_lifecycle(self, listener, mock_settings): + """Test listener connection start and stop lifecycle.""" + mock_connection = AsyncMock() + + with patch('src.services.wiki_change_listener.asyncpg.connect', return_value=mock_connection) as mock_connect: + # Start listener + await listener.start() + + # Verify connection was established with correct parameters + mock_connect.assert_called_once_with( + host=mock_settings.wikijs_db_host, + port=mock_settings.wikijs_db_port, + user=mock_settings.wikijs_db_user, + password=mock_settings.wikijs_db_password, + database=mock_settings.wikijs_db_name + ) + + # Verify listener was added + mock_connection.add_listener.assert_called_once_with( + 'wiki_page_changes', + listener._handle_notification + ) + + assert listener.running is True + assert listener.connection == mock_connection + + # Stop listener + await listener.stop() + + # Verify listener was removed and connection closed + mock_connection.remove_listener.assert_called_once() + mock_connection.close.assert_called_once() + assert listener.running is False + + +class TestLoopPreventionScenarios: + """Integration tests for loop prevention scenarios.""" + + @pytest.mark.asyncio + async def test_full_loop_prevention_flow(self, listener): + """ + Test complete loop prevention flow: + 1. User edits page -> Processes + 2. Entity linking updates page (as automated user) -> Filtered + 3. Rapid duplicate edits -> Debounced + """ + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # 1. User edit - should process + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:100:user@example.com' + ) + assert mock_process.call_count == 1 + + mock_process.reset_mock() + + # 2. Automated edit (entity linking) - should be filtered + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:100:librarian@schweitz.net' + ) + assert mock_process.call_count == 0 + + # 3. Rapid duplicate from same user - should be debounced + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:100:user@example.com' + ) + assert mock_process.call_count == 0 # Debounced + + @pytest.mark.asyncio + async def test_different_pages_not_debounced(self, listener): + """Test that edits to different pages are not debounced.""" + mock_connection = AsyncMock() + + with patch.object(listener, '_process_page_change', new_callable=AsyncMock) as mock_process: + # Edit page 100 + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:100:user@example.com' + ) + assert mock_process.call_count == 1 + + # Edit page 200 immediately - should NOT be debounced + await listener._handle_notification( + mock_connection, 1234, 'wiki_page_changes', + 'UPDATE:200:user@example.com' + ) + assert mock_process.call_count == 2 + + +@pytest.mark.integration +class TestWikiChangeListenerIntegration: + """ + Integration tests (require actual PostgreSQL connection). + These tests are marked with @pytest.mark.integration and skipped by default. + Run with: pytest -m integration + """ + + @pytest.mark.asyncio + async def test_real_database_connection(self): + """Test connection to real PostgreSQL database (requires setup).""" + pytest.skip("Requires actual PostgreSQL setup with triggers") + + listener = WikiChangeListener() + try: + await listener.start() + assert listener.running is True + assert listener.connection is not None + finally: + await listener.stop() + + @pytest.mark.asyncio + async def test_real_notification_handling(self): + """Test handling real NOTIFY events from PostgreSQL.""" + pytest.skip("Requires actual PostgreSQL setup with triggers") + + # This would test actual pg_notify() calls from triggers + # and verify the listener receives and processes them