Forward admin identity when creating analysis sessions

app.py

@@ -606,7 +606,8 @@ async def analyze(request: Request, playerId: str = Form(...), clientId: UUID =
     reference = playerId.strip()
     if not reference or len(reference) > 100 or len(sessionName) > 150:
         raise HTTPException(422, "Invalid player reference or session name")
-    session = backend("", user_id, "POST", {"clientId": client_id, "playerId": reference, "name": sessionName})
+    session = backend("", user_id, "POST", {"clientId": client_id, "playerId": reference, "name": sessionName},
+        is_admin=request.state.is_admin)
     session_id, player_id = session["session_id"], session["player_id"]
     owner_user_id = session.get("user_id") if "user_id" in session else user_id
     actor_is_admin = bool(session.get("actor_is_admin", False))
@@ -751,7 +752,7 @@ def result_action_id(session_id, index):
 
 
 def owned_session_row(session_id, user_id, is_admin=False):
-    row = backend(f"/{session_id}", user_id)
+    row = backend(f"/{session_id}", user_id, is_admin=is_admin)
     if row.get("user_id") != user_id and not is_admin:
         raise HTTPException(404, "Session not found")
     return row

tests/test_admin_analysis.py

@@ -0,0 +1,51 @@
+import asyncio
+from io import BytesIO
+import unittest
+from unittest.mock import patch
+from uuid import UUID
+
+from fastapi import UploadFile
+from starlette.requests import Request
+
+import app as analysis_app
+
+
+class StopBeforeUpload(Exception):
+    pass
+
+
+class AdminAnalysisTest(unittest.TestCase):
+    def test_begin_analysis_forwards_admin_identity_to_internal_backend(self):
+        user_id = "11111111-1111-4111-8111-111111111111"
+        client_id = UUID("22222222-2222-4222-8222-222222222222")
+        player_id = "33333333-3333-4333-8333-333333333333"
+        for is_admin in (False, True):
+            with self.subTest(is_admin=is_admin):
+                request = Request({"type": "http", "method": "POST", "path": "/api/analyze", "headers": []})
+                request.state.user_id = user_id
+                request.state.is_admin = is_admin
+                actions = UploadFile(BytesIO(b"{}"), filename="actions.json")
+                calibration = UploadFile(BytesIO(b"data"), filename="calibration.npz")
+                with patch.object(analysis_app, "backend", side_effect=StopBeforeUpload) as backend:
+                    with self.assertRaises(StopBeforeUpload):
+                        asyncio.run(analysis_app.analyze(
+                            request, playerId=player_id, clientId=client_id,
+                            videoOrders=[1, 2], videos=[], videoDriveLinks=["first", "second"],
+                            actionsJson=actions, calibration=calibration, sessionName="Admin session",
+                        ))
+                backend.assert_called_once_with("", user_id, "POST", {
+                    "clientId": str(client_id), "playerId": player_id, "name": "Admin session",
+                }, is_admin=is_admin)
+
+    def test_admin_session_lookup_forwards_admin_identity(self):
+        session_id = "44444444-4444-4444-8444-444444444444"
+        admin_id = "11111111-1111-4111-8111-111111111111"
+        other_owner = "22222222-2222-4222-8222-222222222222"
+        with patch.object(analysis_app, "backend", return_value={"user_id": other_owner}) as backend:
+            self.assertEqual(analysis_app.owned_session_row(session_id, admin_id, is_admin=True),
+                {"user_id": other_owner})
+        backend.assert_called_once_with(f"/{session_id}", admin_id, is_admin=True)
+
+
+if __name__ == "__main__":
+    unittest.main()