Coverage for app/backend/src/tests/conftest.py: 96%
224 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-16 18:50 +0000
1import os
2import re
3from collections.abc import Generator
4from pathlib import Path
5from tempfile import TemporaryDirectory
6from typing import Any
7from unittest.mock import patch
9import pytest
10from sqlalchemy import Connection, Engine
11from sqlalchemy.sql import text
13# Set up environment variables before any couchers imports (they trigger config loading)
14prometheus_multiproc_dir = TemporaryDirectory()
15os.environ["PROMETHEUS_MULTIPROC_DIR"] = prometheus_multiproc_dir.name
17# Default for running with a database from docker-compose.test.yml.
18if "DATABASE_CONNECTION_STRING" not in os.environ: # pragma: no cover
19 os.environ["DATABASE_CONNECTION_STRING"] = (
20 "postgresql://postgres:06b3890acd2c235c41be0bbfe22f1b386a04bf02eedf8c977486355616be2aa1@localhost:6544/testdb"
21 )
23from couchers import experimentation # noqa: E402
24from couchers.config import config # noqa: E402
25from couchers.db import _get_base_engine # noqa: E402
26from couchers.models import Base # noqa: E402
27from tests.fixtures import query_log # noqa: E402
28from tests.fixtures.db import ( # noqa: E402
29 autocommit_engine,
30 create_schema_from_models,
31 generate_user,
32 populate_testing_resources,
33)
34from tests.fixtures.misc import EmailCollector, Moderator, PushCollector # noqa: E402
35from tests.fixtures.timewarp import ( # noqa: E402
36 FROZEN_TEST_TIME,
37 MOCK_SEARCH_PATH,
38 FrozenTimewarp,
39 Timewarp,
40 create_mock_clock,
41 install_timewarp,
42)
44QUERY_LOG_DIR = Path(__file__).resolve().parents[2] / "test_artifacts" / "queries"
47def pytest_addoption(parser: pytest.Parser) -> None:
48 parser.addoption(
49 "--query-log",
50 action="store_true",
51 help="record every SQL query, grouped by test and by the RPC that issued it, into test_artifacts/queries",
52 )
55def pytest_configure(config: pytest.Config) -> None:
56 if config.getoption("--query-log"):
57 query_log.enable(_get_base_engine())
60def pytest_sessionfinish(session: pytest.Session) -> None:
61 if session.config.getoption("--query-log"):
62 print(f"\nquery log written to {query_log.dump(QUERY_LOG_DIR)}")
65@pytest.fixture(autouse=True)
66def _record_queries_for_test(request: pytest.FixtureRequest) -> Generator[None]:
67 """Attributes every query to the running test. Cheap no-op unless --query-log is on."""
68 if not request.config.getoption("--query-log"):
69 yield
70 return
71 query_log.set_current_test(request.node.nodeid)
72 yield
73 query_log.set_current_test(None)
76@pytest.fixture(scope="session")
77def postgres_engine() -> Generator[Engine]:
78 """
79 SQLAlchemy engine connected to "postgres" database.
80 """
81 dsn = config.DATABASE_CONNECTION_STRING
82 if not dsn.endswith("/testdb"): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise RuntimeError(f"DATABASE_CONNECTION_STRING must point to /testdb, but was {dsn}")
85 postgres_dsn = re.sub(r"/testdb$", "/postgres", dsn)
87 with autocommit_engine(postgres_dsn) as engine:
88 yield engine
91@pytest.fixture(scope="session")
92def postgres_conn(postgres_engine: Engine) -> Generator[Connection]:
93 """
94 Acquiring a connection takes time, so we cache it.
95 """
96 with postgres_engine.connect() as conn:
97 yield conn
100@pytest.fixture(scope="session")
101def testdb_engine() -> Generator[Engine]:
102 """
103 SQLAlchemy engine connected to "testdb" database.
104 """
105 dsn = config.DATABASE_CONNECTION_STRING
106 with autocommit_engine(dsn) as engine:
107 yield engine
110@pytest.fixture(scope="session")
111def testdb_conn(testdb_engine: Engine) -> Generator[Connection]:
112 """
113 Connection to testdb for truncating tables between tests.
114 """
115 with testdb_engine.connect() as conn:
116 yield conn
119# Static tables that should not be truncated between tests
120STATIC_TABLES = frozenset({"languages", "timezone_areas", "regions"})
123@pytest.fixture(scope="session")
124def setup_testdb(postgres_conn: Connection, testdb_engine: Engine) -> None:
125 """
126 Creates the test database with all the extensions, tables,
127 and static data (languages, regions, timezones). This is done only once
128 per session. Between tests, we truncate all non-static tables.
129 """
130 # running in non-UTC catches some timezone errors
131 os.environ["TZ"] = "America/New_York"
133 postgres_conn.execute(text("DROP DATABASE IF EXISTS testdb WITH (FORCE)"))
134 postgres_conn.execute(text("CREATE DATABASE testdb"))
136 # A column DEFAULT resolves now() once, when the column is created, and stores the function
137 # identity forever after; later search_path changes don't reach it. So mock.now() has to
138 # already shadow pg_catalog.now() on every connection before any DDL runs, which means
139 # setting this at the database level here, ahead of the first connect. The mock schema
140 # doesn't exist yet, which postgres tolerates in a search_path.
141 postgres_conn.execute(text(f"ALTER DATABASE testdb SET search_path = {MOCK_SEARCH_PATH}"))
143 with testdb_engine.connect() as conn:
144 conn.execute(
145 text(
146 "CREATE SCHEMA logging;"
147 "CREATE EXTENSION IF NOT EXISTS postgis;"
148 "CREATE EXTENSION IF NOT EXISTS pg_trgm;"
149 "CREATE EXTENSION IF NOT EXISTS btree_gist;"
150 "CREATE EXTENSION IF NOT EXISTS pg_stat_statements;"
151 )
152 )
153 create_mock_clock(conn)
155 create_schema_from_models(testdb_engine)
156 populate_testing_resources(conn)
159def _truncate_non_static_tables(conn: Connection) -> None:
160 """
161 Truncates all non-static tables.
162 Static tables (languages, timezone_areas, regions) are preserved.
163 """
164 tables_to_truncate = []
165 for name in Base.metadata.tables.keys():
166 # Skip static tables
167 if name in STATIC_TABLES:
168 continue
169 # Handle schema-qualified names (e.g., "logging.api_calls" -> logging."api_calls")
170 if "." in name:
171 schema, table = name.split(".", 1)
172 tables_to_truncate.append(f'{schema}."{table}"')
173 else:
174 tables_to_truncate.append(f'"{name}"')
175 if tables_to_truncate: 175 ↛ 180line 175 didn't jump to line 180 because the condition on line 175 was always true
176 conn.execute(text(f"TRUNCATE {', '.join(tables_to_truncate)} RESTART IDENTITY CASCADE"))
178 # Reset standalone sequences, not owned by any table column
179 # (RESTART IDENTITY only resets sequences owned by truncated columns)
180 conn.execute(text("ALTER SEQUENCE communities_seq RESTART WITH 1"))
181 conn.execute(text("ALTER SEQUENCE moderation_seq RESTART WITH 2000000"))
184@pytest.fixture
185def db(setup_testdb: None, testdb_conn: Connection) -> None:
186 """
187 Truncates all non-static tables before each test.
188 Static tables (languages, timezone_areas, regions) are preserved.
189 """
190 _truncate_non_static_tables(testdb_conn)
193@pytest.fixture(scope="class")
194def db_class(setup_testdb: None, testdb_conn: Connection) -> None:
195 """
196 The same as above, but with a different scope. Used in test_communities.py.
197 """
198 _truncate_non_static_tables(testdb_conn)
201@pytest.fixture
202def timewarp() -> Generator[Timewarp]:
203 """
204 Lets a test move the clock, which keeps running from wherever it's put; see Timewarp.
206 Works without `db`, for a test that only reads the clock from python: nothing connects until
207 something runs a query.
208 """
209 yield from install_timewarp(Timewarp())
212@pytest.fixture
213def frozen_timewarp() -> Generator[FrozenTimewarp]:
214 """
215 Like `timewarp`, but the clock is stopped dead at 2020-01-01 UTC and stays stopped wherever it's
216 moved to, so both python and postgres read back exactly the instant the test asked for.
217 """
218 yield from install_timewarp(FrozenTimewarp(FROZEN_TEST_TIME))
221# Production gates forced True so tests run as "everything enabled". Used by testconfig and the `flags`
222# fixture; tests flip individual values via `flags`.
223_TEST_FLAG_DEFAULTS: dict[str, Any] = {
224 "test_growthbook_integration": True,
225 "sms_enabled": True,
226 "strong_verification_enabled": True,
227 "log_native_ota_requests": True,
228 "donations_enabled": True,
229 "antibot_enabled": True,
230 "postal_verification_enabled": True,
231 "listmonk_enabled": True,
232 "remove_removed_users_from_mailing_list_enabled": True,
233 "notification_translations_enabled": True,
234 "email_ics_attachments_enabled": True,
235 "public_trips_enabled": True,
236}
239@pytest.fixture(scope="class")
240def testconfig():
241 prevconfig = config.copy()
242 prev_initialized = experimentation._initialized
243 prev_load_local_flags = experimentation._load_local_flags
245 config.IN_TEST = True
247 config.DEV = True
248 config.SECRET = bytes.fromhex("448697d3886aec65830a1ea1497cdf804981e0c260d2f812cf2787c4ed1a262b")
249 config.VERSION = "testing_version"
250 config.BASE_URL = "http://localhost:3000"
251 config.BACKEND_BASE_URL = "http://localhost:8888"
252 config.CONSOLE_BASE_URL = "http://localhost:8888"
253 config.COOKIE_DOMAIN = "localhost"
255 config.SMS_SENDER_ID = "invalid"
257 config.ENABLE_EMAIL = False
258 config.NOTIFICATION_EMAIL_SENDER = "Couchers.org"
259 config.NOTIFICATION_EMAIL_ADDRESS = "notify@couchers.org.invalid"
260 config.NOTIFICATION_PREFIX = "[TEST] "
261 config.REPORTS_EMAIL_RECIPIENT = "reports@couchers.org.invalid"
262 config.CONTRIBUTOR_FORM_EMAIL_RECIPIENT = "forms@couchers.org.invalid"
263 config.MODS_EMAIL_RECIPIENT = "mods@couchers.org.invalid"
265 config.STRIPE_API_KEY = ""
266 config.STRIPE_WEBHOOK_SECRET = ""
267 config.STRIPE_RECURRING_PRODUCT_ID = ""
269 config.IRIS_ID_PUBKEY = ""
270 config.IRIS_ID_SECRET = ""
271 # corresponds to private key e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272
272 config.VERIFICATION_DATA_PUBLIC_KEY = bytes.fromhex(
273 "dd740a2b2a35bf05041a28257ea439b30f76f056f3698000b71e6470cd82275f"
274 )
276 config.MYPOSTCARD_API_KEY = "test-api-key"
277 config.MYPOSTCARD_USERNAME = "test-username"
278 config.MYPOSTCARD_PASSWORD = "test-password"
279 config.MYPOSTCARD_PRODUCT_CODE = "J9GCU"
280 config.MYPOSTCARD_CAMPAIGN_ID = "295"
282 config.SMTP_HOST = "localhost"
283 config.SMTP_PORT = 587
284 config.SMTP_USERNAME = "username"
285 config.SMTP_PASSWORD = "password"
287 config.ENABLE_MEDIA = True
288 config.MEDIA_SERVER_SECRET_KEY = bytes.fromhex("91e29bbacc74fa7e23c5d5f34cca5015cb896e338a620003de94a502a461f4bc")
289 config.MEDIA_SERVER_BEARER_TOKEN = "c02d383897d3b82774ced09c9e17802164c37e7e105d8927553697bf4550e91e"
290 config.MEDIA_SERVER_BASE_URL = "http://localhost:5001"
291 config.MEDIA_SERVER_UPLOAD_BASE_URL = "http://localhost:5001"
293 config.BUG_TOOL_ENABLED = False
294 config.BUG_TOOL_GITHUB_REPO = "org/repo"
295 config.BUG_TOOL_GITHUB_USERNAME = "user"
296 config.BUG_TOOL_GITHUB_TOKEN = "token"
298 config.SENTRY_FRONTEND_PROJECT_ID = "1234"
300 config.LISTMONK_BASE_URL = "https://localhost"
301 config.LISTMONK_API_USERNAME = "..."
302 config.LISTMONK_API_KEY = "..."
303 config.LISTMONK_LIST_ID = 3
305 config.PUSH_NOTIFICATIONS_ENABLED = True
306 config.PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY = "uI1DCR4G1AdlmMlPfRLemMxrz9f3h4kvjfnI8K9WsVI"
307 config.PUSH_NOTIFICATIONS_VAPID_SUBJECT = "mailto:testing@couchers.org.invalid"
309 config.ACTIVENESS_PROBES_ENABLED = True
311 # File-override mode; gates forced True via the stubbed loader below. Tests needing GrowthBook use `feature_flags`.
312 config.FEATURE_FLAGS_FILE_OVERRIDE_PATH = "feature-flags.dev.json"
313 config.GROWTHBOOK_API_HOST = "https://cdn.growthbook.io"
314 config.GROWTHBOOK_CLIENT_KEY = ""
315 config.GROWTHBOOK_CACHE_PATH = ""
316 experimentation._initialized = True
317 experimentation._load_local_flags = lambda _path: _TEST_FLAG_DEFAULTS # type: ignore[assignment]
319 # Moderation auto-approval deadline - 0 disables, set in tests that need it
320 config.MODERATION_AUTO_APPROVE_DEADLINE_SECONDS = 0
321 # Bot user ID for automated moderation - will be set to a real user in tests that need it
322 config.MODERATION_BOT_USER_ID = 1
324 # Dev APIs disabled by default in tests
325 config.ENABLE_DEV_APIS = False
327 # Slack notifications disabled by default in tests
328 config.SLACK_ENABLED = False
329 config.SLACK_BOT_TOKEN = ""
330 config.SLACK_DONATIONS_CHANNEL = ""
331 config.SLACK_MERCH_CHANNEL = ""
333 # Profiling disabled by default in tests
334 config.PYROSCOPE_ENABLED = False
335 config.PYROSCOPE_SERVER = "https://localhost"
336 config.PYROSCOPE_AUTH_TOKEN = "token"
338 yield None
340 config.copy_from(prevconfig)
341 experimentation._initialized = prev_initialized
342 experimentation._load_local_flags = prev_load_local_flags
345class Flags:
346 """Test handle for setting feature flag values in file-override mode; see the `flags` fixture."""
348 def __init__(self, values: dict[str, Any]) -> None:
349 self._values = values
351 def set_boolean(self, key: str, value: bool) -> None:
352 self._values[key] = value
354 def set_string(self, key: str, value: str) -> None:
355 self._values[key] = value
357 def set_integer(self, key: str, value: int) -> None:
358 self._values[key] = value
360 def set_float(self, key: str, value: float) -> None:
361 self._values[key] = value
363 def set_object(self, key: str, value: Any) -> None:
364 self._values[key] = value
367@pytest.fixture
368def flags(monkeypatch) -> Flags:
369 """
370 Override feature flag values for a test (file-override mode).
372 Starts from the test defaults (production gates on), so a test flips individual flags:
374 def test_x(flags):
375 flags.set_boolean("test_growthbook_integration", False)
376 """
377 values = dict(_TEST_FLAG_DEFAULTS)
378 monkeypatch.setattr(experimentation, "_load_local_flags", lambda _path: values)
379 monkeypatch.setitem(config, "FEATURE_FLAGS_FILE_OVERRIDE_PATH", "feature-flags.dev.json")
380 return Flags(values)
383class FeatureFlags:
384 """Test handle for controlling feature flag values; see the `feature_flags` fixture."""
386 def __init__(self, features: dict[str, Any]) -> None:
387 self._features = features
389 def set(self, key: str, value: Any) -> None:
390 """Make `key` resolve to `value` for every user (logged in or anonymous)."""
391 self._features[key] = {"defaultValue": value}
393 def set_definition(self, key: str, definition: dict[str, Any]) -> None:
394 """Set a raw GrowthBook feature definition, for exercising rollouts/experiments."""
395 self._features[key] = definition
398@pytest.fixture
399def feature_flags(monkeypatch) -> FeatureFlags:
400 """
401 Enable GrowthBook-mode flag evaluation against an in-memory snapshot; tests set values by key.
403 Usage:
404 def test_x(db, feature_flags):
405 feature_flags.set("my_flag", True)
406 ...
407 """
408 features: dict[str, Any] = {}
409 monkeypatch.setattr(experimentation, "_initialized", True)
410 monkeypatch.setattr(experimentation, "_state", {"features": features, "savedGroups": {}})
411 # Switch to GrowthBook mode (empty override path).
412 monkeypatch.setitem(config, "FEATURE_FLAGS_FILE_OVERRIDE_PATH", "")
413 return FeatureFlags(features)
416@pytest.fixture
417def fast_passwords():
418 # password hashing, by design, takes a lot of time, which slows down the tests.
419 # here we jump through some hoops to make this fast by removing the hashing step
421 def fast_hash(password: bytes) -> bytes:
422 return b"fake hash:" + password
424 def fast_verify(hashed: bytes, password: bytes) -> bool:
425 return hashed == fast_hash(password)
427 with patch("couchers.crypto.nacl.pwhash.verify", fast_verify):
428 with patch("couchers.crypto.nacl.pwhash.str", fast_hash):
429 yield
432@pytest.fixture
433def email_collector():
434 """Captures emails and allows inspecting them."""
436 with EmailCollector() as collector:
437 yield collector
440@pytest.fixture
441def push_collector():
442 """
443 See test_SendTestPushNotification for an example on how to use this fixture
444 """
445 with PushCollector() as collector:
446 yield collector
449@pytest.fixture
450def moderator():
451 """
452 Creates a moderator (superuser) and provides methods to exercise the moderation API.
454 Usage:
455 def test_example(db, moderator):
456 # ... create a host request ...
457 moderator.approve_host_request(host_request_id)
458 """
459 user, token = generate_user(is_superuser=True)
460 yield Moderator(user, token)