forked from cardosofelipe/fast-next-template
test(agents): add comprehensive API route tests
Add 22 tests for agents API covering: - CRUD operations (spawn, list, get, update, delete) - Lifecycle management (pause, resume) - Agent metrics (single and project-level) - Authorization and access control - Status filtering 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
619
backend/tests/api/routes/syndarix/test_agents.py
Normal file
619
backend/tests/api/routes/syndarix/test_agents.py
Normal file
@@ -0,0 +1,619 @@
|
||||
# tests/api/routes/syndarix/test_agents.py
|
||||
"""Tests for agent instance management endpoints.
|
||||
|
||||
Tests cover:
|
||||
- Agent instance CRUD operations
|
||||
- Agent lifecycle management (pause, resume)
|
||||
- Agent status filtering
|
||||
- Agent metrics
|
||||
- Authorization and access control
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from starlette import status
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_project(client, user_token):
|
||||
"""Create a test project for agent tests."""
|
||||
response = await client.post(
|
||||
"/api/v1/projects",
|
||||
json={
|
||||
"name": "Agent Test Project",
|
||||
"slug": "agent-test-project",
|
||||
"autonomy_level": "milestone",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
return response.json()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_agent_type(client, superuser_token):
|
||||
"""Create a test agent type for spawning agents."""
|
||||
import uuid as uuid_mod
|
||||
|
||||
unique_slug = f"test-developer-agent-{uuid_mod.uuid4().hex[:8]}"
|
||||
response = await client.post(
|
||||
"/api/v1/agent-types",
|
||||
json={
|
||||
"name": "Test Developer Agent",
|
||||
"slug": unique_slug,
|
||||
"expertise": ["python", "testing"],
|
||||
"primary_model": "claude-3-opus",
|
||||
"personality_prompt": "You are a helpful developer agent for testing.",
|
||||
"description": "A test developer agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {superuser_token}"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED, f"Failed: {response.json()}"
|
||||
return response.json()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSpawnAgent:
|
||||
"""Tests for POST /api/v1/projects/{project_id}/agents endpoint."""
|
||||
|
||||
async def test_spawn_agent_success(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test successfully spawning a new agent."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "My Developer Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
data = response.json()
|
||||
assert data["name"] == "My Developer Agent"
|
||||
assert data["status"] == "idle"
|
||||
assert data["project_id"] == project_id
|
||||
|
||||
async def test_spawn_agent_with_initial_memory(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test spawning agent with initial short-term memory."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Memory Agent",
|
||||
"short_term_memory": {"context": "test setup"},
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
data = response.json()
|
||||
assert data["short_term_memory"]["context"] == "test setup"
|
||||
|
||||
async def test_spawn_agent_nonexistent_project(
|
||||
self, client, user_token, test_agent_type
|
||||
):
|
||||
"""Test spawning agent in nonexistent project."""
|
||||
fake_project_id = str(uuid.uuid4())
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{fake_project_id}/agents",
|
||||
json={
|
||||
"project_id": fake_project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Orphan Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
async def test_spawn_agent_nonexistent_type(
|
||||
self, client, user_token, test_project
|
||||
):
|
||||
"""Test spawning agent with nonexistent agent type."""
|
||||
project_id = test_project["id"]
|
||||
fake_type_id = str(uuid.uuid4())
|
||||
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": fake_type_id,
|
||||
"name": "Invalid Type Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
async def test_spawn_agent_mismatched_project_id(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test spawning agent with mismatched project_id in body."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
different_project_id = str(uuid.uuid4())
|
||||
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": different_project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Mismatched Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestListAgents:
|
||||
"""Tests for GET /api/v1/projects/{project_id}/agents endpoint."""
|
||||
|
||||
async def test_list_agents_empty(self, client, user_token, test_project):
|
||||
"""Test listing agents when none exist."""
|
||||
project_id = test_project["id"]
|
||||
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["data"] == []
|
||||
assert data["pagination"]["total"] == 0
|
||||
|
||||
async def test_list_agents_with_data(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test listing agents with data."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agents
|
||||
await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Agent One",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Agent Two",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data["data"]) == 2
|
||||
assert data["pagination"]["total"] == 2
|
||||
|
||||
async def test_list_agents_filter_by_status(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test filtering agents by status."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Idle Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
# Filter by idle status
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents?status=idle",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert all(agent["status"] == "idle" for agent in data["data"])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestGetAgent:
|
||||
"""Tests for GET /api/v1/projects/{project_id}/agents/{agent_id} endpoint."""
|
||||
|
||||
async def test_get_agent_success(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test getting agent by ID."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Get Test Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Get agent
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["id"] == agent_id
|
||||
assert data["name"] == "Get Test Agent"
|
||||
|
||||
async def test_get_agent_not_found(self, client, user_token, test_project):
|
||||
"""Test getting a nonexistent agent."""
|
||||
project_id = test_project["id"]
|
||||
fake_agent_id = str(uuid.uuid4())
|
||||
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents/{fake_agent_id}",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestUpdateAgent:
|
||||
"""Tests for PATCH /api/v1/projects/{project_id}/agents/{agent_id} endpoint."""
|
||||
|
||||
async def test_update_agent_current_task(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test updating agent current_task."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Task Update Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Update current_task
|
||||
response = await client.patch(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}",
|
||||
json={"current_task": "Working on feature #123"},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["current_task"] == "Working on feature #123"
|
||||
|
||||
async def test_update_agent_memory(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test updating agent short-term memory."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Memory Update Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Update memory
|
||||
response = await client.patch(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}",
|
||||
json={"short_term_memory": {"last_context": "updated", "step": 2}},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["short_term_memory"]["last_context"] == "updated"
|
||||
assert data["short_term_memory"]["step"] == 2
|
||||
|
||||
async def test_update_agent_not_found(self, client, user_token, test_project):
|
||||
"""Test updating a nonexistent agent."""
|
||||
project_id = test_project["id"]
|
||||
fake_agent_id = str(uuid.uuid4())
|
||||
|
||||
response = await client.patch(
|
||||
f"/api/v1/projects/{project_id}/agents/{fake_agent_id}",
|
||||
json={"current_task": "Some task"},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAgentLifecycle:
|
||||
"""Tests for agent lifecycle management endpoints."""
|
||||
|
||||
async def test_pause_agent(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test pausing an agent."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Pause Test Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Pause agent
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}/pause",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["status"] == "paused"
|
||||
|
||||
async def test_resume_paused_agent(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test resuming a paused agent."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create and pause agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Resume Test Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Pause first
|
||||
await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}/pause",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
# Resume agent
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}/resume",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["status"] == "idle"
|
||||
|
||||
async def test_pause_nonexistent_agent(self, client, user_token, test_project):
|
||||
"""Test pausing a nonexistent agent."""
|
||||
project_id = test_project["id"]
|
||||
fake_agent_id = str(uuid.uuid4())
|
||||
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents/{fake_agent_id}/pause",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestDeleteAgent:
|
||||
"""Tests for DELETE /api/v1/projects/{project_id}/agents/{agent_id} endpoint."""
|
||||
|
||||
async def test_delete_agent_success(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test deleting an agent."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Delete Test Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Delete agent
|
||||
response = await client.delete(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["success"] is True
|
||||
|
||||
async def test_delete_agent_not_found(self, client, user_token, test_project):
|
||||
"""Test deleting a nonexistent agent."""
|
||||
project_id = test_project["id"]
|
||||
fake_agent_id = str(uuid.uuid4())
|
||||
|
||||
response = await client.delete(
|
||||
f"/api/v1/projects/{project_id}/agents/{fake_agent_id}",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAgentMetrics:
|
||||
"""Tests for agent metrics endpoints."""
|
||||
|
||||
async def test_get_agent_metrics(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test getting metrics for a single agent."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent
|
||||
create_response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Metrics Test Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
agent_id = create_response.json()["id"]
|
||||
|
||||
# Get metrics
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents/{agent_id}/metrics",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
# AgentInstanceMetrics schema
|
||||
assert "total_instances" in data
|
||||
assert "total_tasks_completed" in data
|
||||
assert "total_tokens_used" in data
|
||||
assert "total_cost_incurred" in data
|
||||
|
||||
async def test_get_project_agents_metrics(
|
||||
self, client, user_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test getting metrics for all agents in a project."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agents
|
||||
await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Metrics Agent 1",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Metrics Agent 2",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
# Get project-wide metrics
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{project_id}/agents/metrics",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAgentAuthorization:
|
||||
"""Tests for agent authorization."""
|
||||
|
||||
async def test_superuser_can_manage_any_project_agents(
|
||||
self, client, user_token, superuser_token, test_project, test_agent_type
|
||||
):
|
||||
"""Test that superuser can manage agents in any project."""
|
||||
project_id = test_project["id"]
|
||||
agent_type_id = test_agent_type["id"]
|
||||
|
||||
# Create agent as superuser in user's project
|
||||
response = await client.post(
|
||||
f"/api/v1/projects/{project_id}/agents",
|
||||
json={
|
||||
"project_id": project_id,
|
||||
"agent_type_id": agent_type_id,
|
||||
"name": "Superuser Created Agent",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {superuser_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
async def test_user_cannot_access_other_project_agents(
|
||||
self, client, user_token, superuser_token, test_agent_type
|
||||
):
|
||||
"""Test that user cannot access agents in another user's project."""
|
||||
# Create a project as superuser (not owned by regular user)
|
||||
project_response = await client.post(
|
||||
"/api/v1/projects",
|
||||
json={
|
||||
"name": "Other User Project",
|
||||
"slug": f"other-user-project-{uuid.uuid4().hex[:8]}",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {superuser_token}"},
|
||||
)
|
||||
other_project_id = project_response.json()["id"]
|
||||
|
||||
# Regular user tries to list agents - should fail
|
||||
response = await client.get(
|
||||
f"/api/v1/projects/{other_project_id}/agents",
|
||||
headers={"Authorization": f"Bearer {user_token}"},
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
Reference in New Issue
Block a user