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

1from datetime import timedelta 

2 

3from google.protobuf import wrappers_pb2 

4from sqlalchemy import select 

5 

6from couchers.crypto import random_hex 

7from couchers.db import session_scope 

8from couchers.helpers.upload_uses import UploadUseType, get_upload_uses, get_upload_uses_for_keys 

9from couchers.models import Base, EventOccurrence, Page, PageType, PhotoGallery, PhotoGalleryItem, Upload 

10from couchers.proto import events_pb2, pages_pb2 

11from couchers.utils import datetime_to_iso8601_local, now 

12from tests.fixtures.db import generate_user 

13from tests.fixtures.sessions import events_session, pages_session 

14from tests.test_communities import create_community 

15 

16 

17def _add_upload(user_id, key): 

18 with session_scope() as session: 

19 session.add(Upload(key=key, filename=f"{key}.jpg", creator_user_id=user_id)) 

20 

21 

22def test_get_upload_uses_unused(db): 

23 user, _ = generate_user() 

24 _add_upload(user.id, "unused_key") 

25 

26 with session_scope() as session: 

27 assert get_upload_uses(session, "unused_key") == [] 

28 

29 

30def test_get_upload_uses_profile_gallery(db): 

31 user, _ = generate_user() 

32 _add_upload(user.id, "avatar_key") 

33 _add_upload(user.id, "other_key") 

34 

35 with session_scope() as session: 

36 gallery = PhotoGallery(owner_user_id=user.id) 

37 session.add(gallery) 

38 session.flush() 

39 # lower position sorts first, so it's the avatar 

40 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="avatar_key", position=1.0)) 

41 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="other_key", position=2.0)) 

42 

43 with session_scope() as session: 

44 avatar_uses = get_upload_uses(session, "avatar_key") 

45 assert len(avatar_uses) == 1 

46 assert avatar_uses[0].use_type == UploadUseType.profile_gallery_photo_avatar 

47 assert avatar_uses[0].is_current 

48 assert avatar_uses[0].user_id == user.id 

49 assert avatar_uses[0].url is not None 

50 assert f"/user/{user.username}" in avatar_uses[0].url 

51 

52 other_uses = get_upload_uses(session, "other_key") 

53 assert len(other_uses) == 1 

54 assert other_uses[0].use_type == UploadUseType.profile_gallery_photo 

55 assert other_uses[0].is_current 

56 assert other_uses[0].user_id == user.id 

57 

58 

59def test_get_upload_uses_event(db): 

60 user, token = generate_user() 

61 start_time = now() + timedelta(hours=2) 

62 end_time = start_time + timedelta(hours=3) 

63 

64 with session_scope() as session: 

65 create_community(session, 0, 2, "Community", [user], [], None) 

66 _add_upload(user.id, "event_key") 

67 

68 with events_session(token) as api: 

69 res = api.CreateEvent( 

70 events_pb2.CreateEventReq( 

71 title="Event With Photo", 

72 content="content", 

73 photo_key="event_key", 

74 location=events_pb2.EventLocation(address="Null Island", lat=0.1, lng=0.2), 

75 start_datetime_iso8601_local=datetime_to_iso8601_local(start_time), 

76 end_datetime_iso8601_local=datetime_to_iso8601_local(end_time), 

77 ) 

78 ) 

79 event_id = res.event_id 

80 

81 with session_scope() as session: 

82 uses = get_upload_uses(session, "event_key") 

83 assert len(uses) == 1 

84 assert uses[0].use_type == UploadUseType.event 

85 assert uses[0].is_current 

86 assert uses[0].event_id == event_id 

87 assert uses[0].url is not None 

88 assert f"/event/{event_id}/" in uses[0].url 

89 

90 # a deleted occurrence still references the upload, but is no longer shown 

91 with session_scope() as session: 

92 occurrence = session.execute( 

93 select(EventOccurrence).where(EventOccurrence.photo_key == "event_key") 

94 ).scalar_one() 

95 occurrence.is_deleted = True 

96 

97 with session_scope() as session: 

98 uses = get_upload_uses(session, "event_key") 

99 assert len(uses) == 1 

100 assert uses[0].use_type == UploadUseType.event 

101 assert not uses[0].is_current 

102 

103 

104def test_get_upload_uses_page(db): 

105 user, token = generate_user() 

106 with session_scope() as session: 

107 create_community(session, 0, 2, "Root node", [user], [], None) 

108 

109 key = random_hex(32) 

110 _add_upload(user.id, key) 

111 

112 with pages_session(token) as api: 

113 res = api.CreatePlace( 

114 pages_pb2.CreatePlaceReq( 

115 title="title", 

116 content="content", 

117 photo_key=key, 

118 address="address", 

119 location=pages_pb2.Coordinate(lat=1, lng=1), 

120 ) 

121 ) 

122 page_id = res.page_id 

123 

124 with session_scope() as session: 

125 uses = get_upload_uses(session, key) 

126 assert len(uses) == 1 

127 assert uses[0].use_type == UploadUseType.page 

128 assert uses[0].is_current 

129 assert uses[0].page_id == page_id 

130 

131 # clearing the photo creates a new version; the old version still references the upload 

132 api.UpdatePage(pages_pb2.UpdatePageReq(page_id=page_id, photo_key=wrappers_pb2.StringValue(value=""))) 

133 

134 with session_scope() as session: 

135 uses = get_upload_uses(session, key) 

136 assert len(uses) == 1 

137 assert uses[0].use_type == UploadUseType.page 

138 assert not uses[0].is_current 

139 assert uses[0].page_id == page_id 

140 

141 

142def test_get_upload_uses_community_page(db): 

143 user, _ = generate_user() 

144 with session_scope() as session: 

145 node_id = create_community(session, 0, 2, "Community", [user], [], None).id 

146 

147 key = random_hex(32) 

148 _add_upload(user.id, key) 

149 

150 with session_scope() as session: 

151 main_page = session.execute( 

152 select(Page).where(Page.type == PageType.main_page).where(Page.parent_node_id == node_id) 

153 ).scalar_one() 

154 main_page.versions[-1].photo_key = key 

155 

156 with session_scope() as session: 

157 uses = get_upload_uses(session, key) 

158 assert len(uses) == 1 

159 assert uses[0].use_type == UploadUseType.page 

160 assert uses[0].is_current 

161 assert uses[0].url is not None 

162 assert f"/community/{node_id}/" in uses[0].url 

163 

164 

165def test_get_upload_uses_multiple(db): 

166 """An upload can be used in several places at once; all are returned.""" 

167 user, token = generate_user() 

168 with session_scope() as session: 

169 create_community(session, 0, 2, "Root node", [user], [], None) 

170 

171 key = random_hex(32) 

172 _add_upload(user.id, key) 

173 

174 with session_scope() as session: 

175 gallery = PhotoGallery(owner_user_id=user.id) 

176 session.add(gallery) 

177 session.flush() 

178 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key=key, position=1.0)) 

179 

180 with pages_session(token) as api: 

181 api.CreatePlace( 

182 pages_pb2.CreatePlaceReq( 

183 title="title", 

184 content="content", 

185 photo_key=key, 

186 address="address", 

187 location=pages_pb2.Coordinate(lat=1, lng=1), 

188 ) 

189 ) 

190 

191 with session_scope() as session: 

192 uses = get_upload_uses(session, key) 

193 assert {use.use_type for use in uses} == { 

194 UploadUseType.profile_gallery_photo_avatar, 

195 UploadUseType.page, 

196 } 

197 

198 

199def test_get_upload_uses_for_keys_batch(db): 

200 user, _ = generate_user() 

201 _add_upload(user.id, "gallery_key") 

202 _add_upload(user.id, "unused_key") 

203 

204 with session_scope() as session: 

205 gallery = PhotoGallery(owner_user_id=user.id) 

206 session.add(gallery) 

207 session.flush() 

208 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="gallery_key", position=1.0)) 

209 

210 with session_scope() as session: 

211 result = get_upload_uses_for_keys(session, ["gallery_key", "unused_key", "nonexistent_key"]) 

212 

213 # only keys with uses appear in the mapping 

214 assert set(result.keys()) == {"gallery_key"} 

215 assert len(result["gallery_key"]) == 1 

216 assert result["gallery_key"][0].use_type == UploadUseType.profile_gallery_photo_avatar 

217 assert result["gallery_key"][0].user_id == user.id 

218 

219 

220def test_get_upload_uses_for_keys_empty(db): 

221 with session_scope() as session: 

222 assert get_upload_uses_for_keys(session, []) == {} 

223 

224 

225def test_upload_uses_covers_all_foreign_keys(db): 

226 """ 

227 Guards against drift: every foreign key targeting uploads.key must be handled by get_upload_uses. 

228 

229 If this fails, you added a new reference to uploads.key. Add it to get_upload_uses (and likely a new 

230 UploadUseType), then add the (table, column) here. 

231 """ 

232 referencing = set() 

233 for table in Base.metadata.tables.values(): 

234 for fk in table.foreign_keys: 

235 if fk.column.table.name == "uploads" and fk.column.name == "key": 

236 referencing.add((table.name, fk.parent.name)) 

237 

238 assert referencing == { 

239 ("photo_gallery_items", "upload_key"), 

240 ("event_occurrences", "photo_key"), 

241 ("page_versions", "photo_key"), 

242 }