"""Tests for the GraphRAG plugin — service and route layers. Uses AsyncMock for all DB operations. No real DB connections required. """ from __future__ import annotations import uuid from datetime import UTC, datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from app.plugins.builtins.graph_rag.models import EntityRelationship from app.plugins.builtins.graph_rag.routes import router as graph_rag_router from app.plugins.builtins.graph_rag.services import ( create_relationship, delete_relationship, traverse_graph, ) # Override conftest DB fixtures — these tests use mocks, no real DB needed @pytest.fixture(autouse=True, scope="session") def db_setup(): """No-op override of conftest db_setup.""" yield @pytest.fixture(autouse=True) def clean_tables(db_setup): """No-op override of conftest clean_tables.""" yield @pytest.fixture(autouse=True) def _mock_check_permission(): """Patch check_permission to always return True for route tests.""" with patch("app.core.permissions.check_permission", return_value=True): yield # ─── Helpers ─── def _make_relationship( *, tenant_id: uuid.UUID | None = None, source_type: str = "contact", source_id: uuid.UUID | None = None, target_type: str = "company", target_id: uuid.UUID | None = None, relationship_type: str = "works_for", metadata: dict | None = None, owner_id: uuid.UUID | None = None, ) -> EntityRelationship: """Create an EntityRelationship instance with defaults.""" return EntityRelationship( id=uuid.uuid4(), tenant_id=tenant_id or uuid.uuid4(), source_type=source_type, source_id=source_id or uuid.uuid4(), target_type=target_type, target_id=target_id or uuid.uuid4(), relationship_type=relationship_type, meta=metadata or {}, owner_id=owner_id, created_at=datetime.now(UTC), updated_at=datetime.now(UTC), ) def _mock_session() -> AsyncMock: """Create a mock AsyncSession with common patterns.""" session = AsyncMock() session.add = MagicMock() session.flush = AsyncMock() session.refresh = AsyncMock() session.delete = AsyncMock() session.execute = AsyncMock() session.commit = AsyncMock() session.rollback = AsyncMock() return session # ─── Service-Layer Tests ─── class TestCreateRelationship: """Tests for create_relationship() service function.""" @pytest.mark.asyncio async def test_create_relationship_success(self): """create_relationship creates a new relationship.""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() target_id = uuid.uuid4() owner_id = uuid.uuid4() db = _mock_session() # No duplicate found dup_result = MagicMock() dup_result.scalar_one_or_none.return_value = None db.execute.return_value = dup_result def _refresh_side_effect(obj): obj.id = uuid.uuid4() obj.created_at = datetime.now(UTC) obj.updated_at = datetime.now(UTC) db.refresh.side_effect = _refresh_side_effect result = await create_relationship( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, target_type="company", target_id=target_id, relationship_type="works_for", metadata={"since": "2024"}, owner_id=owner_id, ) assert result["source_type"] == "contact" assert result["target_type"] == "company" assert result["relationship_type"] == "works_for" assert result["source_id"] == str(source_id) assert result["target_id"] == str(target_id) assert result["metadata"] == {"since": "2024"} assert result["owner_id"] == str(owner_id) assert "id" in result assert "created_at" in result db.add.assert_called_once() @pytest.mark.asyncio async def test_create_relationship_duplicate(self): """create_relationship returns error dict when duplicate exists.""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() target_id = uuid.uuid4() db = _mock_session() existing_rel = _make_relationship( tenant_id=tenant_id, source_id=source_id, target_id=target_id, relationship_type="works_for", ) dup_result = MagicMock() dup_result.scalar_one_or_none.return_value = existing_rel db.execute.return_value = dup_result result = await create_relationship( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, target_type="company", target_id=target_id, relationship_type="works_for", ) assert "error" in result assert result["code"] == "duplicate" db.add.assert_not_called() @pytest.mark.asyncio async def test_create_relationship_with_metadata(self): """create_relationship stores metadata dict.""" db = _mock_session() dup_result = MagicMock() dup_result.scalar_one_or_none.return_value = None db.execute.return_value = dup_result def _refresh_side_effect(obj): obj.id = uuid.uuid4() obj.created_at = datetime.now(UTC) db.refresh.side_effect = _refresh_side_effect result = await create_relationship( db=db, tenant_id=uuid.uuid4(), source_type="contact", source_id=uuid.uuid4(), target_type="email", target_id=uuid.uuid4(), relationship_type="has_email", metadata={"verified": True, "primary": True}, ) assert result["metadata"] == {"verified": True, "primary": True} @pytest.mark.asyncio async def test_create_relationship_no_metadata_defaults_to_empty(self): """create_relationship defaults metadata to empty dict when None.""" db = _mock_session() dup_result = MagicMock() dup_result.scalar_one_or_none.return_value = None db.execute.return_value = dup_result def _refresh_side_effect(obj): obj.id = uuid.uuid4() obj.created_at = datetime.now(UTC) db.refresh.side_effect = _refresh_side_effect result = await create_relationship( db=db, tenant_id=uuid.uuid4(), source_type="contact", source_id=uuid.uuid4(), target_type="task", target_id=uuid.uuid4(), relationship_type="assigned_to", metadata=None, ) assert result["metadata"] == {} class TestTraverseGraph: """Tests for traverse_graph() service function.""" @pytest.mark.asyncio async def test_traverse_single_node_no_relationships(self): """traverse_graph returns just the starting node when no relationships exist.""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() db = _mock_session() # No outgoing or incoming relationships empty_result = MagicMock() empty_result.scalars.return_value.all.return_value = [] db.execute.return_value = empty_result result = await traverse_graph( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, max_hops=3, ) assert result["total_nodes"] == 1 assert result["total_edges"] == 0 assert result["nodes"][0]["entity_type"] == "contact" assert result["nodes"][0]["entity_id"] == str(source_id) assert result["nodes"][0]["depth"] == 0 @pytest.mark.asyncio async def test_traverse_outgoing_relationships(self): """traverse_graph follows outgoing relationships.""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() target_id = uuid.uuid4() db = _mock_session() rel = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=source_id, target_type="company", target_id=target_id, relationship_type="works_for", ) rel_result = MagicMock() rel_result.scalars.return_value.all.return_value = [rel] empty_result = MagicMock() empty_result.scalars.return_value.all.return_value = [] # First call: outgoing relationships from source # Second call: incoming relationships to source (empty) # Third call: outgoing from target (empty) # Fourth call: incoming to target (empty) db.execute.side_effect = [rel_result, empty_result, empty_result, empty_result] result = await traverse_graph( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, max_hops=3, ) assert result["total_nodes"] == 2 # source + target assert result["total_edges"] == 1 # Check depth of target node target_node = [n for n in result["nodes"] if n["entity_type"] == "company"][0] assert target_node["depth"] == 1 assert target_node["path"] == ["works_for"] @pytest.mark.asyncio async def test_traverse_incoming_relationships(self): """traverse_graph follows incoming relationships (bidirectional).""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() other_source_id = uuid.uuid4() db = _mock_session() # An incoming relationship: other_source -> source incoming_rel = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=other_source_id, target_type="contact", target_id=source_id, relationship_type="knows", ) empty_result = MagicMock() empty_result.scalars.return_value.all.return_value = [] incoming_result = MagicMock() incoming_result.scalars.return_value.all.return_value = [incoming_rel] # First call: outgoing from source (empty) # Second call: incoming to source (has the rel) # Third call: outgoing from other_source (empty) # Fourth call: incoming to other_source (empty) db.execute.side_effect = [empty_result, incoming_result, empty_result, empty_result] result = await traverse_graph( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, max_hops=3, ) assert result["total_nodes"] == 2 # source + other_source assert result["total_edges"] == 1 # The other source should be found via incoming relationship other_node = [n for n in result["nodes"] if n["entity_id"] == str(other_source_id)][0] assert other_node["depth"] == 1 assert "inverse_knows" in other_node["path"] @pytest.mark.asyncio async def test_traverse_max_hops_limit(self): """traverse_graph respects max_hops limit.""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() mid_id = uuid.uuid4() far_id = uuid.uuid4() db = _mock_session() # source -> mid rel1 = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=source_id, target_type="contact", target_id=mid_id, relationship_type="knows", ) # mid -> far rel2 = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=mid_id, target_type="contact", target_id=far_id, relationship_type="knows", ) empty_result = MagicMock() empty_result.scalars.return_value.all.return_value = [] rel1_result = MagicMock() rel1_result.scalars.return_value.all.return_value = [rel1] rel2_result = MagicMock() rel2_result.scalars.return_value.all.return_value = [rel2] # With max_hops=1: only source and mid should be found # Call 1: outgoing from source -> [rel1] # Call 2: incoming to source -> [] # Call 3: outgoing from mid -> [rel2] (but depth=1 >= max_hops=1, so mid's neighbors not queued) # Actually: mid is at depth 1, and max_hops=1 means depth >= 1 stops, so we still query # but the far node won't be added because depth+1 > max_hops... wait, the code checks depth >= max_hops # before querying. So at depth 1 >= max_hops 1, it won't query for mid's neighbors. # So only 2 DB calls: outgoing from source, incoming to source db.execute.side_effect = [rel1_result, empty_result] result = await traverse_graph( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, max_hops=1, ) assert result["total_nodes"] == 2 # source + mid only # far_id should not be in nodes node_ids = [n["entity_id"] for n in result["nodes"]] assert str(far_id) not in node_ids @pytest.mark.asyncio async def test_traverse_relationship_types_filter(self): """traverse_graph filters by relationship_types when provided.""" tenant_id = uuid.uuid4() source_id = uuid.uuid4() target_id = uuid.uuid4() db = _mock_session() # Only relationships matching the filter should be returned rel = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=source_id, target_type="company", target_id=target_id, relationship_type="works_for", ) rel_result = MagicMock() rel_result.scalars.return_value.all.return_value = [rel] empty_result = MagicMock() empty_result.scalars.return_value.all.return_value = [] db.execute.side_effect = [rel_result, empty_result, empty_result, empty_result] result = await traverse_graph( db=db, tenant_id=tenant_id, source_type="contact", source_id=source_id, max_hops=3, relationship_types=["works_for"], ) assert result["total_nodes"] == 2 assert result["edges"][0]["relationship_type"] == "works_for" @pytest.mark.asyncio async def test_traverse_avoids_cycles(self): """traverse_graph does not revisit already-visited nodes.""" tenant_id = uuid.uuid4() node_a = uuid.uuid4() node_b = uuid.uuid4() db = _mock_session() # A -> B rel_a_to_b = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=node_a, target_type="contact", target_id=node_b, relationship_type="knows", ) # B -> A (cycle) rel_b_to_a = _make_relationship( tenant_id=tenant_id, source_type="contact", source_id=node_b, target_type="contact", target_id=node_a, relationship_type="knows", ) # From A: outgoing -> [rel_a_to_b], incoming -> [] # From B: outgoing -> [rel_b_to_a], incoming -> [] # B's outgoing has rel_b_to_a but A is already visited, so edge is added but node is not from_a_out = MagicMock() from_a_out.scalars.return_value.all.return_value = [rel_a_to_b] empty = MagicMock() empty.scalars.return_value.all.return_value = [] from_b_out = MagicMock() from_b_out.scalars.return_value.all.return_value = [rel_b_to_a] db.execute.side_effect = [from_a_out, empty, from_b_out, empty] result = await traverse_graph( db=db, tenant_id=tenant_id, source_type="contact", source_id=node_a, max_hops=3, ) # Only 2 nodes (A and B), but 2 edges (A->B and B->A) assert result["total_nodes"] == 2 assert result["total_edges"] == 2 class TestDeleteRelationship: """Tests for delete_relationship() service function.""" @pytest.mark.asyncio async def test_delete_relationship_success(self): """delete_relationship returns True when relationship exists.""" tenant_id = uuid.uuid4() rel_id = uuid.uuid4() db = _mock_session() rel = _make_relationship(tenant_id=tenant_id) mock_result = MagicMock() mock_result.scalar_one_or_none.return_value = rel db.execute.return_value = mock_result result = await delete_relationship(db, tenant_id, rel_id) assert result is True db.delete.assert_awaited_once_with(rel) @pytest.mark.asyncio async def test_delete_relationship_not_found(self): """delete_relationship returns False when relationship does not exist.""" db = _mock_session() mock_result = MagicMock() mock_result.scalar_one_or_none.return_value = None db.execute.return_value = mock_result result = await delete_relationship(db, uuid.uuid4(), uuid.uuid4()) assert result is False db.delete.assert_not_awaited() # ─── Model Tests ─── class TestEntityRelationshipModel: """Tests for the EntityRelationship model.""" def test_model_has_required_fields(self): """EntityRelationship has source_type, source_id, target_type, target_id, relationship_type, metadata.""" rel = EntityRelationship( tenant_id=uuid.uuid4(), source_type="contact", source_id=uuid.uuid4(), target_type="email", target_id=uuid.uuid4(), relationship_type="has_email", ) assert rel.source_type == "contact" assert rel.target_type == "email" assert rel.relationship_type == "has_email" def test_model_metadata_default(self): """EntityRelationship has default=dict for metadata column.""" col = EntityRelationship.__table__.c.metadata assert col.default.arg is dict or callable(col.default.arg) def test_model_table_name(self): """EntityRelationship uses correct table name.""" assert EntityRelationship.__tablename__ == "entity_relationships" # ─── Route-Layer Tests ─── def _create_graph_rag_app() -> FastAPI: """Create a minimal FastAPI app with graph_rag router and mocked dependencies.""" app = FastAPI() app.include_router(graph_rag_router) async def _mock_get_db(): db = _mock_session() yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(uuid.uuid4()), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user return app class TestGraphRAGRoutes: """Tests for GraphRAG API routes.""" @pytest.mark.asyncio async def test_create_relationship_route(self): """POST /api/v1/graph/relationships creates a relationship.""" app = _create_graph_rag_app() tenant_id = uuid.uuid4() source_id = uuid.uuid4() target_id = uuid.uuid4() user_id = uuid.uuid4() db = _mock_session() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(user_id), "tenant_id": str(tenant_id), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user # No duplicate dup_result = MagicMock() dup_result.scalar_one_or_none.return_value = None db.execute.return_value = dup_result def _refresh_side_effect(obj): obj.id = uuid.uuid4() obj.created_at = datetime.now(UTC) db.refresh.side_effect = _refresh_side_effect transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.post( "/api/v1/graph/relationships", json={ "source_type": "contact", "source_id": str(source_id), "target_type": "company", "target_id": str(target_id), "relationship_type": "works_for", "metadata": {"since": "2024"}, }, ) assert resp.status_code == 201 data = resp.json() assert data["relationship_type"] == "works_for" assert data["source_id"] == str(source_id) @pytest.mark.asyncio async def test_create_relationship_route_duplicate(self): """POST /api/v1/graph/relationships returns 409 for duplicate.""" app = _create_graph_rag_app() tenant_id = uuid.uuid4() db = _mock_session() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(tenant_id), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user existing = _make_relationship(tenant_id=tenant_id) dup_result = MagicMock() dup_result.scalar_one_or_none.return_value = existing db.execute.return_value = dup_result transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.post( "/api/v1/graph/relationships", json={ "source_type": "contact", "source_id": str(existing.source_id), "target_type": "company", "target_id": str(existing.target_id), "relationship_type": "works_for", }, ) assert resp.status_code == 409 assert resp.json()["detail"]["code"] == "duplicate" @pytest.mark.asyncio async def test_traverse_route(self): """POST /api/v1/graph/traverse returns traversal results.""" app = _create_graph_rag_app() source_id = uuid.uuid4() db = _mock_session() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(uuid.uuid4()), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user # No relationships found empty_result = MagicMock() empty_result.scalars.return_value.all.return_value = [] db.execute.return_value = empty_result transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.post( "/api/v1/graph/traverse", json={ "source_type": "contact", "source_id": str(source_id), "max_hops": 3, }, ) assert resp.status_code == 200 data = resp.json() assert data["total_nodes"] == 1 assert data["nodes"][0]["entity_id"] == str(source_id) @pytest.mark.asyncio async def test_delete_relationship_route_success(self): """DELETE /api/v1/graph/relationships/{id} returns 204 on success.""" app = _create_graph_rag_app() rel_id = uuid.uuid4() db = _mock_session() rel = _make_relationship() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(uuid.uuid4()), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user mock_result = MagicMock() mock_result.scalar_one_or_none.return_value = rel db.execute.return_value = mock_result transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.delete(f"/api/v1/graph/relationships/{rel_id}") assert resp.status_code == 204 @pytest.mark.asyncio async def test_delete_relationship_route_not_found(self): """DELETE /api/v1/graph/relationships/{id} returns 404 when not found.""" app = _create_graph_rag_app() rel_id = uuid.uuid4() db = _mock_session() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(uuid.uuid4()), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user mock_result = MagicMock() mock_result.scalar_one_or_none.return_value = None db.execute.return_value = mock_result transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.delete(f"/api/v1/graph/relationships/{rel_id}") assert resp.status_code == 404 @pytest.mark.asyncio async def test_list_relationships_route(self): """GET /api/v1/graph/relationships lists relationships with filters.""" app = _create_graph_rag_app() db = _mock_session() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(uuid.uuid4()), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user rel = _make_relationship(relationship_type="works_for") count_result = MagicMock() count_result.scalar_one.return_value = 1 list_result = MagicMock() list_result.scalars.return_value.all.return_value = [rel] db.execute.side_effect = [count_result, list_result] transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.get( "/api/v1/graph/relationships", params={"relationship_type": "works_for"}, ) assert resp.status_code == 200 data = resp.json() assert data["total"] == 1 assert data["items"][0]["relationship_type"] == "works_for" @pytest.mark.asyncio async def test_create_relationship_invalid_uuid(self): """POST /api/v1/graph/relationships returns 400 for invalid UUID.""" app = _create_graph_rag_app() db = _mock_session() async def _mock_get_db(): yield db async def _mock_get_current_user(): return { "user_id": str(uuid.uuid4()), "tenant_id": str(uuid.uuid4()), "is_system_admin": True, } from app.deps import get_current_user from app.core.db import get_db app.dependency_overrides[get_db] = _mock_get_db app.dependency_overrides[get_current_user] = _mock_get_current_user transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.post( "/api/v1/graph/relationships", json={ "source_type": "contact", "source_id": "not-a-uuid", "target_type": "company", "target_id": str(uuid.uuid4()), "relationship_type": "works_for", }, ) assert resp.status_code == 400 assert resp.json()["detail"]["code"] == "invalid_id"