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