Spaces:
Running
Running
| 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() | |