""" 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}}]) mock.execute_write = 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 the write transaction was used assert mock_neo4j.execute_write.called assert result.success is True # Find the document creation query calls = mock_neo4j.execute_write.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_write.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 assert mock_neo4j.execute_write.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 assert mock_neo4j.execute_write.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_write.called if __name__ == "__main__": pytest.main([__file__, "-v", "-s"])