Coverage for src/couchers/servicers/media.py: 96%

25 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 16:44 +0000

1import logging 

2 

3import grpc 

4from google.protobuf import empty_pb2 

5 

6from couchers import errors 

7from couchers.crypto import secure_compare 

8from couchers.db import session_scope 

9from couchers.interceptors import ManualAuthValidatorInterceptor 

10from couchers.models import InitiatedUpload, Upload 

11from couchers.sql import couchers_select as select 

12from proto import media_pb2_grpc 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17def get_media_auth_interceptor(secret_token): 

18 def is_authorized(token): 

19 return secure_compare(token.encode("ascii"), secret_token.encode("ascii")) 

20 

21 return ManualAuthValidatorInterceptor(is_authorized) 

22 

23 

24class Media(media_pb2_grpc.MediaServicer): 

25 def UploadConfirmation(self, request, context): 

26 with session_scope() as session: 

27 initiated_upload = session.execute( 

28 select(InitiatedUpload).where(InitiatedUpload.key == request.key).where(InitiatedUpload.is_valid) 

29 ).scalar_one_or_none() 

30 

31 if not initiated_upload: 

32 context.abort(grpc.StatusCode.NOT_FOUND, errors.UPLOAD_NOT_FOUND) 

33 

34 # move it to a completed upload 

35 upload = Upload( 

36 key=request.key, 

37 filename=request.filename, 

38 creator_user_id=initiated_upload.initiator_user_id, 

39 ) 

40 session.add(upload) 

41 

42 # delete the old upload 

43 session.delete(initiated_upload) 

44 

45 return empty_pb2.Empty()