Coverage for app/backend/src/tests/test_upload_uses.py: 100%
143 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1from google.protobuf import wrappers_pb2
2from sqlalchemy import select
4from couchers.crypto import random_hex
5from couchers.db import session_scope
6from couchers.helpers.upload_uses import UploadUseType, get_upload_uses, get_upload_uses_for_keys
7from couchers.models import Base, EventOccurrence, Page, PageType, PhotoGallery, PhotoGalleryItem, Upload
8from couchers.proto import pages_pb2
9from tests.fixtures.api import events as events_api
10from tests.fixtures.db import generate_user
11from tests.fixtures.sessions import pages_session
12from tests.test_communities import create_community
15def _add_upload(user_id, key):
16 with session_scope() as session:
17 session.add(Upload(key=key, filename=f"{key}.jpg", creator_user_id=user_id))
20def test_get_upload_uses_unused(db):
21 user, _ = generate_user()
22 _add_upload(user.id, "unused_key")
24 with session_scope() as session:
25 assert get_upload_uses(session, "unused_key") == []
28def test_get_upload_uses_profile_gallery(db):
29 user, _ = generate_user()
30 _add_upload(user.id, "avatar_key")
31 _add_upload(user.id, "other_key")
33 with session_scope() as session:
34 gallery = PhotoGallery(owner_user_id=user.id)
35 session.add(gallery)
36 session.flush()
37 # lower position sorts first, so it's the avatar
38 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="avatar_key", position=1.0))
39 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="other_key", position=2.0))
41 with session_scope() as session:
42 avatar_uses = get_upload_uses(session, "avatar_key")
43 assert len(avatar_uses) == 1
44 assert avatar_uses[0].use_type == UploadUseType.profile_gallery_photo_avatar
45 assert avatar_uses[0].is_current
46 assert avatar_uses[0].user_id == user.id
47 assert avatar_uses[0].url is not None
48 assert f"/user/{user.username}" in avatar_uses[0].url
50 other_uses = get_upload_uses(session, "other_key")
51 assert len(other_uses) == 1
52 assert other_uses[0].use_type == UploadUseType.profile_gallery_photo
53 assert other_uses[0].is_current
54 assert other_uses[0].user_id == user.id
57def test_get_upload_uses_event(db):
58 user, token = generate_user()
60 with session_scope() as session:
61 create_community(session, 0, 2, "Community", [user], [], None)
62 _add_upload(user.id, "event_key")
64 event_id = events_api.create_event(token=token, photo_key="event_key").event_id
66 with session_scope() as session:
67 uses = get_upload_uses(session, "event_key")
68 assert len(uses) == 1
69 assert uses[0].use_type == UploadUseType.event
70 assert uses[0].is_current
71 assert uses[0].event_id == event_id
72 assert uses[0].url is not None
73 assert f"/event/{event_id}/" in uses[0].url
75 # a deleted occurrence still references the upload, but is no longer shown
76 with session_scope() as session:
77 occurrence = session.execute(
78 select(EventOccurrence).where(EventOccurrence.photo_key == "event_key")
79 ).scalar_one()
80 occurrence.is_deleted = True
82 with session_scope() as session:
83 uses = get_upload_uses(session, "event_key")
84 assert len(uses) == 1
85 assert uses[0].use_type == UploadUseType.event
86 assert not uses[0].is_current
89def test_get_upload_uses_page(db):
90 user, token = generate_user()
91 with session_scope() as session:
92 create_community(session, 0, 2, "Root node", [user], [], None)
94 key = random_hex(32)
95 _add_upload(user.id, key)
97 with pages_session(token) as api:
98 res = api.CreatePlace(
99 pages_pb2.CreatePlaceReq(
100 title="title",
101 content="content",
102 photo_key=key,
103 address="address",
104 location=pages_pb2.Coordinate(lat=1, lng=1),
105 )
106 )
107 page_id = res.page_id
109 with session_scope() as session:
110 uses = get_upload_uses(session, key)
111 assert len(uses) == 1
112 assert uses[0].use_type == UploadUseType.page
113 assert uses[0].is_current
114 assert uses[0].page_id == page_id
116 # clearing the photo creates a new version; the old version still references the upload
117 api.UpdatePage(pages_pb2.UpdatePageReq(page_id=page_id, photo_key=wrappers_pb2.StringValue(value="")))
119 with session_scope() as session:
120 uses = get_upload_uses(session, key)
121 assert len(uses) == 1
122 assert uses[0].use_type == UploadUseType.page
123 assert not uses[0].is_current
124 assert uses[0].page_id == page_id
127def test_get_upload_uses_community_page(db):
128 user, _ = generate_user()
129 with session_scope() as session:
130 node_id = create_community(session, 0, 2, "Community", [user], [], None).id
132 key = random_hex(32)
133 _add_upload(user.id, key)
135 with session_scope() as session:
136 main_page = session.execute(
137 select(Page).where(Page.type == PageType.main_page).where(Page.parent_node_id == node_id)
138 ).scalar_one()
139 main_page.versions[-1].photo_key = key
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 uses[0].is_current
146 assert uses[0].url is not None
147 assert f"/community/{node_id}/" in uses[0].url
150def test_get_upload_uses_multiple(db):
151 """An upload can be used in several places at once; all are returned."""
152 user, token = generate_user()
153 with session_scope() as session:
154 create_community(session, 0, 2, "Root node", [user], [], None)
156 key = random_hex(32)
157 _add_upload(user.id, key)
159 with session_scope() as session:
160 gallery = PhotoGallery(owner_user_id=user.id)
161 session.add(gallery)
162 session.flush()
163 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key=key, position=1.0))
165 with pages_session(token) as api:
166 api.CreatePlace(
167 pages_pb2.CreatePlaceReq(
168 title="title",
169 content="content",
170 photo_key=key,
171 address="address",
172 location=pages_pb2.Coordinate(lat=1, lng=1),
173 )
174 )
176 with session_scope() as session:
177 uses = get_upload_uses(session, key)
178 assert {use.use_type for use in uses} == {
179 UploadUseType.profile_gallery_photo_avatar,
180 UploadUseType.page,
181 }
184def test_get_upload_uses_for_keys_batch(db):
185 user, _ = generate_user()
186 _add_upload(user.id, "gallery_key")
187 _add_upload(user.id, "unused_key")
189 with session_scope() as session:
190 gallery = PhotoGallery(owner_user_id=user.id)
191 session.add(gallery)
192 session.flush()
193 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="gallery_key", position=1.0))
195 with session_scope() as session:
196 result = get_upload_uses_for_keys(session, ["gallery_key", "unused_key", "nonexistent_key"])
198 # only keys with uses appear in the mapping
199 assert set(result.keys()) == {"gallery_key"}
200 assert len(result["gallery_key"]) == 1
201 assert result["gallery_key"][0].use_type == UploadUseType.profile_gallery_photo_avatar
202 assert result["gallery_key"][0].user_id == user.id
205def test_get_upload_uses_for_keys_empty(db):
206 with session_scope() as session:
207 assert get_upload_uses_for_keys(session, []) == {}
210def test_upload_uses_covers_all_foreign_keys(db):
211 """
212 Guards against drift: every foreign key targeting uploads.key must be handled by get_upload_uses.
214 If this fails, you added a new reference to uploads.key. Add it to get_upload_uses (and likely a new
215 UploadUseType), then add the (table, column) here.
216 """
217 referencing = set()
218 for table in Base.metadata.tables.values():
219 for fk in table.foreign_keys:
220 if fk.column.table.name == "uploads" and fk.column.name == "key":
221 referencing.add((table.name, fk.parent.name))
223 assert referencing == {
224 ("photo_gallery_items", "upload_key"),
225 ("event_occurrences", "photo_key"),
226 ("page_versions", "photo_key"),
227 }