1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
"""
The migration chain runs, and it builds the schema the models expect.
`create_all()` is not a migration: it creates missing *tables* and never a
missing *column*, so the schema the suite runs on is not the schema a deployed
hub has. The symptom is one endpoint answering 500 with an HTML body while
everything else works, and an `UndefinedColumn` in the journal.
Two things went unnoticed because nothing ran the chain to the end. The only
test that ran alembic at all stopped at `c3d4e5f6a7b8`, the revision before
`add_email_verification` — and that one wrote PostgreSQL's
`(now() at time zone 'utc')` as a literal server default, which is a syntax
error on SQLite. So the chain could not reach head on the database the suite
and the local-hub workflow both use, and the two newest migrations had been run
by exactly one thing: a production deploy.
Hence the two tests here, and they are cheap enough to stay cheap: **upgrade to
head**, and **compare what that built against `Base.metadata`**. A column added
to a model and to no migration fails the second one; a migration that cannot run
fails the first.
What this does *not* check is PostgreSQL-only behaviour — a default, an index
type or a constraint SQLite accepts and PostgreSQL does not, or the reverse.
Running the chain somewhere is enormously better than running it nowhere, and it
is not the same thing as running it where it ships.
"""
from pathlib import Path
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, inspect
from meshbay_hub.db.models import Base
HUB = Path(__file__).resolve().parents[1]
@pytest.fixture
def migrated(tmp_path, monkeypatch):
"""A database built by the migrations alone, at head."""
db = tmp_path / "hub.db"
monkeypatch.setenv("MESHBAY_DATABASE_URL", f"sqlite+aiosqlite:///{db}")
command.upgrade(Config(str(HUB / "alembic.ini")), "head")
return db
def test_the_chain_reaches_head(migrated):
"""It ran at all — which is the half that was false."""
insp = inspect(create_engine(f"sqlite:///{migrated}"))
assert "alembic_version" in insp.get_table_names()
# A table from the newest revision, so "head" means head and not "as far as
# the last revision anybody happened to run".
assert "login_throttle" in insp.get_table_names()
def test_the_migrated_schema_is_the_one_the_models_expect(migrated):
"""Every table and column, both directions.
Both directions on purpose: a column in the models and in no migration
never reaches production, and a column in the migrations and in no model is
a write nothing performs and a rename somebody abandoned halfway.
"""
insp = inspect(create_engine(f"sqlite:///{migrated}"))
migrated_tables = {
name: {c["name"] for c in insp.get_columns(name)}
for name in insp.get_table_names() if name != "alembic_version"
}
problems: list[str] = []
for name, table in Base.metadata.tables.items():
if name not in migrated_tables:
problems.append(f"{name}: in the models, no migration creates it")
continue
expected = {c.name for c in table.columns}
for missing in sorted(expected - migrated_tables[name]):
problems.append(f"{name}.{missing}: in the models, in no migration")
for extra in sorted(migrated_tables[name] - expected):
problems.append(f"{name}.{extra}: in the migrations, in no model")
for name in sorted(set(migrated_tables) - set(Base.metadata.tables)):
problems.append(f"{name}: a migration creates it, no model uses it")
assert not problems, (
"the deployed schema and the tested one have drifted:\n "
+ "\n ".join(problems))
|