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

1from datetime import timedelta 

2 

3import pytest 

4from google.protobuf import wrappers_pb2 

5from sqlalchemy import select 

6 

7from couchers.crypto import random_hex 

8from couchers.db import session_scope 

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

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

11from couchers.proto import events_pb2, pages_pb2 

12from couchers.utils import datetime_to_iso8601_local, now 

13from tests.conftest import testconfig # noqa 

14from tests.fixtures.db import generate_user 

15from tests.fixtures.sessions import events_session, pages_session 

16from tests.test_communities import create_community 

17 

18 

19@pytest.fixture(autouse=True) 

20def _(testconfig): 

21 pass 

22 

23 

24def _add_upload(user_id, key): 

25 with session_scope() as session: 

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

27 

28 

29def test_get_upload_uses_unused(db): 

30 user, _ = generate_user() 

31 _add_upload(user.id, "unused_key") 

32 

33 with session_scope() as session: 

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

35 

36 

37def test_get_upload_uses_profile_gallery(db): 

38 user, _ = generate_user() 

39 _add_upload(user.id, "avatar_key") 

40 _add_upload(user.id, "other_key") 

41 

42 with session_scope() as session: 

43 gallery = PhotoGallery(owner_user_id=user.id) 

44 session.add(gallery) 

45 session.flush() 

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

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

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

49 

50 with session_scope() as session: 

51 avatar_uses = get_upload_uses(session, "avatar_key") 

52 assert len(avatar_uses) == 1 

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

54 assert avatar_uses[0].is_current 

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

56 assert avatar_uses[0].url is not None 

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

58 

59 other_uses = get_upload_uses(session, "other_key") 

60 assert len(other_uses) == 1 

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

62 assert other_uses[0].is_current 

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

64 

65 

66def test_get_upload_uses_event(db): 

67 user, token = generate_user() 

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

69 end_time = start_time + timedelta(hours=3) 

70 

71 with session_scope() as session: 

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

73 _add_upload(user.id, "event_key") 

74 

75 with events_session(token) as api: 

76 res = api.CreateEvent( 

77 events_pb2.CreateEventReq( 

78 title="Event With Photo", 

79 content="content", 

80 photo_key="event_key", 

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

82 start_datetime_iso8601_local=datetime_to_iso8601_local(start_time), 

83 end_datetime_iso8601_local=datetime_to_iso8601_local(end_time), 

84 ) 

85 ) 

86 event_id = res.event_id 

87 

88 with session_scope() as session: 

89 uses = get_upload_uses(session, "event_key") 

90 assert len(uses) == 1 

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

92 assert uses[0].is_current 

93 assert uses[0].event_id == event_id 

94 assert uses[0].url is not None 

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

96 

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

98 with session_scope() as session: 

99 occurrence = session.execute( 

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

101 ).scalar_one() 

102 occurrence.is_deleted = True 

103 

104 with session_scope() as session: 

105 uses = get_upload_uses(session, "event_key") 

106 assert len(uses) == 1 

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

108 assert not uses[0].is_current 

109 

110 

111def test_get_upload_uses_page(db): 

112 user, token = generate_user() 

113 with session_scope() as session: 

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

115 

116 key = random_hex(32) 

117 _add_upload(user.id, key) 

118 

119 with pages_session(token) as api: 

120 res = api.CreatePlace( 

121 pages_pb2.CreatePlaceReq( 

122 title="title", 

123 content="content", 

124 photo_key=key, 

125 address="address", 

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

127 ) 

128 ) 

129 page_id = res.page_id 

130 

131 with session_scope() as session: 

132 uses = get_upload_uses(session, key) 

133 assert len(uses) == 1 

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

135 assert uses[0].is_current 

136 assert uses[0].page_id == page_id 

137 

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

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

140 

141 with session_scope() as session: 

142 uses = get_upload_uses(session, key) 

143 assert len(uses) == 1 

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

145 assert not uses[0].is_current 

146 assert uses[0].page_id == page_id 

147 

148 

149def test_get_upload_uses_community_page(db): 

150 user, _ = generate_user() 

151 with session_scope() as session: 

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

153 

154 key = random_hex(32) 

155 _add_upload(user.id, key) 

156 

157 with session_scope() as session: 

158 main_page = session.execute( 

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

160 ).scalar_one() 

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

162 

163 with session_scope() as session: 

164 uses = get_upload_uses(session, key) 

165 assert len(uses) == 1 

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

167 assert uses[0].is_current 

168 assert uses[0].url is not None 

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

170 

171 

172def test_get_upload_uses_multiple(db): 

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

174 user, token = generate_user() 

175 with session_scope() as session: 

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

177 

178 key = random_hex(32) 

179 _add_upload(user.id, key) 

180 

181 with session_scope() as session: 

182 gallery = PhotoGallery(owner_user_id=user.id) 

183 session.add(gallery) 

184 session.flush() 

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

186 

187 with pages_session(token) as api: 

188 api.CreatePlace( 

189 pages_pb2.CreatePlaceReq( 

190 title="title", 

191 content="content", 

192 photo_key=key, 

193 address="address", 

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

195 ) 

196 ) 

197 

198 with session_scope() as session: 

199 uses = get_upload_uses(session, key) 

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

201 UploadUseType.profile_gallery_photo_avatar, 

202 UploadUseType.page, 

203 } 

204 

205 

206def test_get_upload_uses_for_keys_batch(db): 

207 user, _ = generate_user() 

208 _add_upload(user.id, "gallery_key") 

209 _add_upload(user.id, "unused_key") 

210 

211 with session_scope() as session: 

212 gallery = PhotoGallery(owner_user_id=user.id) 

213 session.add(gallery) 

214 session.flush() 

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

216 

217 with session_scope() as session: 

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

219 

220 # only keys with uses appear in the mapping 

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

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

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

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

225 

226 

227def test_get_upload_uses_for_keys_empty(db): 

228 with session_scope() as session: 

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

230 

231 

232def test_upload_uses_covers_all_foreign_keys(db): 

233 """ 

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

235 

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

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

238 """ 

239 referencing = set() 

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

241 for fk in table.foreign_keys: 

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

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

244 

245 assert referencing == { 

246 ("photo_gallery_items", "upload_key"), 

247 ("event_occurrences", "photo_key"), 

248 ("page_versions", "photo_key"), 

249 }