250 lines
6.9 KiB
Python
250 lines
6.9 KiB
Python
"""
|
|
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"])
|