Dirk Riemann
1c60c30084
Lands the async-friendly Alembic env (NullPool, reads IX_POSTGRES_URL), the hand-written 001 migration matching the spec's table layout exactly (CHECK on status, partial index on pending rows, UNIQUE on (client_id, request_id)), the SQLAlchemy 2.0 ORM mapping, and a lazy engine/session factory. The factory reads the URL through ix.config when available; Task 3.2 makes that the only path. Smoke-tested: alembic upgrade head + downgrade base against a live postgres:16 produce the expected table shape and tear down cleanly. Unit tests assert the migration source contains every required column/index so the migration can't drift from spec at import time. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
111 lines
4 KiB
Python
111 lines
4 KiB
Python
"""Hermetic smoke test for the Alembic migration module.
|
|
|
|
Does NOT run ``alembic upgrade head`` — that requires a real database and is
|
|
exercised by the integration suite. This test only verifies the migration
|
|
module's structural integrity:
|
|
|
|
* the initial migration can be imported without side effects,
|
|
* its revision / down_revision pair is well-formed,
|
|
* ``upgrade()`` and ``downgrade()`` are callable,
|
|
* the SQL emitted by ``upgrade()`` mentions every column the spec requires.
|
|
|
|
We capture emitted SQL via ``alembic.op`` in offline mode so we don't need a
|
|
live connection. The point is that callers can look at this one test and know
|
|
the migration won't silently drift from spec §4 at import time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
ALEMBIC_DIR = Path(__file__).resolve().parents[2] / "alembic"
|
|
INITIAL_PATH = ALEMBIC_DIR / "versions" / "001_initial_ix_jobs.py"
|
|
|
|
|
|
def _load_migration_module(path: Path):
|
|
spec = importlib.util.spec_from_file_location(f"_test_migration_{path.stem}", path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_initial_migration_file_exists() -> None:
|
|
assert INITIAL_PATH.exists(), f"missing migration: {INITIAL_PATH}"
|
|
|
|
|
|
def test_initial_migration_revision_ids() -> None:
|
|
module = _load_migration_module(INITIAL_PATH)
|
|
# Revision must be a non-empty string; down_revision must be None for the
|
|
# initial migration (no parent).
|
|
assert isinstance(module.revision, str) and module.revision
|
|
assert module.down_revision is None
|
|
|
|
|
|
def test_initial_migration_has_upgrade_and_downgrade() -> None:
|
|
module = _load_migration_module(INITIAL_PATH)
|
|
assert callable(module.upgrade)
|
|
assert callable(module.downgrade)
|
|
|
|
|
|
def test_initial_migration_source_mentions_required_columns() -> None:
|
|
"""Spec §4 columns must all appear in the migration source.
|
|
|
|
We grep the source file rather than running the migration because running
|
|
it needs Postgres. This is belt-and-braces: if someone renames a column
|
|
they'll see this test fail and go update both sides in lockstep.
|
|
"""
|
|
|
|
source = INITIAL_PATH.read_text(encoding="utf-8")
|
|
for column in (
|
|
"job_id",
|
|
"ix_id",
|
|
"client_id",
|
|
"request_id",
|
|
"status",
|
|
"request",
|
|
"response",
|
|
"callback_url",
|
|
"callback_status",
|
|
"attempts",
|
|
"created_at",
|
|
"started_at",
|
|
"finished_at",
|
|
):
|
|
assert column in source, f"migration missing column {column!r}"
|
|
|
|
|
|
def test_initial_migration_source_mentions_indexes_and_constraint() -> None:
|
|
source = INITIAL_PATH.read_text(encoding="utf-8")
|
|
# Unique correlation index on (client_id, request_id).
|
|
assert "ix_jobs_client_request" in source
|
|
# Partial index on pending rows for the claim query.
|
|
assert "ix_jobs_status_created" in source
|
|
# CHECK constraint on status values.
|
|
assert "pending" in source and "running" in source
|
|
assert "done" in source and "error" in source
|
|
|
|
|
|
def test_models_module_declares_ix_job() -> None:
|
|
"""The ORM model mirrors the migration; both must stay in sync."""
|
|
|
|
from ix.store.models import Base, IxJob
|
|
|
|
assert IxJob.__tablename__ == "ix_jobs"
|
|
# Registered in the shared Base.metadata so alembic autogenerate could
|
|
# in principle see it — we don't rely on autogenerate, but having the
|
|
# model in the shared metadata is what lets integration tests do
|
|
# ``Base.metadata.create_all`` as a fast path when Alembic isn't desired.
|
|
assert "ix_jobs" in Base.metadata.tables
|
|
|
|
|
|
def test_engine_module_exposes_factory() -> None:
|
|
from ix.store.engine import get_engine, reset_engine
|
|
|
|
# The engine factory is lazy and idempotent. We don't actually call
|
|
# ``get_engine()`` here — that would require IX_POSTGRES_URL and a real
|
|
# DB. Just confirm the symbols exist and ``reset_engine`` is safe to call
|
|
# on a cold cache.
|
|
assert callable(get_engine)
|
|
reset_engine() # no-op when nothing is cached
|