Coverage for app / backend / src / tests / conftest.py: 98%

1import os 

2import re 

3from collections.abc import Generator 

4from tempfile import TemporaryDirectory 

5from unittest.mock import patch 

6 

7import pytest 

8from sqlalchemy import Connection, Engine 

9from sqlalchemy.sql import text 

10 

11# Set up environment variables before any couchers imports (they trigger config loading) 

12prometheus_multiproc_dir = TemporaryDirectory() 

13os.environ["PROMETHEUS_MULTIPROC_DIR"] = prometheus_multiproc_dir.name 

14 

15# Default for running with a database from docker-compose.test.yml. 

16if "DATABASE_CONNECTION_STRING" not in os.environ: # pragma: no cover 

17 os.environ["DATABASE_CONNECTION_STRING"] = ( 

18 "postgresql://postgres:06b3890acd2c235c41be0bbfe22f1b386a04bf02eedf8c977486355616be2aa1@localhost:6544/testdb" 

19 ) 

20 

21from couchers.config import config # noqa: E402 

22from couchers.models import Base # noqa: E402 

23from tests.fixtures.db import ( # noqa: E402 

24 autocommit_engine, 

25 create_schema_from_models, 

26 generate_user, 

27 populate_testing_resources, 

28) 

29from tests.fixtures.misc import Moderator, PushCollector # noqa: E402 

30 

31 

32@pytest.fixture(scope="session") 

33def postgres_engine() -> Generator[Engine]: 

34 """ 

35 SQLAlchemy engine connected to "postgres" database. 

36 """ 

37 dsn = config["DATABASE_CONNECTION_STRING"] 

38 if not dsn.endswith("/testdb"): 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 raise RuntimeError(f"DATABASE_CONNECTION_STRING must point to /testdb, but was {dsn}") 

40 

41 postgres_dsn = re.sub(r"/testdb$", "/postgres", dsn) 

42 

43 with autocommit_engine(postgres_dsn) as engine: 

44 yield engine 

45 

46 

47@pytest.fixture(scope="session") 

48def postgres_conn(postgres_engine: Engine) -> Generator[Connection]: 

49 """ 

50 Acquiring a connection takes time, so we cache it. 

51 """ 

52 with postgres_engine.connect() as conn: 

53 yield conn 

54 

55 

56@pytest.fixture(scope="session") 

57def testdb_engine() -> Generator[Engine]: 

58 """ 

59 SQLAlchemy engine connected to "testdb" database. 

60 """ 

61 dsn = config["DATABASE_CONNECTION_STRING"] 

62 with autocommit_engine(dsn) as engine: 

63 yield engine 

64 

65 

66@pytest.fixture(scope="session") 

67def testdb_conn(testdb_engine: Engine) -> Generator[Connection]: 

68 """ 

69 Connection to testdb for truncating tables between tests. 

70 """ 

71 with testdb_engine.connect() as conn: 

72 yield conn 

73 

74 

75# Static tables that should not be truncated between tests 

76STATIC_TABLES = frozenset({"languages", "timezone_areas", "regions"}) 

77 

78 

79@pytest.fixture(scope="session") 

80def setup_testdb(postgres_conn: Connection, testdb_engine: Engine) -> None: 

81 """ 

82 Creates the test database with all the extensions, tables, 

83 and static data (languages, regions, timezones). This is done only once 

84 per session. Between tests, we truncate all non-static tables. 

85 """ 

86 # running in non-UTC catches some timezone errors 

87 os.environ["TZ"] = "America/New_York" 

88 

89 postgres_conn.execute(text("DROP DATABASE IF EXISTS testdb WITH (FORCE)")) 

90 postgres_conn.execute(text("CREATE DATABASE testdb")) 

91 

92 with testdb_engine.connect() as conn: 

93 conn.execute( 

94 text( 

95 "CREATE SCHEMA logging;" 

96 "CREATE EXTENSION IF NOT EXISTS postgis;" 

97 "CREATE EXTENSION IF NOT EXISTS pg_trgm;" 

98 "CREATE EXTENSION IF NOT EXISTS btree_gist;" 

99 "CREATE EXTENSION IF NOT EXISTS pg_stat_statements;" 

100 ) 

101 ) 

102 

103 create_schema_from_models(testdb_engine) 

104 populate_testing_resources(conn) 

105 

106 

107def _truncate_non_static_tables(conn: Connection) -> None: 

108 """ 

109 Truncates all non-static tables. 

110 Static tables (languages, timezone_areas, regions) are preserved. 

111 """ 

112 tables_to_truncate = [] 

113 for name in Base.metadata.tables.keys(): 

114 # Skip static tables 

115 if name in STATIC_TABLES: 

116 continue 

117 # Handle schema-qualified names (e.g., "logging.api_calls" -> logging."api_calls") 

118 if "." in name: 

119 schema, table = name.split(".", 1) 

120 tables_to_truncate.append(f'{schema}."{table}"') 

121 else: 

122 tables_to_truncate.append(f'"{name}"') 

123 if tables_to_truncate: 123 ↛ 128line 123 didn't jump to line 128 because the condition on line 123 was always true

124 conn.execute(text(f"TRUNCATE {', '.join(tables_to_truncate)} RESTART IDENTITY CASCADE")) 

125 

126 # Reset standalone sequences, not owned by any table column 

127 # (RESTART IDENTITY only resets sequences owned by truncated columns) 

128 conn.execute(text("ALTER SEQUENCE communities_seq RESTART WITH 1")) 

129 conn.execute(text("ALTER SEQUENCE moderation_seq RESTART WITH 2000000")) 

130 

131 

132@pytest.fixture 

133def db(setup_testdb: None, testdb_conn: Connection) -> None: 

134 """ 

135 Truncates all non-static tables before each test. 

136 Static tables (languages, timezone_areas, regions) are preserved. 

137 """ 

138 _truncate_non_static_tables(testdb_conn) 

139 

140 

141@pytest.fixture(scope="class") 

142def db_class(setup_testdb: None, testdb_conn: Connection) -> None: 

143 """ 

144 The same as above, but with a different scope. Used in test_communities.py. 

145 """ 

146 _truncate_non_static_tables(testdb_conn) 

147 

148 

149@pytest.fixture(scope="class") 

150def testconfig(): 

151 prevconfig = config.copy() 

152 config.clear() 

153 config.update(prevconfig) 

154 

155 config["IN_TEST"] = True 

156 

157 config["DEV"] = True 

158 config["SECRET"] = bytes.fromhex("448697d3886aec65830a1ea1497cdf804981e0c260d2f812cf2787c4ed1a262b") 

159 config["VERSION"] = "testing_version" 

160 config["BASE_URL"] = "http://localhost:3000" 

161 config["BACKEND_BASE_URL"] = "http://localhost:8888" 

162 config["CONSOLE_BASE_URL"] = "http://localhost:8888" 

163 config["COOKIE_DOMAIN"] = "localhost" 

164 

165 config["ENABLE_SMS"] = False 

166 config["SMS_SENDER_ID"] = "invalid" 

167 

168 config["ENABLE_EMAIL"] = False 

169 config["NOTIFICATION_EMAIL_SENDER"] = "Couchers.org" 

170 config["NOTIFICATION_EMAIL_ADDRESS"] = "notify@couchers.org.invalid" 

171 config["NOTIFICATION_PREFIX"] = "[TEST] " 

172 config["REPORTS_EMAIL_RECIPIENT"] = "reports@couchers.org.invalid" 

173 config["CONTRIBUTOR_FORM_EMAIL_RECIPIENT"] = "forms@couchers.org.invalid" 

174 config["MODS_EMAIL_RECIPIENT"] = "mods@couchers.org.invalid" 

175 

176 config["ENABLE_DONATIONS"] = False 

177 config["STRIPE_API_KEY"] = "" 

178 config["STRIPE_WEBHOOK_SECRET"] = "" 

179 config["STRIPE_RECURRING_PRODUCT_ID"] = "" 

180 

181 config["ENABLE_STRONG_VERIFICATION"] = False 

182 config["IRIS_ID_PUBKEY"] = "" 

183 config["IRIS_ID_SECRET"] = "" 

184 # corresponds to private key e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272 

185 config["VERIFICATION_DATA_PUBLIC_KEY"] = bytes.fromhex( 

186 "dd740a2b2a35bf05041a28257ea439b30f76f056f3698000b71e6470cd82275f" 

187 ) 

188 

189 config["ENABLE_POSTAL_VERIFICATION"] = False 

190 config["MYPOSTCARD_API_KEY"] = "test-api-key" 

191 config["MYPOSTCARD_USERNAME"] = "test-username" 

192 config["MYPOSTCARD_PASSWORD"] = "test-password" 

193 config["MYPOSTCARD_PRODUCT_CODE"] = "J9GCU" 

194 config["MYPOSTCARD_CAMPAIGN_ID"] = "295" 

195 

196 config["SMTP_HOST"] = "localhost" 

197 config["SMTP_PORT"] = 587 

198 config["SMTP_USERNAME"] = "username" 

199 config["SMTP_PASSWORD"] = "password" 

200 

201 config["ENABLE_MEDIA"] = True 

202 config["MEDIA_SERVER_SECRET_KEY"] = bytes.fromhex( 

203 "91e29bbacc74fa7e23c5d5f34cca5015cb896e338a620003de94a502a461f4bc" 

204 ) 

205 config["MEDIA_SERVER_BEARER_TOKEN"] = "c02d383897d3b82774ced09c9e17802164c37e7e105d8927553697bf4550e91e" 

206 config["MEDIA_SERVER_BASE_URL"] = "http://localhost:5001" 

207 config["MEDIA_SERVER_UPLOAD_BASE_URL"] = "http://localhost:5001" 

208 

209 config["BUG_TOOL_ENABLED"] = False 

210 config["BUG_TOOL_GITHUB_REPO"] = "org/repo" 

211 config["BUG_TOOL_GITHUB_USERNAME"] = "user" 

212 config["BUG_TOOL_GITHUB_TOKEN"] = "token" 

213 

214 config["LISTMONK_ENABLED"] = False 

215 config["LISTMONK_BASE_URL"] = "https://localhost" 

216 config["LISTMONK_API_USERNAME"] = "..." 

217 config["LISTMONK_API_KEY"] = "..." 

218 config["LISTMONK_LIST_ID"] = 3 

219 

220 config["PUSH_NOTIFICATIONS_ENABLED"] = True 

221 config["PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY"] = "uI1DCR4G1AdlmMlPfRLemMxrz9f3h4kvjfnI8K9WsVI" 

222 config["PUSH_NOTIFICATIONS_VAPID_SUBJECT"] = "mailto:testing@couchers.org.invalid" 

223 

224 config["ACTIVENESS_PROBES_ENABLED"] = True 

225 

226 config["RECAPTHCA_ENABLED"] = False 

227 config["RECAPTHCA_PROJECT_ID"] = "..." 

228 config["RECAPTHCA_API_KEY"] = "..." 

229 config["RECAPTHCA_SITE_KEY"] = "..." 

230 

231 config["EXPERIMENTATION_ENABLED"] = False 

232 config["EXPERIMENTATION_PASS_ALL_GATES"] = True 

233 config["STATSIG_SERVER_SECRET_KEY"] = "" 

234 config["STATSIG_ENVIRONMENT"] = "testing" 

235 

236 # Moderation auto-approval deadline - 0 disables, set in tests that need it 

237 config["MODERATION_AUTO_APPROVE_DEADLINE_SECONDS"] = 0 

238 # Bot user ID for automated moderation - will be set to a real user in tests that need it 

239 config["MODERATION_BOT_USER_ID"] = 1 

240 

241 # Dev APIs disabled by default in tests 

242 config["ENABLE_DEV_APIS"] = False 

243 

244 # Slack notifications disabled by default in tests 

245 config["SLACK_ENABLED"] = False 

246 config["SLACK_BOT_TOKEN"] = "" 

247 config["SLACK_DONATIONS_CHANNEL"] = "" 

248 config["SLACK_MERCH_CHANNEL"] = "" 

249 

250 config["ENABLE_NOTIFICATION_TRANSLATIONS"] = False 

251 

252 yield None 

253 

254 config.clear() 

255 config.update(prevconfig) 

256 

257 

258@pytest.fixture 

259def fast_passwords(): 

260 # password hashing, by design, takes a lot of time, which slows down the tests. 

261 # here we jump through some hoops to make this fast by removing the hashing step 

262 

263 def fast_hash(password: bytes) -> bytes: 

264 return b"fake hash:" + password 

265 

266 def fast_verify(hashed: bytes, password: bytes) -> bool: 

267 return hashed == fast_hash(password) 

268 

269 with patch("couchers.crypto.nacl.pwhash.verify", fast_verify): 

270 with patch("couchers.crypto.nacl.pwhash.str", fast_hash): 

271 yield 

272 

273 

274@pytest.fixture 

275def push_collector(): 

276 """ 

277 See test_SendTestPushNotification for an example on how to use this fixture 

278 """ 

279 collector = PushCollector() 

280 

281 with patch("couchers.notifications.push._push_to_user", collector.push_to_user): 

282 yield collector 

283 

284 

285@pytest.fixture 

286def moderator(): 

287 """ 

288 Creates a moderator (superuser) and provides methods to exercise the moderation API. 

289 

290 Usage: 

291 def test_example(db, moderator): 

292 # ... create a host request ... 

293 moderator.approve_host_request(host_request_id) 

294 """ 

295 user, token = generate_user(is_superuser=True) 

296 yield Moderator(user, token)