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