Coverage for app/backend/src/tests/fixtures/db.py: 100%
136 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1import subprocess
2from collections.abc import Sequence
3from contextlib import contextmanager
4from datetime import date, timedelta
5from pathlib import Path
6from typing import Any, cast
8from sqlalchemy import Connection, Engine, create_engine, func, or_, select, text, update
9from sqlalchemy.orm import Session
11from couchers.constants import GUIDELINES_VERSION, HOST_REQUEST_DUPLICATE_WINDOW_HOURS, TOS_VERSION
12from couchers.context import CouchersContext
13from couchers.crypto import random_hex
14from couchers.db import _get_base_engine, session_scope
15from couchers.helpers.completed_profile import has_completed_profile
16from couchers.models import (
17 Base,
18 Conversation,
19 FriendRelationship,
20 FriendStatus,
21 HostingStatus,
22 LanguageAbility,
23 LanguageFluency,
24 ModerationObjectType,
25 ModerationState,
26 ModerationUserList,
27 ModerationVisibility,
28 PassportSex,
29 PhotoGallery,
30 PhotoGalleryItem,
31 RegionLived,
32 RegionVisited,
33 StrongVerificationAttempt,
34 StrongVerificationAttemptStatus,
35 Upload,
36 User,
37 UserBlock,
38 UserSession,
39 Volunteer,
40)
41from couchers.moderation.utils import create_moderation
42from couchers.servicers.auth import create_session
43from couchers.utils import create_coordinate, now
44from tests.fixtures.sessions import _MockCouchersContext
47def create_schema_from_models(engine: Engine | None = None) -> None:
48 """
49 Create everything from the current models, not incrementally
50 through migrations.
51 """
52 if engine is None:
53 engine = _get_base_engine()
55 # create sql functions (these are created in migrations otherwise)
56 functions = Path(__file__).parent / "sql_functions.sql"
57 with open(functions) as f, engine.connect() as conn:
58 conn.execute(text(f.read()))
59 conn.commit()
61 Base.metadata.create_all(engine)
64def populate_testing_resources(conn: Connection) -> None:
65 """
66 Testing version of couchers.resources.copy_resources_to_database
67 """
68 conn.execute(
69 text("""
70 INSERT INTO regions (code, name) VALUES
71 ('AUS', 'Australia'),
72 ('CAN', 'Canada'),
73 ('CHE', 'Switzerland'),
74 ('CUB', 'Cuba'),
75 ('CXR', 'Christmas Island'),
76 ('CZE', 'Czechia'),
77 ('DEU', 'Germany'),
78 ('EGY', 'Egypt'),
79 ('ESP', 'Spain'),
80 ('EST', 'Estonia'),
81 ('FIN', 'Finland'),
82 ('FRA', 'France'),
83 ('GBR', 'United Kingdom'),
84 ('GEO', 'Georgia'),
85 ('GHA', 'Ghana'),
86 ('GRC', 'Greece'),
87 ('HKG', 'Hong Kong'),
88 ('IRL', 'Ireland'),
89 ('ISR', 'Israel'),
90 ('ITA', 'Italy'),
91 ('JPN', 'Japan'),
92 ('LAO', 'Laos'),
93 ('MEX', 'Mexico'),
94 ('MMR', 'Myanmar'),
95 ('NAM', 'Namibia'),
96 ('NLD', 'Netherlands'),
97 ('NZL', 'New Zealand'),
98 ('POL', 'Poland'),
99 ('PRK', 'North Korea'),
100 ('REU', 'Réunion'),
101 ('SGP', 'Singapore'),
102 ('SWE', 'Sweden'),
103 ('THA', 'Thailand'),
104 ('TUR', 'Turkey'),
105 ('TWN', 'Taiwan'),
106 ('USA', 'United States'),
107 ('VNM', 'Vietnam');
108 """)
109 )
111 # Insert languages as textual SQL
112 conn.execute(
113 text("""
114 INSERT INTO languages (code, name) VALUES
115 ('aka', 'Akan'),
116 ('arb', 'Arabic (Standard)'),
117 ('deu', 'German'),
118 ('eng', 'English'),
119 ('fat', 'Fanti'),
120 ('fil', 'Filipino'),
121 ('fin', 'Finnish'),
122 ('fra', 'French'),
123 ('heb', 'Hebrew'),
124 ('hun', 'Hungarian'),
125 ('jpn', 'Japanese'),
126 ('pol', 'Polish'),
127 ('swe', 'Swedish'),
128 ('tgl', 'Tagalog'),
129 ('twi', 'Twi'),
130 ('cmn', 'Chinese (Mandarin)')
131 """)
132 )
134 with open(Path(__file__).parent.parent.parent.parent / "resources" / "timezone_areas.sql-fake", "r") as f:
135 tz_sql = f.read()
137 conn.execute(text(tz_sql))
140def drop_database() -> None:
141 with session_scope() as session:
142 # postgis is required for all the Geographic Information System (GIS) stuff
143 # pg_trgm is required for trigram-based search
144 # btree_gist is required for gist-based exclusion constraints
145 session.execute(
146 text(
147 "DROP SCHEMA IF EXISTS public CASCADE;"
148 "DROP SCHEMA IF EXISTS logging CASCADE;"
149 "DROP EXTENSION IF EXISTS postgis CASCADE;"
150 "CREATE SCHEMA IF NOT EXISTS public;"
151 "CREATE SCHEMA IF NOT EXISTS logging;"
152 "CREATE EXTENSION postgis;"
153 "CREATE EXTENSION pg_trgm;"
154 "CREATE EXTENSION btree_gist;"
155 "CREATE EXTENSION pg_stat_statements;"
156 )
157 )
160@contextmanager
161def autocommit_engine(url: str):
162 """
163 An engine that executes every statement in a transaction. Mainly needed
164 because CREATE/DROP DATABASE cannot be executed any other way.
165 """
166 engine = create_engine(
167 url,
168 isolation_level="AUTOCOMMIT",
169 )
170 yield engine
171 engine.dispose()
174def make_user(moderation_state_id: int = 0, **kwargs: Any) -> User:
175 """Build an unsaved user. Pass a real moderation_state_id if you intend to save it."""
176 username = "test_user_" + random_hex(16)
178 user = User(
179 moderation_state_id=moderation_state_id,
180 username=username,
181 email=f"{username}@dev.couchers.org",
182 hashed_password=b"$argon2id$v=19$m=65536,t=2,p=1$4cjGg1bRaZ10k+7XbIDmFg$tZG7JaLrkfyfO7cS233ocq7P8rf3znXR7SAfUt34kJg",
183 name=username.capitalize(),
184 hosting_status=HostingStatus.cant_host,
185 city="Testing city",
186 hometown="Test hometown",
187 community_standing=0.5,
188 birthdate=date(year=2000, month=1, day=1),
189 gender="Woman",
190 pronouns="",
191 occupation="Tester",
192 education="UST(esting)",
193 about_me="I test things",
194 things_i_like="Code",
195 about_place="My place has a lot of testing paraphenelia",
196 additional_information="I can be a bit testy",
197 accepted_tos=TOS_VERSION,
198 geom=create_coordinate(40.7108, -73.9740),
199 geom_radius=100,
200 last_onboarding_email_sent=now(),
201 last_donated=now(),
202 )
203 user.accepted_community_guidelines = GUIDELINES_VERSION
204 user.onboarding_emails_sent = 1
206 # Ensure superusers are also editors (DB constraint)
207 if kwargs.get("is_superuser") and "is_editor" not in kwargs:
208 kwargs["is_editor"] = True
210 for key, value in kwargs.items():
211 setattr(user, key, value)
213 return user
216def generate_user(
217 *,
218 delete_user=False,
219 complete_profile=True,
220 strong_verification=False,
221 regions_visited: Sequence[str] = (),
222 regions_lived: Sequence[str] = (),
223 language_abilities: Sequence[tuple[str, LanguageFluency]] = (),
224 **kwargs: Any,
225) -> tuple[User, str]:
226 """
227 Create a new user, return session token
229 The user is detached from any session, and you can access its static attributes, but you can't modify it
231 Use this most of the time
232 """
233 with session_scope() as session:
234 user: User | None = None
236 def create_user(moderation_state_id: int) -> int:
237 nonlocal user
238 user = make_user(moderation_state_id=moderation_state_id, **kwargs)
239 session.add(user)
240 session.flush()
241 return user.id
243 create_moderation(
244 session=session,
245 object_type=ModerationObjectType.user,
246 object_id=create_user,
247 )
248 assert user is not None
250 # Create a profile gallery for the user and link it
251 profile_gallery = PhotoGallery(owner_user_id=user.id)
252 session.add(profile_gallery)
253 session.flush()
254 user.profile_gallery_id = profile_gallery.id
256 for region in regions_visited:
257 session.add(RegionVisited(user_id=user.id, region_code=region))
259 for region in regions_lived:
260 session.add(RegionLived(user_id=user.id, region_code=region))
262 for lang, fluency in language_abilities:
263 session.add(LanguageAbility(user_id=user.id, language_code=lang, fluency=fluency))
265 # this expires the user, so now it's "dirty"
266 context = cast(CouchersContext, _MockCouchersContext())
267 token, _ = create_session(context, session, user, False, set_cookie=False)
269 # deleted user aborts session creation, hence this follows and necessitates a second commit
270 if delete_user:
271 user.deleted_at = now()
273 # spaced out well beyond the per-viewer search noise so search order follows creation order
274 user.recommendation_score = 1e10 - 100 * user.id
276 if complete_profile:
277 key = random_hex(32)
278 session.add(
279 Upload(
280 key=key,
281 filename=random_hex(32) + ".jpg",
282 creator_user_id=user.id,
283 )
284 )
285 session.add(
286 PhotoGalleryItem(
287 gallery_id=profile_gallery.id,
288 upload_key=key,
289 position=0,
290 )
291 )
292 session.flush()
294 user.about_me = "I have a complete profile!\n" * 20
296 if strong_verification:
297 attempt = StrongVerificationAttempt(
298 verification_attempt_token=f"verification_attempt_token_{user.id}",
299 user_id=user.id,
300 status=StrongVerificationAttemptStatus.succeeded,
301 has_full_data=True,
302 passport_encrypted_data=b"not real",
303 passport_date_of_birth=user.birthdate,
304 passport_sex={"Woman": PassportSex.female, "Man": PassportSex.male}.get(
305 user.gender, PassportSex.unspecified
306 ),
307 has_minimal_data=True,
308 passport_expiry_date=date.today() + timedelta(days=10),
309 passport_nationality="UTO",
310 passport_last_three_document_chars=f"{user.id:03}",
311 iris_token=f"iris_token_{user.id}",
312 iris_session_id=user.id,
313 )
314 session.add(attempt)
315 session.flush()
316 assert attempt.has_strong_verification(user)
318 session.commit()
320 assert has_completed_profile(session, user) == complete_profile
322 # refresh it, undoes the expiry
323 session.refresh(user)
325 # this loads the user's timezone info which is lazy loaded, otherwise we'll get issues if we try to refer to it
326 user.timezone # noqa: B018
328 # allows detaches the user from the session, allowing its use outside this session
329 session.expunge(user)
331 return user, token
334def get_user_id_and_token(session: Session, username: str) -> tuple[int, str]:
335 user_id = session.execute(select(User.id).where(User.username == username)).scalar_one()
336 token = session.execute(select(UserSession.token).where(UserSession.user_id == user_id)).scalar_one()
337 return user_id, token
340def make_friends(user1: User, user2: User) -> None:
341 with session_scope() as session:
342 # Create moderation state with VISIBLE status (approved friendship for tests)
343 moderation_state = ModerationState(
344 object_type=ModerationObjectType.friend_request,
345 object_id=0, # Placeholder, will be updated
346 visibility=ModerationVisibility.visible,
347 )
348 session.add(moderation_state)
349 session.flush()
351 friend_relationship = FriendRelationship(
352 from_user_id=user1.id,
353 to_user_id=user2.id,
354 status=FriendStatus.accepted,
355 moderation_state_id=moderation_state.id,
356 )
357 session.add(friend_relationship)
358 session.flush()
360 # Update the moderation state with the actual object id
361 moderation_state.object_id = friend_relationship.id
364def make_user_block(user1: User, user2: User) -> None:
365 with session_scope() as session:
366 user_block = UserBlock(
367 blocking_user_id=user1.id,
368 blocked_user_id=user2.id,
369 )
370 session.add(user_block)
373def make_user_invisible(user_id: int) -> None:
374 with session_scope() as session:
375 session.execute(update(User).where(User.id == user_id).values(banned_at=func.now()))
378def backdate_conversations() -> None:
379 """
380 Shifts every existing conversation back past the duplicate-request window so the next
381 CreateHostRequest to the same host isn't rejected. Shifting them all by the same amount keeps
382 their relative order, which the listing tests rely on.
383 """
384 with session_scope() as session:
385 session.execute(
386 update(Conversation).values(
387 created=Conversation.created - timedelta(hours=HOST_REQUEST_DUPLICATE_WINDOW_HOURS, minutes=1)
388 )
389 )
392# This doubles as get_FriendRequest, since a friend request is just a pending friend relationship
393def get_friend_relationship(user1: User, user2: User) -> FriendRelationship | None:
394 with session_scope() as session:
395 friend_relationship = session.execute(
396 select(FriendRelationship).where(
397 or_(
398 (FriendRelationship.from_user_id == user1.id and FriendRelationship.to_user_id == user2.id),
399 (FriendRelationship.from_user_id == user2.id and FriendRelationship.to_user_id == user1.id),
400 )
401 )
402 ).scalar_one_or_none()
404 session.expunge(friend_relationship)
405 return friend_relationship
408def add_users_to_new_moderation_list(users: list[User]) -> int:
409 """Group users as duplicated accounts"""
410 with session_scope() as session:
411 moderation_user_list = ModerationUserList()
412 session.add(moderation_user_list)
413 session.flush()
414 for user in users:
415 refreshed_user = session.get_one(User, user.id)
416 moderation_user_list.users.append(refreshed_user)
417 return moderation_user_list.id
420def pg_dump_is_available() -> bool:
421 result = subprocess.run(["which", "pg_dump"], stdout=subprocess.PIPE, encoding="ascii")
422 return result.returncode == 0
425def make_volunteer(started_volunteering: date, show_on_team_page: bool = True, **kwargs: Any) -> Volunteer:
426 vol = Volunteer(show_on_team_page=show_on_team_page, **kwargs)
427 vol.started_volunteering = started_volunteering
429 return vol