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()