tests/test_projects_foundation.py

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"