Spaces:
Running
Running
| from __future__ import annotations | |
| from uuid import uuid4 | |
| import unittest | |
| from fastapi.testclient import TestClient | |
| from app.container import build_container | |
| from app.core.config import Settings | |
| from app.security.schemas import APIKeyCreate | |
| from app.security.service import APIKeyService | |
| from main import create_app | |
| def _security_settings(tmp_path): | |
| return Settings( | |
| _env_file=None, | |
| temp_dir=tmp_path / "temp", | |
| output_dir=tmp_path / "outputs", | |
| database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}", | |
| social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}", | |
| social_auto_migrate=True, | |
| social_worker_enabled=False, | |
| social_oauth_encryption_key="test-only-encryption-material", | |
| auth_enabled=True, | |
| auth_last_used_update_seconds=0, | |
| cleanup_interval_seconds=3600, | |
| whisper_model="tiny", | |
| max_workers=1, | |
| ) | |
| async def _create_app(tmp_path): | |
| settings = _security_settings(tmp_path) | |
| container = build_container(settings) | |
| await container.security_database.initialize() | |
| application = create_app(settings) | |
| return application, container | |
| async def _project_context(container, name, scopes): | |
| record, secret = await container.api_keys.create( | |
| APIKeyCreate(name=name, environment="test", role=None, scopes=scopes), | |
| created_by="tests", | |
| ) | |
| return record, secret, await container.api_keys.authenticate(secret) | |
| class ApprovalAuthorizationTests(unittest.TestCase): | |
| def test_list_approval_requests_requires_workflow_workspace_membership(self) -> None: | |
| import asyncio | |
| application, _ = asyncio.run(_create_app(self)) | |
| with TestClient(application) as client: | |
| unauthorized = client.get( | |
| "/v1/projects/workspace/workflows/other-workflow/requests", | |
| headers={"Authorization": "Bearer invalid"}, | |
| ) | |
| self.assertEqual(unauthorized.status_code, 401) | |
| def test_approve_and_reject_endpoints_authorize_by_request_workspace(self) -> None: | |
| import asyncio | |
| application, _ = asyncio.run(_create_app(self)) | |
| with TestClient(application) as client: | |
| missing = client.post( | |
| "/v1/projects/workspace/requests/missing/approve", | |
| headers={"Authorization": "Bearer invalid"}, | |
| ) | |
| self.assertEqual(missing.status_code, 404) | |
| def test_approval_workflow_requests_are_scoped_to_owner_workspace(self) -> None: | |
| import asyncio | |
| application, container = asyncio.run(_create_app(self)) | |
| with TestClient(application) as client: | |
| owner_key, _, actor_a = asyncio.run(_project_context(container, "Workspace A", [ | |
| "projects:read", | |
| "projects:create", | |
| "approvals:create", | |
| "approvals:read", | |
| "approvals:review", | |
| ])) | |
| _, _, actor_b = asyncio.run(_project_context(container, "Workspace B", [ | |
| "projects:read", | |
| "projects:create", | |
| "approvals:create", | |
| "approvals:read", | |
| "approvals:review", | |
| ])) | |
| project = client.post( | |
| "/v1/projects", | |
| json={"name": "Approval Project"}, | |
| headers={"Authorization": f"Bearer {owner_key}"}, | |
| ).json() | |
| workflow = client.post( | |
| "/v1/projects/workspace/workflows", | |
| json={"project_id": project["id"], "name": "Review"}, | |
| headers={"Authorization": f"Bearer {owner_key}"}, | |
| ).json() | |
| approval_request = client.post( | |
| f"/v1/projects/workspace/workflows/{workflow['id']}/requests", | |
| json={"project_id": project["id"]}, | |
| headers={"Authorization": f"Bearer {owner_key}"}, | |
| ).json() | |
| self.assertEqual( | |
| client.get( | |
| f"/v1/projects/workspace/workflows/{workflow['id']}/requests", | |
| headers={"Authorization": f"Bearer {actor_b.api_key_id}"}, | |
| ).status_code, | |
| 404, | |
| ) | |
| self.assertEqual( | |
| client.post( | |
| f"/v1/projects/workspace/requests/{approval_request['id']}/approve", | |
| headers={"Authorization": f"Bearer {actor_b.api_key_id}"}, | |
| ).status_code, | |
| 404, | |
| ) | |
| self.assertEqual( | |
| client.post( | |
| f"/v1/projects/workspace/requests/{approval_request['id']}/reject", | |
| headers={"Authorization": f"Bearer {actor_b.api_key_id}"}, | |
| ).status_code, | |
| 404, | |
| ) | |
| if __name__ == "__main__": | |
| unittest.main() | |