Agent Zero
db97a39133
Check Cross-Plugin Imports / check (push) Has been cancelled
- P1-Tests: 12 test files with new cross-tenant isolation + RBAC tests - P3-Tests: 8 fixes (duplicate fixtures, sys.path.insert, unused imports, KeyError) - P3-Frontend: LucideIcons → ICON_MAP (2 files), inline styles → Tailwind (2 files) - P3-Frontend: DOMPurify for iframe XSS, redundant regex removed, console.log → console.debug - P2-Frontend: 2 notification API TODOs retained (requires larger refactor) - conftest.py: create_no_perm_user helper added - pyproject.toml: pythonpath for scripts/ added - All checks green: ruff 0, F821 0, tsc 0, app 495 routes, cross-plugin 0
229 lines
9.0 KiB
Python
229 lines
9.0 KiB
Python
"""Tasks plugin tests — CRUD, assign, status update, filtering."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
|
|
|
# Use tasks_client fixture (with Tasks plugin activated) instead of default client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestTaskList:
|
|
"""GET /api/v1/tasks"""
|
|
|
|
async def test_list_tasks_returns_200(self, tasks_client: AsyncClient, db_session):
|
|
"""GET /tasks returns 200 with paginated list."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
resp = await tasks_client.get("/api/v1/tasks", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
assert "page" in data
|
|
assert "page_size" in data
|
|
|
|
async def test_list_tasks_with_status_filter(self, tasks_client: AsyncClient, db_session):
|
|
"""GET /tasks?status=open filters by status."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
resp = await tasks_client.get("/api/v1/tasks?status=open", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
for item in resp.json()["items"]:
|
|
assert item["status"] == "open"
|
|
|
|
async def test_list_tasks_requires_auth(self, tasks_client: AsyncClient, db_session):
|
|
"""GET /tasks without auth returns 401."""
|
|
resp = await tasks_client.get("/api/v1/tasks", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestTaskCreate:
|
|
"""POST /api/v1/tasks"""
|
|
|
|
async def test_create_task_returns_201(self, tasks_client: AsyncClient, db_session):
|
|
"""POST /tasks creates a task and returns 201."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "Call customer", "priority": "high"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["title"] == "Call customer"
|
|
assert data["priority"] == "high"
|
|
assert data["status"] == "open"
|
|
|
|
async def test_create_task_with_due_date(self, tasks_client: AsyncClient, db_session):
|
|
"""POST /tasks with due_date stores it correctly."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "Follow up", "due_date": "2025-12-31T10:00:00Z"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["due_date"] is not None
|
|
|
|
async def test_create_task_empty_title_returns_422(self, tasks_client: AsyncClient, db_session):
|
|
"""POST /tasks with empty title returns 422."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": ""},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestTaskUpdate:
|
|
"""PATCH /api/v1/tasks/{id}"""
|
|
|
|
async def test_update_task_returns_200(self, tasks_client: AsyncClient, db_session):
|
|
"""PATCH /tasks/{id} updates the task."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
# Create
|
|
create_resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "Original"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
task_id = create_resp.json()["id"]
|
|
# Update
|
|
resp = await tasks_client.patch(
|
|
f"/api/v1/tasks/{task_id}",
|
|
json={"title": "Updated", "status": "in_progress"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["title"] == "Updated"
|
|
assert resp.json()["status"] == "in_progress"
|
|
|
|
async def test_update_task_not_found_returns_404(self, tasks_client: AsyncClient, db_session):
|
|
"""PATCH non-existent task returns 404."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
resp = await tasks_client.patch(
|
|
"/api/v1/tasks/00000000-0000-0000-0000-000000000000",
|
|
json={"title": "Updated"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestTaskStatus:
|
|
"""POST /api/v1/tasks/{id}/status"""
|
|
|
|
async def test_update_status_returns_200(self, tasks_client: AsyncClient, db_session):
|
|
"""POST /tasks/{id}/status updates status."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
create_resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "Task to complete"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
task_id = create_resp.json()["id"]
|
|
resp = await tasks_client.post(
|
|
f"/api/v1/tasks/{task_id}/status",
|
|
json={"status": "done"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "done"
|
|
|
|
async def test_update_status_invalid_returns_422(self, tasks_client: AsyncClient, db_session):
|
|
"""POST /tasks/{id}/status with invalid status returns 422."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
create_resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "Task"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
task_id = create_resp.json()["id"]
|
|
resp = await tasks_client.post(
|
|
f"/api/v1/tasks/{task_id}/status",
|
|
json={"status": "invalid"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestTaskDelete:
|
|
"""DELETE /api/v1/tasks/{id}"""
|
|
|
|
async def test_delete_task_returns_204(self, tasks_client: AsyncClient, db_session):
|
|
"""DELETE /tasks/{id} soft-deletes the task."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(tasks_client, "admin@tenanta.com")
|
|
create_resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "To delete"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
task_id = create_resp.json()["id"]
|
|
resp = await tasks_client.delete(f"/api/v1/tasks/{task_id}", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 204
|
|
# Verify it's gone from list
|
|
list_resp = await tasks_client.get("/api/v1/tasks", headers=ORIGIN_HEADER)
|
|
assert not any(t["id"] == task_id for t in list_resp.json()["items"])
|
|
|
|
|
|
# ── Cross-tenant isolation test ──
|
|
@pytest.mark.asyncio
|
|
class TestTaskCrossTenant:
|
|
"""Tasks must not leak across tenants."""
|
|
|
|
async def test_cross_tenant_isolation(self, tasks_app, db_session):
|
|
"""Task created in tenant A is not visible to tenant B."""
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
await seed_tenant_and_users(db_session)
|
|
transport = ASGITransport(app=tasks_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client_a:
|
|
await login_client(client_a, "admin@tenanta.com")
|
|
create_resp = await client_a.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "Tenant A Task"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert create_resp.status_code == 201
|
|
task_id = create_resp.json()["id"]
|
|
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client_b:
|
|
await login_client(client_b, "admin@tenantb.com")
|
|
# List must not contain tenant A's task
|
|
list_resp = await client_b.get("/api/v1/tasks", headers=ORIGIN_HEADER)
|
|
assert list_resp.status_code == 200
|
|
assert all(t["id"] != task_id for t in list_resp.json()["items"])
|
|
# Get must 404
|
|
get_resp = await client_b.get(f"/api/v1/tasks/{task_id}", headers=ORIGIN_HEADER)
|
|
assert get_resp.status_code == 404
|
|
|
|
async def test_rbac_no_permission(self, tasks_client: AsyncClient, db_session):
|
|
"""User without tasks:write permission gets 403 on create."""
|
|
from tests.conftest import create_no_perm_user
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await create_no_perm_user(db_session, seed)
|
|
await login_client(tasks_client, "noperm@tenanta.com")
|
|
resp = await tasks_client.post(
|
|
"/api/v1/tasks",
|
|
json={"title": "No Perm Task"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 403
|