Coverage for app/backend/src/tests/test_media.py: 100%

1import json 

2from urllib.parse import parse_qs, urlparse 

3 

4from google.protobuf import empty_pb2 

5from sqlalchemy import inspect, select 

6from sqlalchemy.orm import undefer 

7 

8from couchers.crypto import b64decode, random_hex 

9from couchers.db import session_scope 

10from couchers.models import InitiatedUpload, Upload 

11from couchers.proto import media_pb2 

12from tests.fixtures.db import generate_user 

13from tests.fixtures.sessions import api_session, media_session 

14 

15METADATA_COLUMNS = [attr.key for attr in inspect(Upload).column_attrs if attr.deferred] 

16 

17 

18def test_media_upload(db): 

19 user, token = generate_user() 

20 

21 media_bearer_token = random_hex(32) 

22 

23 with api_session(token) as api: 

24 res = api.InitiateMediaUpload(empty_pb2.Empty()) 

25 

26 params = parse_qs(urlparse(res.upload_url).query) 

27 data = b64decode(params["data"][0]) 

28 

29 response = media_pb2.UploadRequest.FromString(data) 

30 key = response.key 

31 

32 filename = random_hex(32) 

33 

34 req = media_pb2.UploadConfirmationReq(key=key, filename=filename) 

35 

36 with session_scope() as session: 

37 # make sure it exists 

38 assert ( 

39 session.execute(select(InitiatedUpload).where(InitiatedUpload.key == key)).scalar_one_or_none() is not None 

40 ) 

41 

42 with media_session(media_bearer_token) as media: 

43 res = media.UploadConfirmation(req) 

44 

45 with session_scope() as session: 

46 # make sure it exists 

47 assert ( 

48 session.execute( 

49 select(Upload) 

50 .where(Upload.key == key) 

51 .where(Upload.filename == filename) 

52 .where(Upload.creator_user_id == user.id) 

53 ).scalar_one_or_none() 

54 is not None 

55 ) 

56 

57 with session_scope() as session: 

58 # make sure it was deleted 

59 assert not session.execute(select(InitiatedUpload)).scalar_one_or_none() 

60 

61 

62def initiate_upload(token): 

63 with api_session(token) as api: 

64 res = api.InitiateMediaUpload(empty_pb2.Empty()) 

65 

66 params = parse_qs(urlparse(res.upload_url).query) 

67 return media_pb2.UploadRequest.FromString(b64decode(params["data"][0])).key 

68 

69 

70def get_upload(session, key): 

71 return session.execute(select(Upload).where(Upload.key == key).options(undefer("*"))).scalar_one() 

72 

73 

74def test_media_upload_metadata(db): 

75 user, token = generate_user() 

76 key = initiate_upload(token) 

77 

78 req = media_pb2.UploadConfirmationReq( 

79 key=key, 

80 filename=random_hex(32), 

81 metadata=media_pb2.UploadMetadata( 

82 exif=b"Exif\0\0raw exif segment", 

83 xmp=b"<?xpacket ...", 

84 iptc=b"Photoshop 3.0\0", 

85 parsed_json=json.dumps({"GPS GPSLatitudeRef": "N", "Image Make": "TestCam"}), 

86 original_filename="IMG_1234.HEIC", 

87 original_format="heif", 

88 original_size=4194304, 

89 original_width=4032, 

90 original_height=3024, 

91 ), 

92 ) 

93 

94 with media_session(random_hex(32)) as media: 

95 media.UploadConfirmation(req) 

96 

97 with session_scope() as session: 

98 upload = get_upload(session, key) 

99 

100 assert upload.metadata_exif == b"Exif\0\0raw exif segment" 

101 assert upload.metadata_xmp == b"<?xpacket ..." 

102 assert upload.metadata_iptc == b"Photoshop 3.0\0" 

103 assert upload.metadata_parsed == {"GPS GPSLatitudeRef": "N", "Image Make": "TestCam"} 

104 assert upload.metadata_parse_error is None 

105 assert upload.original_filename == "IMG_1234.HEIC" 

106 assert upload.original_format == "heif" 

107 assert upload.original_size == 4194304 

108 assert upload.original_width == 4032 

109 assert upload.original_height == 3024 

110 

111 

112def test_media_upload_without_metadata(db): 

113 user, token = generate_user() 

114 key = initiate_upload(token) 

115 

116 with media_session(random_hex(32)) as media: 

117 media.UploadConfirmation(media_pb2.UploadConfirmationReq(key=key, filename=random_hex(32))) 

118 

119 with session_scope() as session: 

120 upload = get_upload(session, key) 

121 

122 assert upload.metadata_exif is None 

123 assert upload.metadata_parsed is None 

124 assert upload.metadata_parse_error is None 

125 assert upload.original_filename is None 

126 assert upload.original_format is None 

127 assert upload.original_size is None 

128 

129 

130def test_a_failed_parse_is_recorded_next_to_the_raw_segment(db): 

131 user, token = generate_user() 

132 key = initiate_upload(token) 

133 

134 req = media_pb2.UploadConfirmationReq( 

135 key=key, 

136 filename=random_hex(32), 

137 metadata=media_pb2.UploadMetadata( 

138 exif=b"Exif\0\0raw exif segment", 

139 parse_error="Traceback (most recent call last):\n ...\nAssertionError\n", 

140 ), 

141 ) 

142 

143 with media_session(random_hex(32)) as media: 

144 media.UploadConfirmation(req) 

145 

146 with session_scope() as session: 

147 upload = get_upload(session, key) 

148 

149 assert upload.metadata_exif == b"Exif\0\0raw exif segment" 

150 assert upload.metadata_parsed is None 

151 assert upload.metadata_parse_error == "Traceback (most recent call last):\n ...\nAssertionError\n" 

152 

153 

154def test_upload_metadata_is_deferred(db): 

155 user, token = generate_user() 

156 key = initiate_upload(token) 

157 

158 with media_session(random_hex(32)) as media: 

159 media.UploadConfirmation( 

160 media_pb2.UploadConfirmationReq( 

161 key=key, filename=random_hex(32), metadata=media_pb2.UploadMetadata(exif=b"Exif\0\0blob") 

162 ) 

163 ) 

164 

165 with session_scope() as session: 

166 upload = session.execute(select(Upload).where(Upload.key == key)).scalar_one() 

167 

168 assert "metadata_exif" not in upload.__dict__ 

169 assert "metadata_parsed" not in upload.__dict__ 

170 assert "metadata_parse_error" not in upload.__dict__ 

171 assert upload.metadata_exif == b"Exif\0\0blob" 

172 

173 

174def test_no_metadata_column_rides_along_on_an_entity_load(db): 

175 user, token = generate_user() 

176 key = initiate_upload(token) 

177 

178 with media_session(random_hex(32)) as media: 

179 media.UploadConfirmation( 

180 media_pb2.UploadConfirmationReq( 

181 key=key, 

182 filename=random_hex(32), 

183 metadata=media_pb2.UploadMetadata(original_format="webp", original_width=800), 

184 ) 

185 ) 

186 

187 with session_scope() as session: 

188 upload = session.execute(select(Upload).where(Upload.key == key)).scalar_one() 

189 

190 assert not [column for column in METADATA_COLUMNS if column in upload.__dict__] 

191 assert upload.original_format == "webp"