71 lines
1.9 KiB
Python
71 lines
1.9 KiB
Python
"""Migration integration coverage against a genuinely fresh database."""
|
|
|
|
import pytest
|
|
|
|
|
|
def test_fresh_upgrade_is_complete_idempotent_and_reported(isolatedDatabase, capsys):
|
|
from core import postgres
|
|
from core.migrations import (
|
|
MigrationError,
|
|
discover_migrations,
|
|
migration_status,
|
|
upgrade,
|
|
)
|
|
from core.migrations.__main__ import main
|
|
|
|
discovered = discover_migrations()
|
|
assert discovered
|
|
|
|
applied = upgrade()
|
|
assert [(item.namespace, item.version) for item in applied] == [
|
|
(item.namespace, item.version) for item in discovered
|
|
]
|
|
|
|
expectedTables = {
|
|
"schema_migrations",
|
|
"users",
|
|
"notifications",
|
|
"scheduled_jobs",
|
|
"outbound_messages",
|
|
"provider_identities",
|
|
"api_keys",
|
|
"reminders",
|
|
}
|
|
rows = postgres.execute(
|
|
"""
|
|
SELECT table_name
|
|
FROM information_schema.tables
|
|
WHERE table_schema = 'public'
|
|
"""
|
|
)
|
|
assert expectedTables.issubset({row["table_name"] for row in rows})
|
|
|
|
assert upgrade() == []
|
|
status = migration_status()
|
|
assert len(status) == len(discovered)
|
|
assert {record["state"] for record in status} == {"applied"}
|
|
assert all(record["applied_at"] is not None for record in status)
|
|
|
|
assert main(["status"]) == 0
|
|
output = capsys.readouterr().out
|
|
assert "applied" in output
|
|
assert "core:1" in output
|
|
assert "reminders:1" in output
|
|
|
|
postgres.execute(
|
|
"""
|
|
UPDATE schema_migrations
|
|
SET checksum = %(checksum)s
|
|
WHERE namespace = 'core' AND version = 1
|
|
""",
|
|
{"checksum": "0" * 64},
|
|
)
|
|
changed = migration_status()
|
|
coreBaseline = next(
|
|
row for row in changed if row["namespace"] == "core" and row["version"] == 1
|
|
)
|
|
assert coreBaseline["state"] == "changed"
|
|
assert main(["status"]) == 1
|
|
with pytest.raises(MigrationError, match="checksum changed"):
|
|
upgrade()
|