Spaces:
Running
Running
File size: 2,368 Bytes
0a19e69 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | 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()
|