Coverage for app/backend/src/tests/fixtures/db.py: 100%
136 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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 ('arb', 'Arabic (Standard)'),
116 ('deu', 'German'),
117 ('eng', 'English'),
118 ('fin', 'Finnish'),
119 ('fra', 'French'),
120 ('heb', 'Hebrew'),
121 ('hun', 'Hungarian'),
122 ('jpn', 'Japanese'),
123 ('pol', 'Polish'),
124 ('swe', 'Swedish'),
125 ('cmn', 'Chinese (Mandarin)')
126 """)
127 )
129 with open(Path(__file__).parent.parent.parent.parent / "resources" / "timezone_areas.sql-fake", "r") as f:
130 tz_sql = f.read()
132 conn.execute(text(tz_sql))
135def drop_database() -> None:
136 with session_scope() as session:
137 # postgis is required for all the Geographic Information System (GIS) stuff
138 # pg_trgm is required for trigram-based search
139 # btree_gist is required for gist-based exclusion constraints
140 session.execute(
141 text(
142 "DROP SCHEMA IF EXISTS public CASCADE;"
143 "DROP SCHEMA IF EXISTS logging CASCADE;"
144 "DROP EXTENSION IF EXISTS postgis CASCADE;"
145 "CREATE SCHEMA IF NOT EXISTS public;"
146 "CREATE SCHEMA IF NOT EXISTS logging;"
147 "CREATE EXTENSION postgis;"
148 "CREATE EXTENSION pg_trgm;"
149 "CREATE EXTENSION btree_gist;"
150 "CREATE EXTENSION pg_stat_statements;"
151 )
152 )
155@contextmanager
156def autocommit_engine(url: str):
157 """
158 An engine that executes every statement in a transaction. Mainly needed
159 because CREATE/DROP DATABASE cannot be executed any other way.
160 """
161 engine = create_engine(
162 url,
163 isolation_level="AUTOCOMMIT",
164 )
165 yield engine
166 engine.dispose()
169def make_user(moderation_state_id: int = 0, **kwargs: Any) -> User:
170 """Build an unsaved user. Pass a real moderation_state_id if you intend to save it."""
171 username = "test_user_" + random_hex(16)
173 user = User(
174 moderation_state_id=moderation_state_id,
175 username=username,
176 email=f"{username}@dev.couchers.org",
177 hashed_password=b"$argon2id$v=19$m=65536,t=2,p=1$4cjGg1bRaZ10k+7XbIDmFg$tZG7JaLrkfyfO7cS233ocq7P8rf3znXR7SAfUt34kJg",
178 name=username.capitalize(),
179 hosting_status=HostingStatus.cant_host,
180 city="Testing city",
181 hometown="Test hometown",
182 community_standing=0.5,
183 birthdate=date(year=2000, month=1, day=1),
184 gender="Woman",
185 pronouns="",
186 occupation="Tester",
187 education="UST(esting)",
188 about_me="I test things",
189 things_i_like="Code",
190 about_place="My place has a lot of testing paraphenelia",
191 additional_information="I can be a bit testy",
192 accepted_tos=TOS_VERSION,
193 geom=create_coordinate(40.7108, -73.9740),
194 geom_radius=100,
195 last_onboarding_email_sent=now(),
196 last_donated=now(),
197 )
198 user.accepted_community_guidelines = GUIDELINES_VERSION
199 user.onboarding_emails_sent = 1
201 # Ensure superusers are also editors (DB constraint)
202 if kwargs.get("is_superuser") and "is_editor" not in kwargs:
203 kwargs["is_editor"] = True
205 for key, value in kwargs.items():
206 setattr(user, key, value)
208 return user
211def generate_user(
212 *,
213 delete_user=False,
214 complete_profile=True,
215 strong_verification=False,
216 regions_visited: Sequence[str] = (),
217 regions_lived: Sequence[str] = (),
218 language_abilities: Sequence[tuple[str, LanguageFluency]] = (),
219 **kwargs: Any,
220) -> tuple[User, str]:
221 """
222 Create a new user, return session token
224 The user is detached from any session, and you can access its static attributes, but you can't modify it
226 Use this most of the time
227 """
228 with session_scope() as session:
229 user: User | None = None
231 def create_user(moderation_state_id: int) -> int:
232 nonlocal user
233 user = make_user(moderation_state_id=moderation_state_id, **kwargs)
234 session.add(user)
235 session.flush()
236 return user.id
238 create_moderation(
239 session=session,
240 object_type=ModerationObjectType.user,
241 object_id=create_user,
242 )
243 assert user is not None
245 # Create a profile gallery for the user and link it
246 profile_gallery = PhotoGallery(owner_user_id=user.id)
247 session.add(profile_gallery)
248 session.flush()
249 user.profile_gallery_id = profile_gallery.id
251 for region in regions_visited:
252 session.add(RegionVisited(user_id=user.id, region_code=region))
254 for region in regions_lived:
255 session.add(RegionLived(user_id=user.id, region_code=region))
257 for lang, fluency in language_abilities:
258 session.add(LanguageAbility(user_id=user.id, language_code=lang, fluency=fluency))
260 # this expires the user, so now it's "dirty"
261 context = cast(CouchersContext, _MockCouchersContext())
262 token, _ = create_session(context, session, user, False, set_cookie=False)
264 # deleted user aborts session creation, hence this follows and necessitates a second commit
265 if delete_user:
266 user.deleted_at = now()
268 user.recommendation_score = 1e10 - user.id
270 if complete_profile:
271 key = random_hex(32)
272 session.add(
273 Upload(
274 key=key,
275 filename=random_hex(32) + ".jpg",
276 creator_user_id=user.id,
277 )
278 )
279 session.add(
280 PhotoGalleryItem(
281 gallery_id=profile_gallery.id,
282 upload_key=key,
283 position=0,
284 )
285 )
286 session.flush()
288 user.about_me = "I have a complete profile!\n" * 20
290 if strong_verification:
291 attempt = StrongVerificationAttempt(
292 verification_attempt_token=f"verification_attempt_token_{user.id}",
293 user_id=user.id,
294 status=StrongVerificationAttemptStatus.succeeded,
295 has_full_data=True,
296 passport_encrypted_data=b"not real",
297 passport_date_of_birth=user.birthdate,
298 passport_sex={"Woman": PassportSex.female, "Man": PassportSex.male}.get(
299 user.gender, PassportSex.unspecified
300 ),
301 has_minimal_data=True,
302 passport_expiry_date=date.today() + timedelta(days=10),
303 passport_nationality="UTO",
304 passport_last_three_document_chars=f"{user.id:03}",
305 iris_token=f"iris_token_{user.id}",
306 iris_session_id=user.id,
307 )
308 session.add(attempt)
309 session.flush()
310 assert attempt.has_strong_verification(user)
312 session.commit()
314 assert has_completed_profile(session, user) == complete_profile
316 # refresh it, undoes the expiry
317 session.refresh(user)
319 # this loads the user's timezone info which is lazy loaded, otherwise we'll get issues if we try to refer to it
320 user.timezone # noqa: B018
322 # allows detaches the user from the session, allowing its use outside this session
323 session.expunge(user)
325 return user, token
328def get_user_id_and_token(session: Session, username: str) -> tuple[int, str]:
329 user_id = session.execute(select(User.id).where(User.username == username)).scalar_one()
330 token = session.execute(select(UserSession.token).where(UserSession.user_id == user_id)).scalar_one()
331 return user_id, token
334def make_friends(user1: User, user2: User) -> None:
335 with session_scope() as session:
336 # Create moderation state with VISIBLE status (approved friendship for tests)
337 moderation_state = ModerationState(
338 object_type=ModerationObjectType.friend_request,
339 object_id=0, # Placeholder, will be updated
340 visibility=ModerationVisibility.visible,
341 )
342 session.add(moderation_state)
343 session.flush()
345 friend_relationship = FriendRelationship(
346 from_user_id=user1.id,
347 to_user_id=user2.id,
348 status=FriendStatus.accepted,
349 moderation_state_id=moderation_state.id,
350 )
351 session.add(friend_relationship)
352 session.flush()
354 # Update the moderation state with the actual object id
355 moderation_state.object_id = friend_relationship.id
358def make_user_block(user1: User, user2: User) -> None:
359 with session_scope() as session:
360 user_block = UserBlock(
361 blocking_user_id=user1.id,
362 blocked_user_id=user2.id,
363 )
364 session.add(user_block)
367def make_user_invisible(user_id: int) -> None:
368 with session_scope() as session:
369 session.execute(update(User).where(User.id == user_id).values(banned_at=func.now()))
372def backdate_conversations() -> None:
373 """
374 Shifts every existing conversation back past the duplicate-request window so the next
375 CreateHostRequest to the same host isn't rejected. Shifting them all by the same amount keeps
376 their relative order, which the listing tests rely on.
377 """
378 with session_scope() as session:
379 session.execute(
380 update(Conversation).values(
381 created=Conversation.created - timedelta(hours=HOST_REQUEST_DUPLICATE_WINDOW_HOURS, minutes=1)
382 )
383 )
386# This doubles as get_FriendRequest, since a friend request is just a pending friend relationship
387def get_friend_relationship(user1: User, user2: User) -> FriendRelationship | None:
388 with session_scope() as session:
389 friend_relationship = session.execute(
390 select(FriendRelationship).where(
391 or_(
392 (FriendRelationship.from_user_id == user1.id and FriendRelationship.to_user_id == user2.id),
393 (FriendRelationship.from_user_id == user2.id and FriendRelationship.to_user_id == user1.id),
394 )
395 )
396 ).scalar_one_or_none()
398 session.expunge(friend_relationship)
399 return friend_relationship
402def add_users_to_new_moderation_list(users: list[User]) -> int:
403 """Group users as duplicated accounts"""
404 with session_scope() as session:
405 moderation_user_list = ModerationUserList()
406 session.add(moderation_user_list)
407 session.flush()
408 for user in users:
409 refreshed_user = session.get_one(User, user.id)
410 moderation_user_list.users.append(refreshed_user)
411 return moderation_user_list.id
414def pg_dump_is_available() -> bool:
415 result = subprocess.run(["which", "pg_dump"], stdout=subprocess.PIPE, encoding="ascii")
416 return result.returncode == 0
419def make_volunteer(started_volunteering: date, show_on_team_page: bool = True, **kwargs: Any) -> Volunteer:
420 vol = Volunteer(show_on_team_page=show_on_team_page, **kwargs)
421 vol.started_volunteering = started_volunteering
423 return vol