Agent Zero
7d976276ae
Check Cross-Plugin Imports / check (push) Has been cancelled
Tests (5 files, 126 tests, all passing): - test_agent_memory.py: 22 tests (store, retrieve, delete, routes, tenant isolation) - test_graph_rag.py: 22 tests (create, traverse BFS, bidirectional, max_hops, cycles, routes) - test_marketplace.py: 26 tests (fetch, download, verify, install, categories, routes) - test_agent_subtasks.py: 25 tests (create, wait, cancel, aggregate, list, model) - test_external_agent_api.py: 31 tests (run, status, stream, auth, rate limit) Bugfixes: - graph_rag/models.py: metadata -> meta (SQLAlchemy reserved attribute) - marketplace/routes.py: fix default parameter validation
893 lines
29 KiB
Python
893 lines
29 KiB
Python
"""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"
|