Coverage for app/backend/src/couchers/servicers/media.py: 92%

1""" 

2When adding a new foreign key to uploads.key, also update the reverse lookup in 

3couchers/helpers/upload_uses.py. 

4""" 

5 

6import logging 

7 

8import grpc 

9from google.protobuf import empty_pb2 

10from sqlalchemy import select 

11from sqlalchemy.orm import Session 

12 

13from couchers.context import CouchersContext 

14from couchers.crypto import secure_compare 

15from couchers.interceptors import MediaInterceptor 

16from couchers.models import InitiatedUpload, Upload 

17from couchers.proto import media_pb2, media_pb2_grpc 

18 

19logger = logging.getLogger(__name__) 

20 

21 

22def get_media_auth_interceptor(secret_token: str) -> MediaInterceptor: 

23 def is_authorized(token: str) -> bool: 

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

25 

26 return MediaInterceptor(is_authorized) 

27 

28 

29class Media(media_pb2_grpc.MediaServicer): 

30 def UploadConfirmation( 

31 self, request: media_pb2.UploadConfirmationReq, context: CouchersContext, session: Session 

32 ) -> empty_pb2.Empty: 

33 initiated_upload = session.execute( 

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

35 ).scalar_one_or_none() 

36 

37 if not initiated_upload: 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true

38 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "upload_not_found") 

39 

40 # move it to a completed upload 

41 upload = Upload( 

42 key=request.key, 

43 filename=request.filename, 

44 creator_user_id=initiated_upload.initiator_user_id, 

45 ) 

46 session.add(upload) 

47 

48 # delete the old upload 

49 session.delete(initiated_upload) 

50 

51 return empty_pb2.Empty()