Test workspace-scoped API key administration

tests/test_api_key_workspace_authorization.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from app.core.config import Settings
+from app.security.models import WorkspaceMembership
+from app.security.schemas import APIKeyCreate
+from main import create_app
+
+
+def _settings(tmp_path: Path) -> Settings:
+    return Settings(
+        _env_file=None,
+        auth_enabled=True,
+        studio_identity_secret="test-only-studio-identity-secret-material",
+        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",
+        max_workers=1,
+    )
+
+
+def test_api_key_admin_routes_are_workspace_scoped_and_role_gated(tmp_path: Path) -> None:
+    settings = _settings(tmp_path)
+    application = create_app(settings)
+    container = application.state.container
+    asyncio.run(container.security_database.initialize())
+
+    owner_record, owner_secret = asyncio.run(
+        container.api_keys.create(
+            APIKeyCreate(name="Workspace owner", environment="test", role=None, scopes=[]),
+            created_by="tests",
+        )
+    )
+    owner_context = asyncio.run(container.api_keys.authenticate(owner_secret))
+    same_workspace_record, _ = asyncio.run(
+        container.api_keys.create(
+            APIKeyCreate(name="Same workspace key", environment="test", role=None, scopes=[]),
+            created_by=owner_record.id,
+            workspace_id=owner_context.workspace_id,
+            user_id=owner_context.user_id,
+        )
+    )
+    other_record, _ = asyncio.run(
+        container.api_keys.create(
+            APIKeyCreate(name="Other workspace key", environment="test", role=None, scopes=[]),
+            created_by="tests",
+        )
+    )
+
+    headers = {"Authorization": f"Bearer {owner_secret}"}
+    with TestClient(application) as client:
+        listed = client.get("/v1/api-keys", headers=headers)
+        assert listed.status_code == 200
+        assert listed.json()["total"] == 2
+        ids = {item["id"] for item in listed.json()["items"]}
+        assert ids == {owner_record.id, same_workspace_record.id}
+        assert client.get(f"/v1/api-keys/{other_record.id}", headers=headers).status_code == 404
+
+        async def downgrade_membership() -> None:
+            async with container.security_database.session() as session:
+                membership = await session.get(WorkspaceMembership, owner_context.membership_id)
+                assert membership is not None
+                membership.role = "member"
+                await session.commit()
+
+        asyncio.run(downgrade_membership())
+        denied = client.get("/v1/api-keys", headers=headers)
+        assert denied.status_code == 403
+
+    asyncio.run(container.security_database.close())