test(backend): add comprehensive tests for OAuth and agent endpoints

- Added tests for OAuth provider admin and consent endpoints covering edge cases.
- Extended agent-related tests to handle incorrect project associations and lifecycle state transitions.
- Introduced tests for sprint status transitions and validation checks.
- Improved multiline formatting consistency across all test functions.
This commit is contained in:
2026-01-03 01:44:11 +01:00
parent acd18ff694
commit 664415111a
28 changed files with 1530 additions and 216 deletions

View File

@@ -299,9 +299,7 @@ class TestListAgentTypes:
class TestGetAgentType:
"""Tests for GET /api/v1/agent-types/{agent_type_id} endpoint."""
async def test_get_agent_type_success(
self, client, user_token, test_agent_type
):
async def test_get_agent_type_success(self, client, user_token, test_agent_type):
"""Test successful retrieval of agent type by ID."""
agent_type_id = test_agent_type["id"]
@@ -383,7 +381,9 @@ class TestGetAgentTypeBySlug:
assert data["errors"][0]["code"] == "SYS_002" # NOT_FOUND
assert "non-existent-slug" in data["errors"][0]["message"]
async def test_get_agent_type_by_slug_unauthenticated(self, client, test_agent_type):
async def test_get_agent_type_by_slug_unauthenticated(
self, client, test_agent_type
):
"""Test that unauthenticated users cannot get agent types by slug."""
slug = test_agent_type["slug"]
@@ -671,9 +671,7 @@ class TestAgentTypeModelParams:
assert data["tool_permissions"]["read_files"] is True
assert data["tool_permissions"]["execute_code"] is False
async def test_update_model_params(
self, client, superuser_token, test_agent_type
):
async def test_update_model_params(self, client, superuser_token, test_agent_type):
"""Test updating model parameters."""
agent_type_id = test_agent_type["id"]
@@ -697,9 +695,7 @@ class TestAgentTypeModelParams:
class TestAgentTypeInstanceCount:
"""Tests for instance count tracking."""
async def test_new_agent_type_has_zero_instances(
self, client, superuser_token
):
async def test_new_agent_type_has_zero_instances(self, client, superuser_token):
"""Test that newly created agent types have zero instances."""
unique_slug = f"zero-instances-{uuid.uuid4().hex[:8]}"
response = await client.post(

View File

@@ -122,9 +122,7 @@ class TestSpawnAgent:
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_spawn_agent_nonexistent_type(
self, client, user_token, test_project
):
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())
@@ -376,9 +374,7 @@ class TestUpdateAgent:
class TestAgentLifecycle:
"""Tests for agent lifecycle management endpoints."""
async def test_pause_agent(
self, client, user_token, test_project, test_agent_type
):
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"]
@@ -617,3 +613,364 @@ class TestAgentAuthorization:
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.asyncio
class TestSpawnAgentEdgeCases:
"""Tests for agent spawn edge cases."""
async def test_spawn_agent_with_inactive_agent_type(
self, client, user_token, superuser_token, test_project
):
"""Test spawning agent with an inactive agent type fails."""
project_id = test_project["id"]
# Create an inactive agent type
unique_slug = f"inactive-agent-type-{uuid.uuid4().hex[:8]}"
create_response = await client.post(
"/api/v1/agent-types",
json={
"name": "Inactive Agent Type",
"slug": unique_slug,
"expertise": ["testing"],
"primary_model": "claude-3-opus",
"personality_prompt": "Test inactive agent.",
"description": "An inactive agent type for testing",
"is_active": False,
},
headers={"Authorization": f"Bearer {superuser_token}"},
)
assert create_response.status_code == status.HTTP_201_CREATED
inactive_type_id = create_response.json()["id"]
# Try to spawn agent with inactive type
response = await client.post(
f"/api/v1/projects/{project_id}/agents",
json={
"project_id": project_id,
"agent_type_id": inactive_type_id,
"name": "Agent With Inactive Type",
},
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
# Error response uses standardized format with "errors" list
data = response.json()
assert "errors" in data
assert any("inactive" in err["message"].lower() for err in data["errors"])
@pytest.mark.asyncio
class TestAgentWrongProject:
"""Tests for agent operations when agent belongs to different project."""
@pytest_asyncio.fixture
async def two_projects_with_agent(
self, client, user_token, superuser_token, test_agent_type
):
"""Create two projects and an agent in project1."""
# Create project1
resp1 = await client.post(
"/api/v1/projects",
json={
"name": "Project One",
"slug": f"project-one-{uuid.uuid4().hex[:8]}",
},
headers={"Authorization": f"Bearer {user_token}"},
)
project1 = resp1.json()
# Create project2
resp2 = await client.post(
"/api/v1/projects",
json={
"name": "Project Two",
"slug": f"project-two-{uuid.uuid4().hex[:8]}",
},
headers={"Authorization": f"Bearer {user_token}"},
)
project2 = resp2.json()
# Create agent in project1
agent_resp = await client.post(
f"/api/v1/projects/{project1['id']}/agents",
json={
"project_id": project1["id"],
"agent_type_id": test_agent_type["id"],
"name": "Project1 Agent",
},
headers={"Authorization": f"Bearer {user_token}"},
)
agent = agent_resp.json()
return {"project1": project1, "project2": project2, "agent": agent}
async def test_get_agent_wrong_project(
self, client, user_token, two_projects_with_agent
):
"""Test getting an agent via wrong project returns 404."""
data = two_projects_with_agent
agent_id = data["agent"]["id"]
wrong_project_id = data["project2"]["id"]
response = await client.get(
f"/api/v1/projects/{wrong_project_id}/agents/{agent_id}",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_update_agent_wrong_project(
self, client, user_token, two_projects_with_agent
):
"""Test updating an agent via wrong project returns 404."""
data = two_projects_with_agent
agent_id = data["agent"]["id"]
wrong_project_id = data["project2"]["id"]
response = await client.patch(
f"/api/v1/projects/{wrong_project_id}/agents/{agent_id}",
json={"current_task": "Test task"},
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_pause_agent_wrong_project(
self, client, user_token, two_projects_with_agent
):
"""Test pausing an agent via wrong project returns 404."""
data = two_projects_with_agent
agent_id = data["agent"]["id"]
wrong_project_id = data["project2"]["id"]
response = await client.post(
f"/api/v1/projects/{wrong_project_id}/agents/{agent_id}/pause",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_resume_agent_wrong_project(
self, client, user_token, two_projects_with_agent
):
"""Test resuming an agent via wrong project returns 404."""
data = two_projects_with_agent
project1_id = data["project1"]["id"]
agent_id = data["agent"]["id"]
wrong_project_id = data["project2"]["id"]
# First pause the agent using correct project
await client.post(
f"/api/v1/projects/{project1_id}/agents/{agent_id}/pause",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to resume via wrong project
response = await client.post(
f"/api/v1/projects/{wrong_project_id}/agents/{agent_id}/resume",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_terminate_agent_wrong_project(
self, client, user_token, two_projects_with_agent
):
"""Test terminating an agent via wrong project returns 404."""
data = two_projects_with_agent
agent_id = data["agent"]["id"]
wrong_project_id = data["project2"]["id"]
response = await client.delete(
f"/api/v1/projects/{wrong_project_id}/agents/{agent_id}",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_get_agent_metrics_wrong_project(
self, client, user_token, two_projects_with_agent
):
"""Test getting agent metrics via wrong project returns 404."""
data = two_projects_with_agent
agent_id = data["agent"]["id"]
wrong_project_id = data["project2"]["id"]
response = await client.get(
f"/api/v1/projects/{wrong_project_id}/agents/{agent_id}/metrics",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
class TestAgentStatusTransitions:
"""Tests for invalid agent status transitions."""
async def test_terminate_already_terminated_agent(
self, client, user_token, test_project, test_agent_type
):
"""Test terminating an already terminated agent fails."""
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": "Double Terminate Agent",
},
headers={"Authorization": f"Bearer {user_token}"},
)
agent_id = create_response.json()["id"]
# Terminate once
first_terminate = await client.delete(
f"/api/v1/projects/{project_id}/agents/{agent_id}",
headers={"Authorization": f"Bearer {user_token}"},
)
assert first_terminate.status_code == status.HTTP_200_OK
# Try to terminate again
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_422_UNPROCESSABLE_ENTITY
data = response.json()
assert "errors" in data
assert any("terminated" in err["message"].lower() for err in data["errors"])
async def test_resume_idle_agent(
self, client, user_token, test_project, test_agent_type
):
"""Test resuming an agent that is not paused fails."""
project_id = test_project["id"]
agent_type_id = test_agent_type["id"]
# Create agent (starts in idle state)
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 Idle Agent",
},
headers={"Authorization": f"Bearer {user_token}"},
)
agent_id = create_response.json()["id"]
# Try to resume without pausing first
response = await client.post(
f"/api/v1/projects/{project_id}/agents/{agent_id}/resume",
headers={"Authorization": f"Bearer {user_token}"},
)
# Should fail since agent is not paused
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_pause_already_paused_agent(
self, client, user_token, test_project, test_agent_type
):
"""Test pausing an already paused agent fails."""
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": "Double Pause Agent",
},
headers={"Authorization": f"Bearer {user_token}"},
)
agent_id = create_response.json()["id"]
# Pause once
first_pause = await client.post(
f"/api/v1/projects/{project_id}/agents/{agent_id}/pause",
headers={"Authorization": f"Bearer {user_token}"},
)
assert first_pause.status_code == status.HTTP_200_OK
# Try to pause again
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_422_UNPROCESSABLE_ENTITY
async def test_pause_terminated_agent(
self, client, user_token, test_project, test_agent_type
):
"""Test pausing a terminated agent fails."""
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 Terminated Agent",
},
headers={"Authorization": f"Bearer {user_token}"},
)
agent_id = create_response.json()["id"]
# Terminate agent
await client.delete(
f"/api/v1/projects/{project_id}/agents/{agent_id}",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to pause terminated 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_422_UNPROCESSABLE_ENTITY
async def test_resume_terminated_agent(
self, client, user_token, test_project, test_agent_type
):
"""Test resuming a terminated agent fails."""
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": "Resume Terminated Agent",
},
headers={"Authorization": f"Bearer {user_token}"},
)
agent_id = create_response.json()["id"]
# Terminate agent
await client.delete(
f"/api/v1/projects/{project_id}/agents/{agent_id}",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to resume terminated 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_422_UNPROCESSABLE_ENTITY

View File

@@ -74,7 +74,9 @@ async def terminated_agent(client, user_token, test_project, test_agent):
f"/api/v1/projects/{test_project['id']}/agents/{test_agent['id']}",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_200_OK, f"Failed to terminate: {response.json()}"
assert response.status_code == status.HTTP_200_OK, (
f"Failed to terminate: {response.json()}"
)
# Return agent info with terminated status
return {**test_agent, "status": "terminated"}
@@ -432,7 +434,7 @@ class TestProjectArchivingEdgeCases:
agent_id = test_agent["id"]
# Set agent to working status
status_response = await client.patch(
await client.patch(
f"/api/v1/projects/{project_id}/agents/{agent_id}/status",
json={"status": "working", "current_task": "Processing something"},
headers={"Authorization": f"Bearer {user_token}"},
@@ -475,7 +477,6 @@ class TestConcurrencyEdgeCases:
If two requests try to start sprints simultaneously, only one should succeed.
"""
import asyncio
from datetime import date, timedelta
project_id = test_project["id"]
@@ -509,7 +510,9 @@ class TestConcurrencyEdgeCases:
)
# Exactly one should succeed
successes = sum(1 for r in [start1, start2] if r.status_code == status.HTTP_200_OK)
successes = sum(
1 for r in [start1, start2] if r.status_code == status.HTTP_200_OK
)
failures = sum(1 for r in [start1, start2] if r.status_code in [400, 409, 422])
assert successes == 1, f"Expected exactly 1 success, got {successes}"
@@ -593,9 +596,7 @@ class TestDataIntegrityEdgeCases:
assert update_response.status_code == status.HTTP_404_NOT_FOUND
async def test_assign_issue_to_other_projects_sprint(
self, client, user_token
):
async def test_assign_issue_to_other_projects_sprint(self, client, user_token):
"""
IDOR Test: Try to assign an issue to a sprint from a different project.
"""
@@ -624,6 +625,7 @@ class TestDataIntegrityEdgeCases:
# Create a sprint in project 2
from datetime import date, timedelta
sprint_response = await client.post(
f"/api/v1/projects/{p2['id']}/sprints",
json={
@@ -662,7 +664,9 @@ class TestDataIntegrityEdgeCases:
status.HTTP_400_BAD_REQUEST,
status.HTTP_404_NOT_FOUND,
status.HTTP_422_UNPROCESSABLE_ENTITY,
], f"IDOR BUG: Assigned issue to another project's sprint! Status: {update_response.status_code}"
], (
f"IDOR BUG: Assigned issue to another project's sprint! Status: {update_response.status_code}"
)
async def test_assign_issue_to_other_projects_agent(
self, client, user_token, superuser_token
@@ -744,7 +748,9 @@ class TestDataIntegrityEdgeCases:
status.HTTP_400_BAD_REQUEST,
status.HTTP_404_NOT_FOUND,
status.HTTP_422_UNPROCESSABLE_ENTITY,
], f"IDOR BUG: Assigned issue to another project's agent! Status: {update_response.status_code}"
], (
f"IDOR BUG: Assigned issue to another project's agent! Status: {update_response.status_code}"
)
@pytest.mark.asyncio
@@ -1084,6 +1090,6 @@ class TestArchiveProjectCleanup:
# BUG CHECK: Sprint should be cancelled after project archive
if sprint_data.get("status") == "active":
pytest.fail(
f"BUG: Sprint status is still 'active' after project archive. "
f"Expected 'cancelled'. Archive should cancel active sprints."
"BUG: Sprint status is still 'active' after project archive. "
"Expected 'cancelled'. Archive should cancel active sprints."
)

View File

@@ -108,7 +108,9 @@ class TestCreateIssue:
assert "urgent" in data["labels"]
assert "frontend" in data["labels"]
async def test_create_issue_with_story_points(self, client, user_token, test_project):
async def test_create_issue_with_story_points(
self, client, user_token, test_project
):
"""Test creating issue with story points."""
project_id = test_project["id"]
@@ -237,7 +239,9 @@ class TestListIssues:
assert len(data["data"]) == 1
assert data["data"][0]["status"] == "open"
async def test_list_issues_filter_by_priority(self, client, user_token, test_project):
async def test_list_issues_filter_by_priority(
self, client, user_token, test_project
):
"""Test filtering issues by priority."""
project_id = test_project["id"]
@@ -703,7 +707,9 @@ class TestIssueAssignment:
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_assign_issue_clears_assignment(self, client, user_token, test_project):
async def test_assign_issue_clears_assignment(
self, client, user_token, test_project
):
"""Test that assigning to null clears both assignments."""
project_id = test_project["id"]
@@ -890,7 +896,9 @@ class TestIssueCrossProjectValidation:
class TestIssueValidation:
"""Tests for issue validation during create/update."""
async def test_create_issue_invalid_priority(self, client, user_token, test_project):
async def test_create_issue_invalid_priority(
self, client, user_token, test_project
):
"""Test creating issue with invalid priority."""
project_id = test_project["id"]
@@ -922,7 +930,9 @@ class TestIssueValidation:
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_update_issue_invalid_priority(self, client, user_token, test_project):
async def test_update_issue_invalid_priority(
self, client, user_token, test_project
):
"""Test updating issue with invalid priority."""
project_id = test_project["id"]

View File

@@ -243,14 +243,22 @@ class TestListProjects:
# Create active project
await client.post(
"/api/v1/projects",
json={"name": "Active Project", "slug": "active-project", "status": "active"},
json={
"name": "Active Project",
"slug": "active-project",
"status": "active",
},
headers={"Authorization": f"Bearer {user_token}"},
)
# Create paused project
await client.post(
"/api/v1/projects",
json={"name": "Paused Project", "slug": "paused-project", "status": "paused"},
json={
"name": "Paused Project",
"slug": "paused-project",
"status": "paused",
},
headers={"Authorization": f"Bearer {user_token}"},
)

View File

@@ -233,7 +233,9 @@ class TestListSprints:
assert len(data["data"]) == 3
assert data["pagination"]["total"] == 3
async def test_list_sprints_filter_by_status(self, client, user_token, test_project):
async def test_list_sprints_filter_by_status(
self, client, user_token, test_project
):
"""Test filtering sprints by status."""
project_id = test_project["id"]
start_date = date.today()
@@ -582,7 +584,9 @@ class TestSprintLifecycle:
class TestDeleteSprint:
"""Tests for DELETE /api/v1/projects/{project_id}/sprints/{sprint_id} endpoint."""
async def test_delete_planned_sprint_success(self, client, user_token, test_project):
async def test_delete_planned_sprint_success(
self, client, user_token, test_project
):
"""Test deleting a planned sprint."""
project_id = test_project["id"]
start_date = date.today()
@@ -1119,3 +1123,419 @@ class TestSprintCrossProjectValidation:
)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
class TestSprintStatusTransitions:
"""Tests for invalid sprint status transitions."""
async def test_cancel_completed_sprint(self, client, user_token, test_project):
"""Test that cancelling a completed sprint fails."""
project_id = test_project["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create, start, and complete sprint
create_response = await client.post(
f"/api/v1/projects/{project_id}/sprints",
json={
"project_id": project_id,
"name": "Sprint to Complete Then Cancel",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = create_response.json()["id"]
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/start",
headers={"Authorization": f"Bearer {user_token}"},
)
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/complete",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to cancel completed sprint
response = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/cancel",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_cancel_already_cancelled_sprint(
self, client, user_token, test_project
):
"""Test that cancelling an already cancelled sprint fails."""
project_id = test_project["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create and cancel sprint
create_response = await client.post(
f"/api/v1/projects/{project_id}/sprints",
json={
"project_id": project_id,
"name": "Double Cancel Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = create_response.json()["id"]
# Cancel once
first_cancel = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/cancel",
headers={"Authorization": f"Bearer {user_token}"},
)
assert first_cancel.status_code == status.HTTP_200_OK
# Try to cancel again
response = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/cancel",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_complete_already_completed_sprint(
self, client, user_token, test_project
):
"""Test that completing an already completed sprint fails."""
project_id = test_project["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create, start, and complete sprint
create_response = await client.post(
f"/api/v1/projects/{project_id}/sprints",
json={
"project_id": project_id,
"name": "Double Complete Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = create_response.json()["id"]
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/start",
headers={"Authorization": f"Bearer {user_token}"},
)
# Complete once
first_complete = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/complete",
headers={"Authorization": f"Bearer {user_token}"},
)
assert first_complete.status_code == status.HTTP_200_OK
# Try to complete again
response = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/complete",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_complete_cancelled_sprint(self, client, user_token, test_project):
"""Test that completing a cancelled sprint fails."""
project_id = test_project["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create and cancel sprint
create_response = await client.post(
f"/api/v1/projects/{project_id}/sprints",
json={
"project_id": project_id,
"name": "Complete Cancelled Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = create_response.json()["id"]
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/cancel",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to complete cancelled sprint
response = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/complete",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_start_cancelled_sprint(self, client, user_token, test_project):
"""Test that starting a cancelled sprint fails."""
project_id = test_project["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create and cancel sprint
create_response = await client.post(
f"/api/v1/projects/{project_id}/sprints",
json={
"project_id": project_id,
"name": "Start Cancelled Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = create_response.json()["id"]
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/cancel",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to start cancelled sprint
response = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/start",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_start_completed_sprint(self, client, user_token, test_project):
"""Test that starting a completed sprint fails."""
project_id = test_project["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create, start, and complete sprint
create_response = await client.post(
f"/api/v1/projects/{project_id}/sprints",
json={
"project_id": project_id,
"name": "Start Completed Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = create_response.json()["id"]
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/start",
headers={"Authorization": f"Bearer {user_token}"},
)
await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/complete",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to start completed sprint
response = await client.post(
f"/api/v1/projects/{project_id}/sprints/{sprint_id}/start",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
class TestSprintWrongProject:
"""Tests for sprint operations when sprint belongs to different project."""
async def test_complete_sprint_wrong_project(self, client, user_token):
"""Test completing a sprint via wrong project returns 404."""
# Create two projects
project1 = await client.post(
"/api/v1/projects",
json={"name": "Complete P1", "slug": f"complete-p1-{uuid.uuid4().hex[:6]}"},
headers={"Authorization": f"Bearer {user_token}"},
)
project2 = await client.post(
"/api/v1/projects",
json={"name": "Complete P2", "slug": f"complete-p2-{uuid.uuid4().hex[:6]}"},
headers={"Authorization": f"Bearer {user_token}"},
)
project1_id = project1.json()["id"]
project2_id = project2.json()["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create and start sprint in project1
sprint_response = await client.post(
f"/api/v1/projects/{project1_id}/sprints",
json={
"project_id": project1_id,
"name": "Complete Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = sprint_response.json()["id"]
await client.post(
f"/api/v1/projects/{project1_id}/sprints/{sprint_id}/start",
headers={"Authorization": f"Bearer {user_token}"},
)
# Try to complete via wrong project
response = await client.post(
f"/api/v1/projects/{project2_id}/sprints/{sprint_id}/complete",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_cancel_sprint_wrong_project(self, client, user_token):
"""Test cancelling a sprint via wrong project returns 404."""
# Create two projects
project1 = await client.post(
"/api/v1/projects",
json={"name": "Cancel P1", "slug": f"cancel-p1-{uuid.uuid4().hex[:6]}"},
headers={"Authorization": f"Bearer {user_token}"},
)
project2 = await client.post(
"/api/v1/projects",
json={"name": "Cancel P2", "slug": f"cancel-p2-{uuid.uuid4().hex[:6]}"},
headers={"Authorization": f"Bearer {user_token}"},
)
project1_id = project1.json()["id"]
project2_id = project2.json()["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create sprint in project1
sprint_response = await client.post(
f"/api/v1/projects/{project1_id}/sprints",
json={
"project_id": project1_id,
"name": "Cancel Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = sprint_response.json()["id"]
# Try to cancel via wrong project
response = await client.post(
f"/api/v1/projects/{project2_id}/sprints/{sprint_id}/cancel",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_delete_sprint_wrong_project(self, client, user_token):
"""Test deleting a sprint via wrong project returns 404."""
# Create two projects
project1 = await client.post(
"/api/v1/projects",
json={"name": "Delete P1", "slug": f"delete-p1-{uuid.uuid4().hex[:6]}"},
headers={"Authorization": f"Bearer {user_token}"},
)
project2 = await client.post(
"/api/v1/projects",
json={"name": "Delete P2", "slug": f"delete-p2-{uuid.uuid4().hex[:6]}"},
headers={"Authorization": f"Bearer {user_token}"},
)
project1_id = project1.json()["id"]
project2_id = project2.json()["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create sprint in project1
sprint_response = await client.post(
f"/api/v1/projects/{project1_id}/sprints",
json={
"project_id": project1_id,
"name": "Delete Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = sprint_response.json()["id"]
# Try to delete via wrong project
response = await client.delete(
f"/api/v1/projects/{project2_id}/sprints/{sprint_id}",
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
async def test_add_issue_to_sprint_wrong_project(self, client, user_token):
"""Test adding issue to sprint via wrong project returns 404."""
# Create two projects
project1 = await client.post(
"/api/v1/projects",
json={
"name": "Add Issue P1",
"slug": f"add-issue-p1-{uuid.uuid4().hex[:6]}",
},
headers={"Authorization": f"Bearer {user_token}"},
)
project2 = await client.post(
"/api/v1/projects",
json={
"name": "Add Issue P2",
"slug": f"add-issue-p2-{uuid.uuid4().hex[:6]}",
},
headers={"Authorization": f"Bearer {user_token}"},
)
project1_id = project1.json()["id"]
project2_id = project2.json()["id"]
start_date = date.today()
end_date = start_date + timedelta(days=14)
# Create sprint in project1
sprint_response = await client.post(
f"/api/v1/projects/{project1_id}/sprints",
json={
"project_id": project1_id,
"name": "Add Issue Sprint",
"number": 1,
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat(),
},
headers={"Authorization": f"Bearer {user_token}"},
)
sprint_id = sprint_response.json()["id"]
# Create issue in project1
issue_response = await client.post(
f"/api/v1/projects/{project1_id}/issues",
json={
"project_id": project1_id,
"title": "Test Issue",
},
headers={"Authorization": f"Bearer {user_token}"},
)
issue_id = issue_response.json()["id"]
# Try to add issue via wrong project
response = await client.post(
f"/api/v1/projects/{project2_id}/sprints/{sprint_id}/issues",
params={"issue_id": issue_id},
headers={"Authorization": f"Bearer {user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND