Coverage for app/backend/src/tests/test_db.py: 94%
213 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import difflib
2import os
3import re
4import subprocess
5from pathlib import Path
7import pytest
8from google.protobuf import empty_pb2
9from sqlalchemy import select, text
10from sqlalchemy.sql import func
12from couchers.config import config
13from couchers.constants import VALID_NAME_MAX_LENGTH
14from couchers.db import _get_base_engine, apply_migrations, get_parent_node_at_location, session_scope
15from couchers.jobs.handlers import DatabaseInconsistencyError, check_database_consistency
16from couchers.models import User
17from couchers.utils import (
18 is_valid_email,
19 is_valid_name,
20 is_valid_user_id,
21 is_valid_username,
22 parse_date,
23)
24from tests.conftest import TEST_DB_NAME
25from tests.fixtures.db import (
26 create_schema_from_models,
27 drop_database,
28 generate_user,
29 pg_dump_is_available,
30)
31from tests.test_communities import create_1d_point, get_community_id, testing_communities # noqa
34def test_is_valid_user_id() -> None:
35 assert is_valid_user_id("10")
36 assert not is_valid_user_id("1a")
37 assert not is_valid_user_id("01")
40def test_is_valid_email() -> None:
41 assert is_valid_email("a@b.cc")
42 assert is_valid_email("te.st+email.valid@a.org.au.xx.yy")
43 assert is_valid_email("invalid@yahoo.co.uk")
44 assert is_valid_email("user+tag@example.com")
45 assert is_valid_email("first.last@example.com")
46 assert not is_valid_email("invalid@.yahoo.co.uk")
47 assert not is_valid_email("test email@couchers.org")
48 assert not is_valid_email(".testemail@couchers.org")
49 assert not is_valid_email("testemail@couchersorg")
50 assert not is_valid_email("b@xxb....blabla")
51 # dot immediately before @ (the original bug)
52 assert not is_valid_email("user.@example.com")
53 # consecutive dots in local part
54 assert not is_valid_email("user..name@example.com")
57def test_is_valid_username() -> None:
58 assert is_valid_username("user")
59 assert is_valid_username("us")
60 assert is_valid_username("us_er")
61 assert is_valid_username("us_er1")
62 assert not is_valid_username("us_")
63 assert not is_valid_username("u")
64 assert not is_valid_username("1us")
65 assert not is_valid_username("User")
68def test_is_valid_name() -> None:
69 # Basics
70 assert is_valid_name("ab")
71 assert is_valid_name("a b")
72 assert is_valid_name("Jean-Luc")
74 # OK punctuation
75 assert is_valid_name("King K. Rool")
76 assert is_valid_name("Doe, John")
77 assert is_valid_name("Alice & Bob")
78 assert is_valid_name("Alice / Bob")
79 assert is_valid_name("Alice | Bob")
81 # Apostrophes and Quotes
82 assert is_valid_name("O'Connor")
83 assert is_valid_name("William “Bill” Clinton")
84 assert is_valid_name("Sha’Nia Jenkins")
86 # Other scripts
87 assert is_valid_name("孙悟空")
88 assert is_valid_name("Combining Diặcritics")
89 assert is_valid_name("काव्य") # Hindi combining diacritics
90 assert is_valid_name("Meritxell Col·lell") # Catalan middle dot
91 assert is_valid_name("レオナルド・ディカプリオ") # Japanese middle dot
92 assert is_valid_name("Lanaʻi") # Hawaiian ʻokina glottal stop
94 # invalid: too short / too long
95 assert not is_valid_name("a")
96 assert not is_valid_name("a" * (VALID_NAME_MAX_LENGTH + 1))
97 # invalid: only whitespace
98 assert not is_valid_name(" ")
99 assert not is_valid_name("")
100 assert not is_valid_name(" ")
101 assert not is_valid_name(" ")
102 # invalid: leading/trailing whitespace
103 assert not is_valid_name(" leading whitespace")
104 assert not is_valid_name("trailing whitespace ")
105 assert not is_valid_name(" surrounding whitespace ")
106 assert not is_valid_name(chr(0xA0) + "Anne") # leading non-breaking space
107 assert not is_valid_name("Anne" + chr(0xA0)) # trailing non-breaking space
108 assert not is_valid_name("\n")
109 # invalid: disallowed characters
110 assert not is_valid_name("digits123")
111 assert not is_valid_name("email@domain.com")
112 assert not is_valid_name("Frosty the ☃️")
113 assert not is_valid_name("exclamative!")
114 assert not is_valid_name("interrogative?")
115 assert not is_valid_name("under_score")
116 assert not is_valid_name("(╯‵□′)╯︵┻━┻")
119def test_parse_date() -> None:
120 assert parse_date("2020-01-01") is not None
121 assert parse_date("1900-01-01") is not None
122 assert parse_date("2099-01-01") is not None
123 assert not parse_date("2019-02-29")
124 assert not parse_date("2019-22-01")
125 assert not parse_date("2020-1-01")
126 assert not parse_date("20-01-01")
127 assert not parse_date("01-01-2020")
128 assert not parse_date("2020/01/01")
131def test_get_parent_node_at_location(testing_communities):
132 with session_scope() as session:
133 w_id = get_community_id(session, "Global") # 0 to 100
134 c1_id = get_community_id(session, "Country 1") # 0 to 50
135 c1r1_id = get_community_id(session, "Country 1, Region 1") # 0 to 10
136 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1") # 0 to 5
137 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2") # 7 to 10
138 c1r2_id = get_community_id(session, "Country 1, Region 2") # 20 to 25
139 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1") # 21 to 23
140 c2_id = get_community_id(session, "Country 2") # 52 to 100
141 c2r1_id = get_community_id(session, "Country 2, Region 1") # 52 to 71
142 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1") # 53 to 70
144 assert get_parent_node_at_location(session, create_1d_point(1)).id == c1r1c1_id # type: ignore[union-attr]
145 assert get_parent_node_at_location(session, create_1d_point(3)).id == c1r1c1_id # type: ignore[union-attr]
146 assert get_parent_node_at_location(session, create_1d_point(6)).id == c1r1_id # type: ignore[union-attr]
147 assert get_parent_node_at_location(session, create_1d_point(8)).id == c1r1c2_id # type: ignore[union-attr]
148 assert get_parent_node_at_location(session, create_1d_point(15)).id == c1_id # type: ignore[union-attr]
149 assert get_parent_node_at_location(session, create_1d_point(51)).id == w_id # type: ignore[union-attr]
152def pg_dump() -> str:
153 return subprocess.run(
154 ["pg_dump", "-s", config.DATABASE_CONNECTION_STRING], stdout=subprocess.PIPE, encoding="ascii", check=True
155 ).stdout
158def sort_pg_dump_output(output: str) -> str:
159 """Sorts the tables, functions and indices dumped by pg_dump in
160 alphabetic order. Also sorts all lists enclosed with parentheses
161 in alphabetic order.
162 """
163 # Temporary replace newline with another character for easier
164 # pattern matching.
165 s = output.replace("\n", "§")
167 # Parameter lists are enclosed with parentheses and every entry
168 # ends with a comma last on the line.
169 s = re.sub(r" \(§(.*?)§\);", lambda m: " (§" + ",§".join(sorted(m.group(1).split(",§"))) + "§);", s)
171 # The header for all objects (tables, functions, indices, etc.)
172 # seems to all start with two dashes and a space. We don't care
173 # which kind of object it is here.
174 s = "§-- ".join(sorted(s.split("§-- ")))
176 # Switch our temporary newline replacement to real newline.
177 return s.replace("§", "\n")
180def test_sort_pg_dump_output() -> None:
181 assert sort_pg_dump_output(" (\nb,\nc,\na\n);\n") == " (\na,\nb,\nc\n);\n"
184def strip_leading_whitespace(lines: list[str]) -> list[str]:
185 return [s.lstrip() for s in lines]
188@pytest.fixture
189def migration_test_db(postgres_conn):
190 """
191 Points everything at a scratch database for the duration of the test.
193 The schemas compared here should look like production's, and the test database doesn't: its
194 column defaults bind mock.now() so the timewarp fixture can shift the clock. Building somewhere
195 else keeps the mock out of both dumps and leaves the test database alone, which this test would
196 otherwise destroy and have to rebuild.
197 """
198 migration_db_name = f"{TEST_DB_NAME}_migrations"
199 postgres_conn.execute(text(f"DROP DATABASE IF EXISTS {migration_db_name} WITH (FORCE)"))
200 postgres_conn.execute(text(f"CREATE DATABASE {migration_db_name}"))
202 previous_dsn = config.DATABASE_CONNECTION_STRING
203 config.DATABASE_CONNECTION_STRING = previous_dsn.rsplit("/", 1)[0] + "/" + migration_db_name
204 # the cached engine still points at the test database, and clearing the cache would drop it
205 # with its pooled connections still open
206 _get_base_engine().dispose()
207 _get_base_engine.cache_clear()
208 try:
209 yield
210 finally:
211 _get_base_engine().dispose()
212 config.DATABASE_CONNECTION_STRING = previous_dsn
213 _get_base_engine.cache_clear()
214 postgres_conn.execute(text(f"DROP DATABASE IF EXISTS {migration_db_name} WITH (FORCE)"))
217@pytest.mark.skipif(not pg_dump_is_available(), reason="Can't run migration tests without pg_dump")
218def test_migrations(migration_test_db) -> None:
219 """
220 Compares the database schema built up from migrations with the
221 schema built by models.py. Both scenarios are started from an
222 empty database and dumped with pg_dump. Any unexplainable
223 differences in the output are reported in unified diff format and
224 fail the test.
226 Note: this takes about 2 minutes in CI, because the real timezone_areas.sql file
227 is used, and it's big. Locally, timezone_areas.sql-fake is used.
228 """
229 drop_database()
230 # rebuild it with alembic migrations
231 apply_migrations()
233 with_migrations = pg_dump()
235 drop_database()
236 # create everything from the current models, not incrementally
237 # through migrations
238 create_schema_from_models()
240 from_scratch = pg_dump()
242 # Save the raw schemas to files for CI artifacts
243 schema_output_dir = os.environ.get("TEST_SCHEMA_OUTPUT_DIR")
244 if schema_output_dir: 244 ↛ 250line 244 didn't jump to line 250 because the condition on line 244 was always true
245 output_path = Path(schema_output_dir)
246 output_path.mkdir(parents=True, exist_ok=True)
247 (output_path / "schema_from_migrations.sql").write_text(with_migrations)
248 (output_path / "schema_from_models.sql").write_text(from_scratch)
250 def message(s: str) -> list[str]:
251 s = sort_pg_dump_output(s)
253 # filter out alembic tables
254 s = "\n-- ".join(x for x in s.split("\n-- ") if not x.startswith("Name: alembic_"))
256 # filter out \restrict and \unrestrict lines (Postgres 16+)
257 s = "\n".join(
258 line for line in s.splitlines() if not line.startswith("\\restrict") and not line.startswith("\\unrestrict")
259 )
261 return strip_leading_whitespace(s.splitlines())
263 diff = "\n".join(
264 difflib.unified_diff(message(with_migrations), message(from_scratch), fromfile="migrations", tofile="model")
265 )
266 print(diff)
267 success = diff == ""
268 assert success
271def test_slugify(db):
272 with session_scope() as session:
273 assert session.execute(func.slugify("this is a test")).scalar_one() == "this-is-a-test"
274 assert session.execute(func.slugify("this is ä test")).scalar_one() == "this-is-a-test"
275 # nothing here gets converted to ascci by unaccent, so it should be empty
276 assert session.execute(func.slugify("Создай группу своего города")).scalar_one() == "slug"
277 assert session.execute(func.slugify("Detta är ett test!")).scalar_one() == "detta-ar-ett-test"
278 assert session.execute(func.slugify("@#(*$&!@#")).scalar_one() == "slug"
279 assert (
280 session.execute(
281 func.slugify("This has a lot ‒ at least relatively speaking ‒ of punctuation! :)")
282 ).scalar_one()
283 == "this-has-a-lot-at-least-relatively-speaking-of-punctuation"
284 )
285 assert (
286 session.execute(func.slugify("Multiple - #@! - non-ascii chars")).scalar_one() == "multiple-non-ascii-chars"
287 )
288 assert session.execute(func.slugify("123")).scalar_one() == "123"
289 assert (
290 session.execute(
291 func.slugify(
292 "A sentence that is over 64 chars long and where the last thing would be replaced by a dash"
293 )
294 ).scalar_one()
295 == "a-sentence-that-is-over-64-chars-long-and-where-the-last-thing"
296 )
299def test_database_consistency_check(db) -> None:
300 """The database consistency check should pass with valid user/gallery setup"""
301 # Create a few users (which auto-creates their profile galleries)
302 generate_user()
303 generate_user()
304 generate_user()
306 # This should not raise any exceptions
307 check_database_consistency(empty_pb2.Empty())
309 # Now break consistency by removing a user's profile gallery
310 with session_scope() as session:
311 user = session.execute(select(User).where(User.deleted_at.is_(None)).limit(1)).scalar_one()
312 user.profile_gallery_id = None
314 # This should now raise an exception
315 with pytest.raises(DatabaseInconsistencyError):
316 check_database_consistency(empty_pb2.Empty())
319def test_migration_ordinals() -> None:
320 """
321 Validates that all migration files use ordinal revision IDs and form a
322 linear chain. Each migration NNNN must have:
323 - revision = "NNNN"
324 - down_revision = "NNNN-1" (or None for 0001)
325 - filename starting with NNNN_
326 """
327 versions_dir = Path(__file__).parent.parent / "couchers" / "migrations" / "versions"
329 migration_files = sorted(f for f in versions_dir.glob("*.py") if re.match(r"^\d{4}_", f.name))
330 assert len(migration_files) > 0, f"No migration files found in {versions_dir}"
332 errors = []
333 prev_ordinal = None
335 for path in migration_files:
336 filename_match = re.match(r"^(\d{4})_", path.name)
337 assert filename_match, f"Migration filename does not start with ordinal: {path.name}"
338 file_ordinal = filename_match.group(1)
340 content = path.read_text()
342 rev_match = re.search(r'^revision\s*=\s*"([^"]+)"', content, re.MULTILINE)
343 down_match = re.search(r"^down_revision\s*=\s*(None|\"([^\"]+)\")", content, re.MULTILINE)
345 if not rev_match: 345 ↛ 346line 345 didn't jump to line 346 because the condition on line 345 was never true
346 errors.append(f"{path.name}: missing 'revision' variable")
347 continue
348 if not down_match: 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 errors.append(f"{path.name}: missing 'down_revision' variable")
350 continue
352 revision = rev_match.group(1)
353 down_revision = down_match.group(2) # None if down_revision = None
355 if revision != file_ordinal: 355 ↛ 356line 355 didn't jump to line 356 because the condition on line 355 was never true
356 errors.append(f'{path.name}: revision = "{revision}" does not match filename ordinal "{file_ordinal}"')
358 if file_ordinal == "0001":
359 if down_revision is not None: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true
360 errors.append(f'{path.name}: first migration must have down_revision = None, got "{down_revision}"')
361 else:
362 expected_down = f"{int(file_ordinal) - 1:04d}"
363 if down_revision != expected_down: 363 ↛ 364line 363 didn't jump to line 364 because the condition on line 363 was never true
364 errors.append(f'{path.name}: down_revision = "{down_revision}" but expected "{expected_down}"')
366 # Check for gaps in the sequence
367 expected_ordinal = f"{int(prev_ordinal) + 1:04d}" if prev_ordinal else "0001"
368 if file_ordinal != expected_ordinal: 368 ↛ 369line 368 didn't jump to line 369 because the condition on line 368 was never true
369 errors.append(f"{path.name}: expected ordinal {expected_ordinal}, got {file_ordinal} (gap in sequence)")
371 prev_ordinal = file_ordinal
373 assert not errors, "Migration ordinal errors:\n" + "\n".join(f" - {e}" for e in errors)