Coverage for app/backend/src/tests/conftest.py: 97%
194 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-20 21:14 +0000
1import os
2import re
3from collections.abc import Generator
4from tempfile import TemporaryDirectory
5from typing import Any
6from unittest.mock import patch
8import pytest
9from sqlalchemy import Connection, Engine
10from sqlalchemy.sql import text
12# Set up environment variables before any couchers imports (they trigger config loading)
13prometheus_multiproc_dir = TemporaryDirectory()
14os.environ["PROMETHEUS_MULTIPROC_DIR"] = prometheus_multiproc_dir.name
16# Default for running with a database from docker-compose.test.yml.
17if "DATABASE_CONNECTION_STRING" not in os.environ: # pragma: no cover
18 os.environ["DATABASE_CONNECTION_STRING"] = (
19 "postgresql://postgres:06b3890acd2c235c41be0bbfe22f1b386a04bf02eedf8c977486355616be2aa1@localhost:6544/testdb"
20 )
22from couchers import experimentation # noqa: E402
23from couchers.config import config # noqa: E402
24from couchers.models import Base # noqa: E402
25from tests.fixtures.db import ( # noqa: E402
26 autocommit_engine,
27 create_schema_from_models,
28 generate_user,
29 populate_testing_resources,
30)
31from tests.fixtures.misc import EmailCollector, Moderator, PushCollector # noqa: E402
34@pytest.fixture(scope="session")
35def postgres_engine() -> Generator[Engine]:
36 """
37 SQLAlchemy engine connected to "postgres" database.
38 """
39 dsn = config.DATABASE_CONNECTION_STRING
40 if not dsn.endswith("/testdb"): 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true
41 raise RuntimeError(f"DATABASE_CONNECTION_STRING must point to /testdb, but was {dsn}")
43 postgres_dsn = re.sub(r"/testdb$", "/postgres", dsn)
45 with autocommit_engine(postgres_dsn) as engine:
46 yield engine
49@pytest.fixture(scope="session")
50def postgres_conn(postgres_engine: Engine) -> Generator[Connection]:
51 """
52 Acquiring a connection takes time, so we cache it.
53 """
54 with postgres_engine.connect() as conn:
55 yield conn
58@pytest.fixture(scope="session")
59def testdb_engine() -> Generator[Engine]:
60 """
61 SQLAlchemy engine connected to "testdb" database.
62 """
63 dsn = config.DATABASE_CONNECTION_STRING
64 with autocommit_engine(dsn) as engine:
65 yield engine
68@pytest.fixture(scope="session")
69def testdb_conn(testdb_engine: Engine) -> Generator[Connection]:
70 """
71 Connection to testdb for truncating tables between tests.
72 """
73 with testdb_engine.connect() as conn:
74 yield conn
77# Static tables that should not be truncated between tests
78STATIC_TABLES = frozenset({"languages", "timezone_areas", "regions"})
81@pytest.fixture(scope="session")
82def setup_testdb(postgres_conn: Connection, testdb_engine: Engine) -> None:
83 """
84 Creates the test database with all the extensions, tables,
85 and static data (languages, regions, timezones). This is done only once
86 per session. Between tests, we truncate all non-static tables.
87 """
88 # running in non-UTC catches some timezone errors
89 os.environ["TZ"] = "America/New_York"
91 postgres_conn.execute(text("DROP DATABASE IF EXISTS testdb WITH (FORCE)"))
92 postgres_conn.execute(text("CREATE DATABASE testdb"))
94 with testdb_engine.connect() as conn:
95 conn.execute(
96 text(
97 "CREATE SCHEMA logging;"
98 "CREATE EXTENSION IF NOT EXISTS postgis;"
99 "CREATE EXTENSION IF NOT EXISTS pg_trgm;"
100 "CREATE EXTENSION IF NOT EXISTS btree_gist;"
101 "CREATE EXTENSION IF NOT EXISTS pg_stat_statements;"
102 )
103 )
105 create_schema_from_models(testdb_engine)
106 populate_testing_resources(conn)
109def _truncate_non_static_tables(conn: Connection) -> None:
110 """
111 Truncates all non-static tables.
112 Static tables (languages, timezone_areas, regions) are preserved.
113 """
114 tables_to_truncate = []
115 for name in Base.metadata.tables.keys():
116 # Skip static tables
117 if name in STATIC_TABLES:
118 continue
119 # Handle schema-qualified names (e.g., "logging.api_calls" -> logging."api_calls")
120 if "." in name:
121 schema, table = name.split(".", 1)
122 tables_to_truncate.append(f'{schema}."{table}"')
123 else:
124 tables_to_truncate.append(f'"{name}"')
125 if tables_to_truncate: 125 ↛ 130line 125 didn't jump to line 130 because the condition on line 125 was always true
126 conn.execute(text(f"TRUNCATE {', '.join(tables_to_truncate)} RESTART IDENTITY CASCADE"))
128 # Reset standalone sequences, not owned by any table column
129 # (RESTART IDENTITY only resets sequences owned by truncated columns)
130 conn.execute(text("ALTER SEQUENCE communities_seq RESTART WITH 1"))
131 conn.execute(text("ALTER SEQUENCE moderation_seq RESTART WITH 2000000"))
134@pytest.fixture
135def db(setup_testdb: None, testdb_conn: Connection) -> None:
136 """
137 Truncates all non-static tables before each test.
138 Static tables (languages, timezone_areas, regions) are preserved.
139 """
140 _truncate_non_static_tables(testdb_conn)
143@pytest.fixture(scope="class")
144def db_class(setup_testdb: None, testdb_conn: Connection) -> None:
145 """
146 The same as above, but with a different scope. Used in test_communities.py.
147 """
148 _truncate_non_static_tables(testdb_conn)
151# Production gates forced True so tests run as "everything enabled". Used by testconfig and the `flags`
152# fixture; tests flip individual values via `flags`.
153_TEST_FLAG_DEFAULTS: dict[str, Any] = {
154 "test_growthbook_integration": True,
155 "sms_enabled": True,
156 "strong_verification_enabled": True,
157 "log_native_ota_requests": True,
158 "donations_enabled": True,
159 "antibot_enabled": True,
160 "postal_verification_enabled": True,
161 "listmonk_enabled": True,
162 "remove_removed_users_from_mailing_list_enabled": True,
163 "notification_translations_enabled": True,
164 "email_ics_attachments_enabled": True,
165}
168@pytest.fixture(scope="class")
169def testconfig():
170 prevconfig = config.copy()
171 prev_initialized = experimentation._initialized
172 prev_load_local_flags = experimentation._load_local_flags
174 config.IN_TEST = True
176 config.DEV = True
177 config.SECRET = bytes.fromhex("448697d3886aec65830a1ea1497cdf804981e0c260d2f812cf2787c4ed1a262b")
178 config.VERSION = "testing_version"
179 config.BASE_URL = "http://localhost:3000"
180 config.BACKEND_BASE_URL = "http://localhost:8888"
181 config.CONSOLE_BASE_URL = "http://localhost:8888"
182 config.COOKIE_DOMAIN = "localhost"
184 config.SMS_SENDER_ID = "invalid"
186 config.ENABLE_EMAIL = False
187 config.NOTIFICATION_EMAIL_SENDER = "Couchers.org"
188 config.NOTIFICATION_EMAIL_ADDRESS = "notify@couchers.org.invalid"
189 config.NOTIFICATION_PREFIX = "[TEST] "
190 config.REPORTS_EMAIL_RECIPIENT = "reports@couchers.org.invalid"
191 config.CONTRIBUTOR_FORM_EMAIL_RECIPIENT = "forms@couchers.org.invalid"
192 config.MODS_EMAIL_RECIPIENT = "mods@couchers.org.invalid"
194 config.STRIPE_API_KEY = ""
195 config.STRIPE_WEBHOOK_SECRET = ""
196 config.STRIPE_RECURRING_PRODUCT_ID = ""
198 config.IRIS_ID_PUBKEY = ""
199 config.IRIS_ID_SECRET = ""
200 # corresponds to private key e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272
201 config.VERIFICATION_DATA_PUBLIC_KEY = bytes.fromhex(
202 "dd740a2b2a35bf05041a28257ea439b30f76f056f3698000b71e6470cd82275f"
203 )
205 config.MYPOSTCARD_API_KEY = "test-api-key"
206 config.MYPOSTCARD_USERNAME = "test-username"
207 config.MYPOSTCARD_PASSWORD = "test-password"
208 config.MYPOSTCARD_PRODUCT_CODE = "J9GCU"
209 config.MYPOSTCARD_CAMPAIGN_ID = "295"
211 config.SMTP_HOST = "localhost"
212 config.SMTP_PORT = 587
213 config.SMTP_USERNAME = "username"
214 config.SMTP_PASSWORD = "password"
216 config.ENABLE_MEDIA = True
217 config.MEDIA_SERVER_SECRET_KEY = bytes.fromhex("91e29bbacc74fa7e23c5d5f34cca5015cb896e338a620003de94a502a461f4bc")
218 config.MEDIA_SERVER_BEARER_TOKEN = "c02d383897d3b82774ced09c9e17802164c37e7e105d8927553697bf4550e91e"
219 config.MEDIA_SERVER_BASE_URL = "http://localhost:5001"
220 config.MEDIA_SERVER_UPLOAD_BASE_URL = "http://localhost:5001"
222 config.BUG_TOOL_ENABLED = False
223 config.BUG_TOOL_GITHUB_REPO = "org/repo"
224 config.BUG_TOOL_GITHUB_USERNAME = "user"
225 config.BUG_TOOL_GITHUB_TOKEN = "token"
227 config.LISTMONK_BASE_URL = "https://localhost"
228 config.LISTMONK_API_USERNAME = "..."
229 config.LISTMONK_API_KEY = "..."
230 config.LISTMONK_LIST_ID = 3
232 config.PUSH_NOTIFICATIONS_ENABLED = True
233 config.PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY = "uI1DCR4G1AdlmMlPfRLemMxrz9f3h4kvjfnI8K9WsVI"
234 config.PUSH_NOTIFICATIONS_VAPID_SUBJECT = "mailto:testing@couchers.org.invalid"
236 config.ACTIVENESS_PROBES_ENABLED = True
238 # File-override mode; gates forced True via the stubbed loader below. Tests needing GrowthBook use `feature_flags`.
239 config.FEATURE_FLAGS_FILE_OVERRIDE_PATH = "feature-flags.dev.json"
240 config.GROWTHBOOK_API_HOST = "https://cdn.growthbook.io"
241 config.GROWTHBOOK_CLIENT_KEY = ""
242 config.GROWTHBOOK_CACHE_PATH = ""
243 experimentation._initialized = True
244 experimentation._load_local_flags = lambda _path: _TEST_FLAG_DEFAULTS # type: ignore[assignment]
246 # Moderation auto-approval deadline - 0 disables, set in tests that need it
247 config.MODERATION_AUTO_APPROVE_DEADLINE_SECONDS = 0
248 # Bot user ID for automated moderation - will be set to a real user in tests that need it
249 config.MODERATION_BOT_USER_ID = 1
251 # Dev APIs disabled by default in tests
252 config.ENABLE_DEV_APIS = False
254 # Slack notifications disabled by default in tests
255 config.SLACK_ENABLED = False
256 config.SLACK_BOT_TOKEN = ""
257 config.SLACK_DONATIONS_CHANNEL = ""
258 config.SLACK_MERCH_CHANNEL = ""
260 # Profiling disabled by default in tests
261 config.PYROSCOPE_ENABLED = False
262 config.PYROSCOPE_SERVER = "https://localhost"
263 config.PYROSCOPE_AUTH_TOKEN = "token"
265 yield None
267 config.copy_from(prevconfig)
268 experimentation._initialized = prev_initialized
269 experimentation._load_local_flags = prev_load_local_flags
272class Flags:
273 """Test handle for setting feature flag values in file-override mode; see the `flags` fixture."""
275 def __init__(self, values: dict[str, Any]) -> None:
276 self._values = values
278 def set_boolean(self, key: str, value: bool) -> None:
279 self._values[key] = value
281 def set_string(self, key: str, value: str) -> None:
282 self._values[key] = value
284 def set_integer(self, key: str, value: int) -> None:
285 self._values[key] = value
287 def set_float(self, key: str, value: float) -> None:
288 self._values[key] = value
290 def set_object(self, key: str, value: Any) -> None:
291 self._values[key] = value
294@pytest.fixture
295def flags(monkeypatch) -> Flags:
296 """
297 Override feature flag values for a test (file-override mode).
299 Starts from the test defaults (production gates on), so a test flips individual flags:
301 def test_x(flags):
302 flags.set_boolean("test_growthbook_integration", False)
303 """
304 values = dict(_TEST_FLAG_DEFAULTS)
305 monkeypatch.setattr(experimentation, "_load_local_flags", lambda _path: values)
306 monkeypatch.setitem(config, "FEATURE_FLAGS_FILE_OVERRIDE_PATH", "feature-flags.dev.json")
307 return Flags(values)
310class FeatureFlags:
311 """Test handle for controlling feature flag values; see the `feature_flags` fixture."""
313 def __init__(self, features: dict[str, Any]) -> None:
314 self._features = features
316 def set(self, key: str, value: Any) -> None:
317 """Make `key` resolve to `value` for every user (logged in or anonymous)."""
318 self._features[key] = {"defaultValue": value}
320 def set_definition(self, key: str, definition: dict[str, Any]) -> None:
321 """Set a raw GrowthBook feature definition, for exercising rollouts/experiments."""
322 self._features[key] = definition
325@pytest.fixture
326def feature_flags(monkeypatch) -> FeatureFlags:
327 """
328 Enable GrowthBook-mode flag evaluation against an in-memory snapshot; tests set values by key.
330 Usage:
331 def test_x(db, feature_flags):
332 feature_flags.set("my_flag", True)
333 ...
334 """
335 features: dict[str, Any] = {}
336 monkeypatch.setattr(experimentation, "_initialized", True)
337 monkeypatch.setattr(experimentation, "_state", {"features": features, "savedGroups": {}})
338 # Switch to GrowthBook mode (empty override path).
339 monkeypatch.setitem(config, "FEATURE_FLAGS_FILE_OVERRIDE_PATH", "")
340 return FeatureFlags(features)
343@pytest.fixture
344def fast_passwords():
345 # password hashing, by design, takes a lot of time, which slows down the tests.
346 # here we jump through some hoops to make this fast by removing the hashing step
348 def fast_hash(password: bytes) -> bytes:
349 return b"fake hash:" + password
351 def fast_verify(hashed: bytes, password: bytes) -> bool:
352 return hashed == fast_hash(password)
354 with patch("couchers.crypto.nacl.pwhash.verify", fast_verify):
355 with patch("couchers.crypto.nacl.pwhash.str", fast_hash):
356 yield
359@pytest.fixture
360def email_collector():
361 """Captures emails and allows inspecting them."""
363 with EmailCollector() as collector:
364 yield collector
367@pytest.fixture
368def push_collector():
369 """
370 See test_SendTestPushNotification for an example on how to use this fixture
371 """
372 with PushCollector() as collector:
373 yield collector
376@pytest.fixture
377def moderator():
378 """
379 Creates a moderator (superuser) and provides methods to exercise the moderation API.
381 Usage:
382 def test_example(db, moderator):
383 # ... create a host request ...
384 moderator.approve_host_request(host_request_id)
385 """
386 user, token = generate_user(is_superuser=True)
387 yield Moderator(user, token)