from __future__ import annotations from pathlib import Path from uuid import uuid4 import pytest from fastapi.testclient import TestClient from sqlalchemy import select from app.container import build_container from app.core.config import Settings from app.generation.models import GenerationJob, GenerationRequest from app.projects.errors import ( ProjectAlreadyArchivedError, ProjectAssetConflictError, ProjectAssetNotFoundError, ProjectJobNotFoundError, ProjectNotFoundError, ProjectThumbnailInvalidError, ) from app.projects.schemas import ProjectCreate, ProjectStatus, ProjectUpdate from app.security.models import AuditEvent from app.security.policy import ScopePolicy from app.security.schemas import APIKeyCreate from app.security.service import APIKeyService from main import create_app def project_settings(tmp_path: Path, **overrides: object) -> Settings: values: dict[str, object] = { "_env_file": None, "auth_enabled": True, "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", "temp_dir": tmp_path / "temp", "output_dir": tmp_path / "outputs", "cleanup_interval_seconds": 3600, "whisper_model": "tiny", "auth_default_requests_per_minute": 10_000, } values.update(overrides) return Settings(**values) async def _project_context(container: object, name: str, scopes: list[str]): record, secret = await container.api_keys.create( # type: ignore[attr-defined] APIKeyCreate(name=name, environment="test", role=None, scopes=scopes), created_by="tests", ) return record, secret, await container.api_keys.authenticate(secret) # type: ignore[attr-defined] @pytest.mark.asyncio async def test_project_service_lifecycle_isolation_pagination_thumbnail_and_audit( tmp_path: Path, ) -> None: container = build_container(project_settings(tmp_path)) await container.security_database.initialize() scopes = [ "projects:read", "projects:create", "projects:update", "projects:delete", ] try: key_a, _, actor_a = await _project_context(container, "Workspace A", scopes) _, _, actor_b = await _project_context(container, "Workspace B", scopes) assert actor_a.workspace_id != actor_b.workspace_id output_id = str(uuid4()) output = container.settings.output_dir / output_id output.mkdir(parents=True) thumbnail_path = output / "thumbnail.png" thumbnail_path.write_bytes(b"canonical thumbnail") thumbnail = await container.assets.register_output( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), request_id=output_id, path=thumbnail_path, mime_type="image/png", ) first = await container.projects.create( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), payload=ProjectCreate( name=" Podcast Episode 41 ", description="Primary project", thumbnail_asset_id=thumbnail.id, metadata={"aspect_ratio": "16:9"}, ), ) assert first.name == "Podcast Episode 41" assert first.status is ProjectStatus.ACTIVE assert first.workspace_id == actor_a.workspace_id assert first.created_by == actor_a.user_id assert first.thumbnail_asset_id == thumbnail.id with pytest.raises(ProjectNotFoundError): await container.projects.get( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), project_id=first.id, ) with pytest.raises(ProjectNotFoundError): await container.projects.update( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), api_key_id=actor_b.api_key_id, request_id=str(uuid4()), project_id=first.id, payload=ProjectUpdate(name="IDOR update"), ) with pytest.raises(ProjectNotFoundError): await container.projects.delete( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), api_key_id=actor_b.api_key_id, request_id=str(uuid4()), project_id=first.id, ) with pytest.raises(ProjectThumbnailInvalidError): await container.projects.create( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), api_key_id=actor_b.api_key_id, request_id=str(uuid4()), payload=ProjectCreate(name="Foreign thumbnail", thumbnail_asset_id=thumbnail.id), ) second = await container.projects.create( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), payload=ProjectCreate(name="Second project"), ) third = await container.projects.create( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), payload=ProjectCreate(name="Third project"), ) page_one = await container.projects.list( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), status=ProjectStatus.ACTIVE, search=None, limit=2, cursor=None, ) assert len(page_one.items) == 2 assert page_one.next_cursor page_two = await container.projects.list( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), status=ProjectStatus.ACTIVE, search=None, limit=2, cursor=page_one.next_cursor, ) assert len(page_two.items) == 1 assert {item.id for item in page_one.items + page_two.items} == { first.id, second.id, third.id, } searched = await container.projects.list( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), status=ProjectStatus.ACTIVE, search="podcast", limit=10, cursor=None, ) assert [item.id for item in searched.items] == [first.id] updated = await container.projects.update( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=first.id, payload=ProjectUpdate(description="Updated description"), ) assert updated.description == "Updated description" archived = await container.projects.update( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=first.id, payload=ProjectUpdate(status=ProjectStatus.ARCHIVED), ) assert archived.status is ProjectStatus.ARCHIVED assert archived.archived_at is not None with pytest.raises(ProjectAlreadyArchivedError): await container.projects.update( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=first.id, payload=ProjectUpdate(name="Blocked"), ) await container.projects.delete( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=second.id, ) async with container.security_database.tenant_session( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id) ) as session: events = list( ( await session.scalars( select(AuditEvent).where( AuditEvent.workspace_id == actor_a.workspace_id, AuditEvent.entity_type == "project", ) ) ).all() ) assert {event.event_type for event in events} >= { "project.created", "project.updated", "project.archived", "project.deleted", } assert all("aspect_ratio" not in event.metadata_json for event in events) finally: await container.security_database.close() @pytest.mark.asyncio async def test_project_resource_service_ownership_lifecycle_and_audit( tmp_path: Path, ) -> None: container = build_container(project_settings(tmp_path)) await container.security_database.initialize() scopes = ["projects:read", "projects:create", "projects:update"] try: key_a, _, actor_a = await _project_context(container, "Resource A", scopes) key_b, _, actor_b = await _project_context(container, "Resource B", scopes) project_a = await container.projects.create( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), payload=ProjectCreate(name="Project A"), ) second_a = await container.projects.create( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), payload=ProjectCreate(name="Second A"), ) project_b = await container.projects.create( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), api_key_id=key_b.id, request_id=str(uuid4()), payload=ProjectCreate(name="Project B"), ) async def canonical_asset(actor: object, name: str): request_id = str(uuid4()) directory = container.settings.output_dir / request_id directory.mkdir(parents=True) path = directory / name path.write_bytes(name.encode()) return await container.assets.register_output( workspace_id=str(actor.workspace_id), # type: ignore[attr-defined] user_id=str(actor.user_id), # type: ignore[attr-defined] request_id=request_id, path=path, mime_type="video/mp4", ) asset_a = await canonical_asset(actor_a, "asset-a.mp4") asset_b = await canonical_asset(actor_b, "asset-b.mp4") attached = await container.projects.attach_asset( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, asset_id=asset_a.id, ) assert attached.project_id == project_a.id assert [ item.id for item in ( await container.projects.list_assets( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), project_id=project_a.id, ) ).items ] == [asset_a.id] with pytest.raises(ProjectAssetConflictError): await container.projects.attach_asset( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=second_a.id, asset_id=asset_a.id, ) with pytest.raises(ProjectAssetNotFoundError): await container.projects.attach_asset( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, asset_id=asset_b.id, ) with pytest.raises(ProjectNotFoundError): await container.projects.list_assets( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), project_id=project_a.id, ) generation_request_a = GenerationRequest( workspace_id=str(actor_a.workspace_id), created_by_user_id=str(actor_a.user_id), provider="test", model_id="test/model", modality="video", spec_json={}, request_fingerprint="a" * 64, idempotency_key=str(uuid4()), status="queued", ) generation_request_b = GenerationRequest( workspace_id=str(actor_b.workspace_id), created_by_user_id=str(actor_b.user_id), provider="test", model_id="test/model", modality="video", spec_json={}, request_fingerprint="b" * 64, idempotency_key=str(uuid4()), status="queued", ) async with container.security_database.session() as session: session.add_all([generation_request_a, generation_request_b]) await session.flush() job_a = GenerationJob( generation_request_id=generation_request_a.id, workspace_id=str(actor_a.workspace_id), provider="test", status="queued", ) job_b = GenerationJob( generation_request_id=generation_request_b.id, workspace_id=str(actor_b.workspace_id), provider="test", status="queued", ) session.add_all([job_a, job_b]) await session.commit() linked_job = await container.projects.attach_generation_job( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, generation_job_id=job_a.id, ) assert linked_job.id == job_a.id assert [ item.id for item in ( await container.projects.list_generation_jobs( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), project_id=project_a.id, ) ).items ] == [job_a.id] with pytest.raises(ProjectJobNotFoundError): await container.projects.attach_generation_job( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, generation_job_id=job_b.id, ) with pytest.raises(ProjectNotFoundError): await container.projects.list_generation_jobs( workspace_id=str(actor_b.workspace_id), user_id=str(actor_b.user_id), project_id=project_a.id, ) await container.projects.detach_generation_job( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, generation_job_id=job_a.id, ) await container.projects.detach_asset( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, asset_id=asset_a.id, ) await container.projects.delete( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, ) with pytest.raises(ProjectAlreadyArchivedError): await container.projects.attach_asset( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id), api_key_id=key_a.id, request_id=str(uuid4()), project_id=project_a.id, asset_id=asset_a.id, ) async with container.security_database.tenant_session( workspace_id=str(actor_a.workspace_id), user_id=str(actor_a.user_id) ) as session: events = list( ( await session.scalars( select(AuditEvent).where( AuditEvent.workspace_id == actor_a.workspace_id, AuditEvent.entity_id == project_a.id, ) ) ).all() ) assert {event.event_type for event in events} >= { "project.asset_attached", "project.asset_detached", "project.job_attached", "project.job_detached", } assert all( set(event.metadata_json) <= {"resource_id", "disposition", "has_thumbnail"} for event in events ) assert project_b.workspace_id == actor_b.workspace_id finally: await container.security_database.close() def _create_key(client: TestClient, admin_headers: dict[str, str], scopes: list[str]) -> str: response = client.post( "/v1/api-keys", headers=admin_headers, json={ "name": f"Project key {uuid4()}", "environment": "test", "role": None, "scopes": scopes, }, ) assert response.status_code == 201 return response.json()["api_key"] def test_project_http_permissions_openapi_validation_and_archive(tmp_path: Path) -> None: material = APIKeyService.generate_material("test") settings = project_settings( tmp_path, auth_bootstrap_key_hash=material.key_hash, auth_bootstrap_key_prefix=material.key_prefix, auth_bootstrap_environment="test", ) admin_headers = {"Authorization": f"Bearer {material.api_key}"} with TestClient(create_app(settings)) as client: schema = client.get("/openapi.json").json() assert set(schema["paths"]["/v1/projects"]) >= {"get", "post"} assert set(schema["paths"]["/v1/projects/{project_id}"]) >= { "get", "patch", "delete", } assert set(schema["paths"]["/v1/projects/{project_id}/assets"]) >= {"get", "post"} assert "delete" in schema["paths"]["/v1/projects/{project_id}/assets/{asset_id}"] assert set(schema["paths"]["/v1/projects/{project_id}/jobs"]) >= {"get", "post"} assert "delete" in schema["paths"]["/v1/projects/{project_id}/jobs/{job_id}"] assert set(schema["paths"]["/v1/projects/{project_id}/editor"]) >= {"get", "put"} assert set(schema["paths"]["/v1/projects/{project_id}/renders"]) >= {"get", "post"} assert "get" in schema["paths"]["/v1/projects/{project_id}/renders/{render_id}"] assert "post" in schema["paths"]["/v1/projects/{project_id}/renders/{render_id}/cancel"] assert "ProjectCreate" in schema["components"]["schemas"] assert "ProjectUpdate" in schema["components"]["schemas"] assert "ProjectResponse" in schema["components"]["schemas"] assert "ProjectListResponse" in schema["components"]["schemas"] assert "ProjectAssetResponse" in schema["components"]["schemas"] assert "ProjectGenerationJobResponse" in schema["components"]["schemas"] assert "EditorSaveRequest" in schema["components"]["schemas"] assert "EditorStateResponse" in schema["components"]["schemas"] assert "ProjectRenderCreate" in schema["components"]["schemas"] assert "ProjectRenderResponse" in schema["components"]["schemas"] assert client.get("/v1/projects").status_code == 401 read_secret = _create_key(client, admin_headers, ["projects:read"]) create_secret = _create_key(client, admin_headers, ["projects:create"]) update_secret = _create_key(client, admin_headers, ["projects:update"]) delete_secret = _create_key(client, admin_headers, ["projects:delete"]) jobs_secret = _create_key(client, admin_headers, ["jobs:create", "jobs:cancel"]) render_secret = _create_key( client, admin_headers, ["projects:update", "jobs:create", "jobs:cancel"], ) read_headers = {"Authorization": f"Bearer {read_secret}"} create_headers = {"Authorization": f"Bearer {create_secret}"} update_headers = {"Authorization": f"Bearer {update_secret}"} delete_headers = {"Authorization": f"Bearer {delete_secret}"} jobs_headers = {"Authorization": f"Bearer {jobs_secret}"} render_headers = {"Authorization": f"Bearer {render_secret}"} assert client.get("/v1/projects", headers=read_headers).status_code == 200 assert ( client.post("/v1/projects", headers=read_headers, json={"name": "No"}).status_code == 403 ) created = client.post( "/v1/projects", headers=create_headers, json={"name": " HTTP Project ", "metadata": {"source": "test"}}, ) assert created.status_code == 201 project = created.json() assert project["name"] == "HTTP Project" assert "workspace_id" not in created.request.content.decode() project_id = project["id"] assert client.get(f"/v1/projects/{project_id}", headers=create_headers).status_code == 403 assert client.get(f"/v1/projects/{project_id}", headers=read_headers).status_code == 200 assert ( client.patch( f"/v1/projects/{project_id}", headers=update_headers, json={"description": "Changed"}, ).status_code == 200 ) empty_editor = { "schemaVersion": 1, "projectId": project_id, "timeline": { "timeUnit": "milliseconds", "tracks": [], "transitions": [], "markers": [], }, "renderSettings": { "format": "mp4", "width": 1280, "height": 720, "frameRate": 30, }, } saved_editor = client.put( f"/v1/projects/{project_id}/editor", headers=update_headers, json={"expected_revision": 0, "schema_version": 1, "state": empty_editor}, ) assert saved_editor.status_code == 200 assert saved_editor.json()["revision"] == 1 assert ( client.get(f"/v1/projects/{project_id}/editor", headers=read_headers).status_code == 200 ) render_payload = { "editor_revision": 1, "output_format": "mp4", "width": 1280, "height": 720, "frame_rate": 30, "quality": "standard", "preset": "balanced", } render_path = f"/v1/projects/{project_id}/renders" assert ( client.post( render_path, headers={**update_headers, "Idempotency-Key": "missing-jobs-scope"}, json=render_payload, ).status_code == 403 ) assert ( client.post( render_path, headers={**jobs_headers, "Idempotency-Key": "missing-project-scope"}, json=render_payload, ).status_code == 403 ) render_rejected = client.post( render_path, headers={**render_headers, "Idempotency-Key": "empty-editor"}, json=render_payload, ) assert render_rejected.status_code == 422 assert render_rejected.json()["error"]["code"] == "PROJECT_RENDER_INVALID" cancel_path = f"{render_path}/{uuid4()}/cancel" assert client.post(cancel_path, headers=update_headers).status_code == 403 assert client.post(cancel_path, headers=render_headers).status_code == 404 assert ( client.delete(f"/v1/projects/{project_id}", headers=update_headers).status_code == 403 ) assert ( client.delete(f"/v1/projects/{project_id}", headers=delete_headers).status_code == 204 ) archived = client.get(f"/v1/projects/{project_id}", headers=read_headers) assert archived.status_code == 200 assert archived.json()["status"] == "archived" assert archived.json()["archived_at"] is not None invalid_name = client.post("/v1/projects", headers=admin_headers, json={"name": " "}) assert invalid_name.status_code == 422 extra_system_field = client.post( "/v1/projects", headers=admin_headers, json={"name": "Unsafe", "workspace_id": str(uuid4())}, ) assert extra_system_field.status_code == 422 oversized_metadata = client.post( "/v1/projects", headers=admin_headers, json={"name": "Large", "metadata": {"value": "x" * 20_000}}, ) assert oversized_metadata.status_code == 422 invalid_cursor = client.get("/v1/projects?cursor=not-a-cursor", headers=admin_headers) assert invalid_cursor.status_code == 422 assert invalid_cursor.json()["error"]["code"] == "PROJECT_INVALID_CURSOR" def test_project_migration_is_additive_and_contains_security_guards() -> None: migration = ( ( Path(__file__).resolve().parents[1] / "app/projects/migrations/0001_projects_foundation.sql" ) .read_text(encoding="utf-8") .casefold() ) for expected in ( "create table if not exists projects", "check (status in ('active', 'archived'))", "ix_projects_workspace_status", "enable row level security", "force row level security", "create policy projects_select", "create policy projects_insert", "create policy projects_update", "create policy projects_delete", "mediarouter_assert_project_ownership", "audit_events", ): assert expected in migration assert "drop table" not in migration resources = ( (Path(__file__).resolve().parents[1] / "app/projects/migrations/0002_project_resources.sql") .read_text(encoding="utf-8") .casefold() ) for expected in ( "alter table media_assets add column if not exists project_id", "fk_media_assets_project", "mediarouter_assert_media_asset_project_workspace", "create table if not exists project_generation_jobs", "uq_project_generation_job", "mediarouter_assert_project_generation_job_workspace", "enable row level security", "force row level security", "project_generation_jobs_select", "project_generation_jobs_insert", "project_generation_jobs_delete", ): assert expected in resources assert "drop table" not in resources def test_project_resource_scope_mapping_uses_project_update() -> None: assert ScopePolicy._project_scope("/v1/projects", "POST") == "projects:create" assert ScopePolicy._project_scope("/v1/projects/id", "DELETE") == "projects:delete" assert ScopePolicy._project_scope("/v1/projects/id/assets", "GET") == "projects:read" assert ScopePolicy._project_scope("/v1/projects/id/assets", "POST") == "projects:update" assert ScopePolicy._project_scope("/v1/projects/id/assets/asset", "DELETE") == "projects:update" assert ScopePolicy._project_scope("/v1/projects/id/jobs/job", "DELETE") == "projects:update" assert ScopePolicy._project_scope("/v1/projects/id/editor", "PUT") == "projects:update" assert ScopePolicy._project_scope("/v1/projects/id/renders", "POST") == "projects:update"