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

1""" 

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

3couchers/helpers/upload_uses.py. 

4""" 

5 

6import json 

7import logging 

8 

9import grpc 

10from google.protobuf import empty_pb2 

11from sqlalchemy import select 

12from sqlalchemy.orm import Session 

13 

14from couchers.context import CouchersContext 

15from couchers.crypto import secure_compare 

16from couchers.middleware.interceptors import MediaInterceptor 

17from couchers.models import InitiatedUpload, Upload 

18from couchers.proto import media_pb2, media_pb2_grpc 

19 

20logger = logging.getLogger(__name__) 

21 

22 

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

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

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

26 

27 return MediaInterceptor(is_authorized) 

28 

29 

30class Media(media_pb2_grpc.MediaServicer): 

31 def UploadConfirmation( 

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

33 ) -> empty_pb2.Empty: 

34 initiated_upload = session.execute( 

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

36 ).scalar_one_or_none() 

37 

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

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

40 

41 metadata = request.metadata 

42 

43 # move it to a completed upload 

44 upload = Upload( 

45 key=request.key, 

46 filename=request.filename, 

47 creator_user_id=initiated_upload.initiator_user_id, 

48 metadata_exif=metadata.exif or None, 

49 metadata_xmp=metadata.xmp or None, 

50 metadata_iptc=metadata.iptc or None, 

51 metadata_parsed=json.loads(metadata.parsed_json) if metadata.parsed_json else None, 

52 metadata_parse_error=metadata.parse_error or None, 

53 original_filename=metadata.original_filename or None, 

54 original_format=metadata.original_format or None, 

55 original_size=metadata.original_size or None, 

56 original_width=metadata.original_width or None, 

57 original_height=metadata.original_height or None, 

58 ) 

59 session.add(upload) 

60 

61 # delete the old upload 

62 session.delete(initiated_upload) 

63 

64 return empty_pb2.Empty()