Coverage for app/backend/src/tests/test_upload_uses.py: 100%
153 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-21 09:29 +0000
1from datetime import timedelta
3import pytest
4from google.protobuf import wrappers_pb2
5from sqlalchemy import select
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 Timestamp_from_datetime, 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
19@pytest.fixture(autouse=True)
20def _(testconfig):
21 pass
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))
29def test_get_upload_uses_unused(db):
30 user, _ = generate_user()
31 _add_upload(user.id, "unused_key")
33 with session_scope() as session:
34 assert get_upload_uses(session, "unused_key") == []
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")
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))
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
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
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)
71 with session_scope() as session:
72 create_community(session, 0, 2, "Community", [user], [], None)
73 _add_upload(user.id, "event_key")
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 offline_information=events_pb2.OfflineEventInformation(address="Null Island", lat=0.1, lng=0.2),
82 start_time=Timestamp_from_datetime(start_time),
83 end_time=Timestamp_from_datetime(end_time),
84 timezone="UTC",
85 )
86 )
87 event_id = res.event_id
89 with session_scope() as session:
90 uses = get_upload_uses(session, "event_key")
91 assert len(uses) == 1
92 assert uses[0].use_type == UploadUseType.event
93 assert uses[0].is_current
94 assert uses[0].event_id == event_id
95 assert uses[0].url is not None
96 assert f"/event/{event_id}/" in uses[0].url
98 # a deleted occurrence still references the upload, but is no longer shown
99 with session_scope() as session:
100 occurrence = session.execute(
101 select(EventOccurrence).where(EventOccurrence.photo_key == "event_key")
102 ).scalar_one()
103 occurrence.is_deleted = True
105 with session_scope() as session:
106 uses = get_upload_uses(session, "event_key")
107 assert len(uses) == 1
108 assert uses[0].use_type == UploadUseType.event
109 assert not uses[0].is_current
112def test_get_upload_uses_page(db):
113 user, token = generate_user()
114 with session_scope() as session:
115 create_community(session, 0, 2, "Root node", [user], [], None)
117 key = random_hex(32)
118 _add_upload(user.id, key)
120 with pages_session(token) as api:
121 res = api.CreatePlace(
122 pages_pb2.CreatePlaceReq(
123 title="title",
124 content="content",
125 photo_key=key,
126 address="address",
127 location=pages_pb2.Coordinate(lat=1, lng=1),
128 )
129 )
130 page_id = res.page_id
132 with session_scope() as session:
133 uses = get_upload_uses(session, key)
134 assert len(uses) == 1
135 assert uses[0].use_type == UploadUseType.page
136 assert uses[0].is_current
137 assert uses[0].page_id == page_id
139 # clearing the photo creates a new version; the old version still references the upload
140 api.UpdatePage(pages_pb2.UpdatePageReq(page_id=page_id, photo_key=wrappers_pb2.StringValue(value="")))
142 with session_scope() as session:
143 uses = get_upload_uses(session, key)
144 assert len(uses) == 1
145 assert uses[0].use_type == UploadUseType.page
146 assert not uses[0].is_current
147 assert uses[0].page_id == page_id
150def test_get_upload_uses_community_page(db):
151 user, _ = generate_user()
152 with session_scope() as session:
153 node_id = create_community(session, 0, 2, "Community", [user], [], None).id
155 key = random_hex(32)
156 _add_upload(user.id, key)
158 with session_scope() as session:
159 main_page = session.execute(
160 select(Page).where(Page.type == PageType.main_page).where(Page.parent_node_id == node_id)
161 ).scalar_one()
162 main_page.versions[-1].photo_key = key
164 with session_scope() as session:
165 uses = get_upload_uses(session, key)
166 assert len(uses) == 1
167 assert uses[0].use_type == UploadUseType.page
168 assert uses[0].is_current
169 assert uses[0].url is not None
170 assert f"/community/{node_id}/" in uses[0].url
173def test_get_upload_uses_multiple(db):
174 """An upload can be used in several places at once; all are returned."""
175 user, token = generate_user()
176 with session_scope() as session:
177 create_community(session, 0, 2, "Root node", [user], [], None)
179 key = random_hex(32)
180 _add_upload(user.id, key)
182 with session_scope() as session:
183 gallery = PhotoGallery(owner_user_id=user.id)
184 session.add(gallery)
185 session.flush()
186 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key=key, position=1.0))
188 with pages_session(token) as api:
189 api.CreatePlace(
190 pages_pb2.CreatePlaceReq(
191 title="title",
192 content="content",
193 photo_key=key,
194 address="address",
195 location=pages_pb2.Coordinate(lat=1, lng=1),
196 )
197 )
199 with session_scope() as session:
200 uses = get_upload_uses(session, key)
201 assert {use.use_type for use in uses} == {
202 UploadUseType.profile_gallery_photo_avatar,
203 UploadUseType.page,
204 }
207def test_get_upload_uses_for_keys_batch(db):
208 user, _ = generate_user()
209 _add_upload(user.id, "gallery_key")
210 _add_upload(user.id, "unused_key")
212 with session_scope() as session:
213 gallery = PhotoGallery(owner_user_id=user.id)
214 session.add(gallery)
215 session.flush()
216 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="gallery_key", position=1.0))
218 with session_scope() as session:
219 result = get_upload_uses_for_keys(session, ["gallery_key", "unused_key", "nonexistent_key"])
221 # only keys with uses appear in the mapping
222 assert set(result.keys()) == {"gallery_key"}
223 assert len(result["gallery_key"]) == 1
224 assert result["gallery_key"][0].use_type == UploadUseType.profile_gallery_photo_avatar
225 assert result["gallery_key"][0].user_id == user.id
228def test_get_upload_uses_for_keys_empty(db):
229 with session_scope() as session:
230 assert get_upload_uses_for_keys(session, []) == {}
233def test_upload_uses_covers_all_foreign_keys(db):
234 """
235 Guards against drift: every foreign key targeting uploads.key must be handled by get_upload_uses.
237 If this fails, you added a new reference to uploads.key. Add it to get_upload_uses (and likely a new
238 UploadUseType), then add the (table, column) here.
239 """
240 referencing = set()
241 for table in Base.metadata.tables.values():
242 for fk in table.foreign_keys:
243 if fk.column.table.name == "uploads" and fk.column.name == "key":
244 referencing.add((table.name, fk.parent.name))
246 assert referencing == {
247 ("photo_gallery_items", "upload_key"),
248 ("event_occurrences", "photo_key"),
249 ("page_versions", "photo_key"),
250 }