diff --git a/CHANGELOG.md b/CHANGELOG.md index f53a721..2c37ada 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,35 @@ the project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Fixed + +- `migrate` no longer blocks forever on the advisory lock: the chain + session's wait is now bounded — `pg_try_advisory_lock` polled every + 0.5s against a monotonic deadline (default 30s), mirroring `push` — + and `migrate` exposes `--advisory-wait` (API: `advisory_wait=`) to + tune or zero it. An exhausted budget raises a typed + `SqlpushError` instead of hanging on a stuck holder. `stamp` shares + the chain session and gets the same bounded default. +- `stamp` no longer silently refreshes the checksum of a file that was + edited after it was applied/stamped: the recorded checksum is read + first, and a registered-but-different checksum now refuses with a + typed error (first mismatch stops the walk — nothing after it + registers) instead of overwriting the registry. `--force` (API: + `force=True`) accepts the new content, so the chain's edit-detection + integrity survives re-stamping. +- `migrate` now sets a per-file transaction-scoped `lock_timeout` + (default 5s), mirroring `push`: a chain file whose DDL is blocked + behind another transaction's lock fails fast with a typed error + instead of queuing indefinitely. Tunable via `--lock-timeout` + (API: `lock_timeout=`; negative values are rejected up front). +- The sync facade (`ensure_schema` / `migrate` / `stamp` — everything + resolved from a DSN or `AsyncEngine`) now accepts `postgresql+asyncpg` + URLs by translating them to the `postgresql+psycopg` driver + (host, database, credentials and query options preserved) instead of + failing on the async-only driver at connect time. asyncpg is never + required to be installed in the sqlpush process; plain psycopg + targets are untouched. + ## [0.4.1] - 2026-09-02 ### Added diff --git a/src/sqlpush/api.py b/src/sqlpush/api.py index ee868d3..61c0b98 100644 --- a/src/sqlpush/api.py +++ b/src/sqlpush/api.py @@ -139,7 +139,12 @@ def revision( risk = max((op.risk for op in p.operations), key=lambda r: RISK_RANK[r]) ops = [(f"[{op.risk.name}] {op.type} {op.table or '?'}", op.sql) for op in p.operations] out = Path(out_dir) - out.mkdir(parents=True, exist_ok=True) + try: + out.mkdir(parents=True, exist_ok=True) + except OSError as exc: + # same typing as the write below: only SqlpushError subclasses + # escape the API surface (e.g. out_dir under a file) + raise SqlpushError(f"cannot create migrations directory {out}: {exc}") from exc rev_id = next_revision_id(out) slug = re.sub(r"[^a-z0-9_]+", "_", (message or "migration").lower())[:40] path = out / f"{rev_id}_{slug}.sql" @@ -163,16 +168,32 @@ def revision( return path -def migrate(target, *, chain_dir="migrations/versions", allow_destructive=False) -> MigrateReport: +def migrate( + target, + *, + chain_dir="migrations/versions", + allow_destructive=False, + advisory_wait=30.0, + lock_timeout=5.0, +) -> MigrateReport: """Replay annotated-SQL chain files with gates + same-txn bookkeeping. ``target`` is a DSN string, sync ``Engine`` or ``AsyncEngine`` (resolved - via ``_sync_engine_from``; engines created here are disposed). See + via ``_sync_engine_from``; engines created here are disposed). + ``advisory_wait`` bounds the advisory-lock wait and ``lock_timeout`` + bounds each per-file transaction's lock wait (seconds; 0 = fail + immediately — same contract as ``push``). See ``chain.migrate.run_migrate`` for the execution contract. """ engine, dispose = _sync_engine_from(target) try: - return run_migrate(engine, chain_dir=chain_dir, allow_destructive=allow_destructive) + return run_migrate( + engine, + chain_dir=chain_dir, + allow_destructive=allow_destructive, + advisory_wait=advisory_wait, + lock_timeout=lock_timeout, + ) except SQLAlchemyError as exc: # MigrationFileError/SqlpushError (typed) pass through untouched _raise_typed(exc) @@ -181,16 +202,19 @@ def migrate(target, *, chain_dir="migrations/versions", allow_destructive=False) engine.dispose() -def stamp(target, *, chain_dir="migrations/versions") -> MigrateReport: +def stamp(target, *, chain_dir="migrations/versions", force: bool = False) -> MigrateReport: """Bootstrap: register chain files as applied WITHOUT executing SQL. - For adopting a DB whose schema already reflects the chain. ``target`` - resolution and error typing match :func:`migrate`. See - ``chain.migrate.run_stamp`` for the report convention. + For adopting a DB whose schema already reflects the chain. A file + already registered with a different checksum (edited after apply) + is refused unless ``force`` is set — edit detection survives + re-stamping. ``target`` resolution and error typing match + :func:`migrate`. See ``chain.migrate.run_stamp`` for the report + convention. """ engine, dispose = _sync_engine_from(target) try: - return run_stamp(engine, chain_dir=chain_dir) + return run_stamp(engine, chain_dir=chain_dir, force=force) except SQLAlchemyError as exc: # MigrationFileError/SqlpushError (typed) pass through untouched _raise_typed(exc) @@ -199,18 +223,35 @@ def stamp(target, *, chain_dir="migrations/versions") -> MigrateReport: engine.dispose() +def _translate_asyncpg(url): + """Map a ``postgresql+asyncpg`` URL onto the sync psycopg driver. + + An asyncpg URL cannot back a SYNC engine (the dialect is + async-only), but psycopg is a runtime dependency — the translated + engine actually connects. asyncpg itself is never required in this + process. Everything but the driver (host/db/credentials/query + options) is preserved via the URL API, no string surgery. + """ + if url.drivername == "postgresql+asyncpg": + return url.set(drivername="postgresql+psycopg") + return url + + def _sync_engine_from(target): if isinstance(target, AsyncEngine): from sqlalchemy import create_engine from sqlalchemy.pool import NullPool - dsn = target.url.render_as_string(hide_password=False) - return create_engine(dsn, poolclass=NullPool), True + dsn = _translate_asyncpg(target.url) + return create_engine(dsn.render_as_string(hide_password=False), poolclass=NullPool), True if isinstance(target, str): from sqlalchemy import create_engine + from sqlalchemy.engine import make_url from sqlalchemy.pool import NullPool - return create_engine(target, poolclass=NullPool), True + # make_url raises the same ArgumentError a bad DSN would raise + # inside create_engine — error typing downstream is unchanged + return create_engine(_translate_asyncpg(make_url(target)), poolclass=NullPool), True return target, False diff --git a/src/sqlpush/chain/migrate.py b/src/sqlpush/chain/migrate.py index b2f2c30..1c93c75 100644 --- a/src/sqlpush/chain/migrate.py +++ b/src/sqlpush/chain/migrate.py @@ -11,13 +11,15 @@ Fail-loud ordering: any blocked file (parse error, checksum mismatch, destructive gate, SQL failure) stops the chain — nothing later runs (R4). -A hung migrate (stuck waiting on the advisory lock) is diagnosed via -pg_locks / pg_stat_activity. +The advisory lock wait is BOUNDED (--advisory-wait, default 30s): an +exhausted budget raises a typed error instead of hanging on a stuck +holder — diagnose the holder via pg_locks / pg_stat_activity. """ from __future__ import annotations import contextlib +import time from collections.abc import Iterator from pathlib import Path @@ -26,7 +28,7 @@ from sqlpush.apply.executor import advisory_key from sqlpush.chain.format import MigrationFileError, checksum, parse_migration_file -from sqlpush.types import MigrateReport, RiskClass +from sqlpush.types import MigrateReport, RiskClass, SqlpushError _VERSIONS_DDL = ( "CREATE TABLE IF NOT EXISTS public.sqlpush_versions (" @@ -45,18 +47,33 @@ def _chain_files(chain_dir: str | Path) -> list[Path]: @contextlib.contextmanager -def _chain_session(engine: Engine) -> Iterator[Connection]: +def _chain_session(engine: Engine, *, advisory_wait: float = 30.0) -> Iterator[Connection]: """Session-scoped advisory lock + versions table, shared by every verb. Same key derivation as push (fnv1a_32(b"sqlpush") ^ db oid): serializes concurrent chain workers and excludes push on the same database. The - txn opened by the key query is committed right away — session advisory - locks survive COMMIT/ROLLBACK, and an idle-in-transaction session is - exposed to idle_in_transaction_session_timeout (executor.py note). + wait is BOUNDED, mirroring executor.with_advisory_lock: + ``pg_try_advisory_lock`` polled every 0.5 s against a + ``time.monotonic()`` deadline; an exhausted budget raises + :class:`SqlpushError` — a blocking ``pg_advisory_lock`` would hang + forever on a stuck holder. The txn opened by the key/probe queries + is committed right away — session advisory locks survive + COMMIT/ROLLBACK, and an idle-in-transaction session is exposed to + idle_in_transaction_session_timeout (executor.py note). """ + if advisory_wait < 0: + raise SqlpushError(f"advisory_wait must be >= 0, got {advisory_wait}") with engine.connect() as conn: key = advisory_key(conn) - conn.execute(text("SELECT pg_advisory_lock(:k)"), {"k": key}) + deadline = time.monotonic() + advisory_wait + locked = conn.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": key}).scalar() + while not locked and time.monotonic() < deadline: + time.sleep(0.5) + locked = conn.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": key}).scalar() + if not locked: + raise SqlpushError( + f"another sqlpush worker holds the advisory lock after {advisory_wait}s" + ) conn.commit() try: conn.execute(text(_VERSIONS_DDL)) @@ -73,14 +90,26 @@ def _chain_session(engine: Engine) -> Iterator[Connection]: conn.commit() -def run_migrate(engine: Engine, *, chain_dir: str | Path, allow_destructive: bool) -> MigrateReport: +def run_migrate( + engine: Engine, + *, + chain_dir: str | Path, + allow_destructive: bool, + advisory_wait: float = 30.0, + lock_timeout: float = 5.0, +) -> MigrateReport: + if lock_timeout < 0: + # same contract as push (executor.with_advisory_lock): budgets + # are typed floats, never user input, and a negative one must + # fail before any file or connection work + raise SqlpushError(f"lock_timeout must be >= 0, got {lock_timeout}") applied: list[str] = [] skipped: list[str] = [] blocked: list[str] = [] notes: list[str] = [] partial = False chain = _chain_files(chain_dir) - with _chain_session(engine) as conn: + with _chain_session(engine, advisory_wait=advisory_wait) as conn: recorded = { row[0]: row[1] for row in conn.execute(text("SELECT name, sha256 FROM public.sqlpush_versions")) @@ -107,6 +136,14 @@ def run_migrate(engine: Engine, *, chain_dir: str | Path, allow_destructive: boo break try: with conn.begin(): + # txn-scoped: SET LOCAL dies with the per-file txn + # (style: push's transactional segment, executor.py). + # NOTE: PostgreSQL does not accept bind parameters for + # SET (utility statement), so the int is inlined; + # lock_timeout is a typed float parameter, not user + # input. A chain file blocked behind another + # transaction's lock fails fast instead of queuing. + conn.execute(text(f"SET LOCAL lock_timeout = {int(lock_timeout * 1000)}")) # whole-file replay: exec_driver_sql bypasses text()'s # bind-param parsing entirely — ":casts" and ":=" in # hand-edited SQL must reach the server verbatim @@ -130,14 +167,18 @@ def run_migrate(engine: Engine, *, chain_dir: str | Path, allow_destructive: boo ) -def run_stamp(engine: Engine, *, chain_dir: str | Path) -> MigrateReport: +def run_stamp(engine: Engine, *, chain_dir: str | Path, force: bool = False) -> MigrateReport: """Register every parseable chain file WITHOUT executing any SQL. Bootstrap seam (spec §5 R7): adopt a DB whose schema already reflects the chain. Only the header must parse (fail-loud on THAT) — invalid SQL - in a body still registers, because stamp never executes anything. Each - registration is an idempotent upsert (``ON CONFLICT (name) DO UPDATE``), - so re-stamping refreshes checksums instead of failing. + in a body still registers, because stamp never executes anything. A + file already recorded with a DIFFERENT checksum is refused — it was + edited after apply/stamp, and silently refreshing would wipe the + edit-detection integrity — unless ``force`` is set; the first mismatch + raises and nothing after it registers (strict order, same as migrate). + Unrecorded files, unchanged re-stamps and forced re-stamps upsert + idempotently (``ON CONFLICT (name) DO UPDATE``). Report convention: registered files are listed in ``skipped`` (stamp never populates ``applied`` and never sets ``partial_failure``); a @@ -149,6 +190,11 @@ def run_stamp(engine: Engine, *, chain_dir: str | Path) -> MigrateReport: notes: list[str] = [] chain = _chain_files(chain_dir) with _chain_session(engine) as conn: + recorded = { + row[0]: row[1] + for row in conn.execute(text("SELECT name, sha256 FROM public.sqlpush_versions")) + } + conn.commit() for f in chain: raw = f.read_text() try: @@ -157,6 +203,11 @@ def run_stamp(engine: Engine, *, chain_dir: str | Path) -> MigrateReport: blocked.append(f.name) notes.append(f"{f.name}: {exc}") break # orden estricto: nada posterior se registra + if f.name in recorded and recorded[f.name] != checksum(raw) and not force: + raise SqlpushError( + f"{f.name}: checksum mismatch (file edited after apply?); " + "pass force=True/--force to accept the new content" + ) with conn.begin(): conn.execute( text( diff --git a/src/sqlpush/cli.py b/src/sqlpush/cli.py index 50bf395..7e47a25 100644 --- a/src/sqlpush/cli.py +++ b/src/sqlpush/cli.py @@ -4,7 +4,8 @@ Exit codes: diff always 0; check 0 clean / 2 drift / 3 destructive drift; push 0 applied / 1 destructive blocked / 2 error (incl. partial failure); revision 0 written / 1 error (empty drift refuses); migrate 0 clean / -1 blocked or partial failure. +1 blocked or partial failure; stamp 0 registered / 1 blocked or refused +(an edited file without --force raises a typed error: exit 1). """ from __future__ import annotations @@ -250,12 +251,20 @@ def revision( def migrate( dsn: str | None = typer.Option(None), allow_destructive: bool = typer.Option(False, "--allow-destructive"), + advisory_wait: float = typer.Option(30.0, "--advisory-wait"), + lock_timeout: float = typer.Option(5.0, "--lock-timeout"), out_dir: DirOpt = Path("migrations/versions"), ): """Replay pending migration files (gates + checksum bookkeeping).""" engine = _engine(dsn) try: - report = api.migrate(engine, chain_dir=out_dir, allow_destructive=allow_destructive) + report = api.migrate( + engine, + chain_dir=out_dir, + allow_destructive=allow_destructive, + advisory_wait=advisory_wait, + lock_timeout=lock_timeout, + ) finally: engine.dispose() typer.echo( @@ -272,12 +281,13 @@ def migrate( @app.command() def stamp( dsn: str | None = typer.Option(None), + force: bool = typer.Option(False, "--force"), out_dir: DirOpt = Path("migrations/versions"), ): """Adopt an existing DB: register chain files without executing SQL.""" engine = _engine(dsn) try: - report = api.stamp(engine, chain_dir=out_dir) + report = api.stamp(engine, chain_dir=out_dir, force=force) finally: engine.dispose() typer.echo( diff --git a/src/sqlpush/core/diff.py b/src/sqlpush/core/diff.py index 065bf1e..69f2bbf 100644 --- a/src/sqlpush/core/diff.py +++ b/src/sqlpush/core/diff.py @@ -405,7 +405,9 @@ def _dedup_enum_types(ops: list[PlannedOperation]) -> list[PlannedOperation]: copy stays embedded in its original add_table render. A metadata that defines one type name two DIFFERENT ways keeps both statements and still fails loudly at apply time: silent first-win would mask a - genuine contradiction. + genuine contradiction. Ops whose statements all survive pass + through as the ORIGINAL object (identity) — a re-joined copy of + unchanged SQL buys nothing. """ seen: set[str] = set() @@ -420,10 +422,12 @@ def _is_type_stmt(norm: str) -> bool: continue stmts = _split_statements(op.sql) kept: list[str] = [] + dropped = False for stmt in stmts: norm = " ".join(stmt.split()) if _is_type_stmt(norm): if norm in seen: + dropped = True continue seen.add(norm) if norm: @@ -433,6 +437,9 @@ def _is_type_stmt(norm: str) -> bool: # survives); guard anyway — an empty op.sql is worse than # the duplicate it replaced continue + if not dropped: + out.append(op) + continue out.append( PlannedOperation( type=op.type, diff --git a/tests/test_api.py b/tests/test_api.py index 4cf498f..b9365cf 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -239,6 +239,89 @@ def test_ensure_schema_async_engine(pg_engine, md): asyncio.run(aengine.dispose()) +def test_sync_engine_from_translates_asyncpg_dsn(pg_engine): + # B2: a postgresql+asyncpg DSN string must yield a WORKING sync + # engine on psycopg (runtime dep) — asyncpg is not installed in this + # environment, so connecting through it proves the translation. + from sqlalchemy.engine import make_url + + from sqlpush.api import _sync_engine_from + + asyncpg_dsn = make_url(pg_engine.url).set(drivername="postgresql+asyncpg") + engine, dispose = _sync_engine_from(asyncpg_dsn.render_as_string(hide_password=False)) + assert dispose + try: + assert engine.dialect.driver == "psycopg" + assert engine.url.drivername == "postgresql+psycopg" + with engine.connect() as conn: + assert conn.execute(text("SELECT 1")).scalar() == 1 + finally: + engine.dispose() + + +def test_asyncpg_url_translation_preserves_components(pg_engine): + # An AsyncEngine over asyncpg cannot even be constructed here + # (create_async_engine imports the driver — verified absent), so the + # AsyncEngine-branch translation is pinned at the URL level: driver + # swapped, every other component preserved, non-asyncpg passthrough. + from sqlalchemy.engine import URL + + from sqlpush.api import _translate_asyncpg + + url = URL.create( + "postgresql+asyncpg", + username="sqlpush", + password="sqlpush", + host="localhost", + port=5433, + database="sqlpush_test", + query={"connect_timeout": "10"}, + ) + out = _translate_asyncpg(url) + assert out.drivername == "postgresql+psycopg" + assert (out.username, out.password, out.host, out.port, out.database) == ( + "sqlpush", + "sqlpush", + "localhost", + 5433, + "sqlpush_test", + ) + assert dict(out.query) == {"connect_timeout": "10"} + plain = URL.create("postgresql+psycopg", host="localhost", database="x") + assert _translate_asyncpg(plain) is plain + + +def test_sync_engine_from_plain_psycopg_unchanged(pg_engine): + # Non-asyncpg targets keep the exact pre-B2 behavior: str DSN and + # plain-psycopg AsyncEngine still resolve to working disposable + # psycopg engines, and a sync Engine passes through untouched. + import asyncio + + from sqlalchemy.ext.asyncio import create_async_engine + + from sqlpush.api import _sync_engine_from + + dsn = pg_engine.url.render_as_string(hide_password=False) + eng, dispose = _sync_engine_from(dsn) + assert dispose and eng.url.drivername == "postgresql+psycopg" + with eng.connect() as conn: + assert conn.execute(text("SELECT 1")).scalar() == 1 + eng.dispose() + + aeng = create_async_engine(dsn) + try: + eng2, dispose2 = _sync_engine_from(aeng) + assert dispose2 and eng2.url.drivername == "postgresql+psycopg" + with eng2.connect() as conn: + assert conn.execute(text("SELECT 1")).scalar() == 1 + eng2.dispose() + finally: + asyncio.run(aeng.dispose()) + + eng3, dispose3 = _sync_engine_from(pg_engine) + assert eng3 is pg_engine and not dispose3 + + @pytest.mark.timescale def test_push_default_locked_path_applies_hypertable(pg_engine, md_ht): # Regression pin: push() defaults to lock=True, whose winner path diff --git a/tests/test_cli.py b/tests/test_cli.py index 027a39e..d469205 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -217,6 +217,64 @@ def test_dsn_env_fallback_and_missing(tmp_path, monkeypatch, hero): assert r2.exit_code == 1 +# --- migrate/stamp verbs (0.4.2 hardening) --------------------------------- + +CLI_SAFE_0001 = ( + "-- sqlpush: revision=0001 risk=SAFE\nCREATE TABLE cli_mt (id integer PRIMARY KEY);\n" +) + + +@pytest.fixture() +def cli_chain(pg_engine): + def _clean() -> None: + with pg_engine.begin() as conn: + for t in ("sqlpush_versions", "cli_mt"): + conn.execute(text(f"DROP TABLE IF EXISTS {t}")) + + _clean() + yield pg_engine + _clean() + + +@pytest.mark.pg +def test_cli_migrate_new_flags_parse_and_succeed(cli_chain, tmp_path): + # --advisory-wait / --lock-timeout parse and route through; no + # holder + an empty (existing) dir is a legitimate idle run -> 0 + tmp_path.mkdir(exist_ok=True) + r = runner.invoke( + app, + [ + "migrate", + "--dsn", + DSN, + "--dir", + str(tmp_path), + "--advisory-wait", + "3", + "--lock-timeout", + "2", + ], + ) + assert r.exit_code == 0 + + +@pytest.mark.pg +def test_cli_stamp_force_on_edited_file(cli_chain, tmp_path): + # B4 via the verb: stamp, edit the file, re-stamp without --force -> + # typed SqlpushError (exit 1 via main()'s fallback); with --force -> + # exit 0 with the checksum refreshed + (tmp_path / "0001_init.sql").write_text(CLI_SAFE_0001) + r0 = runner.invoke(app, ["stamp", *DSN_ARG, "--dir", str(tmp_path)]) + assert r0.exit_code == 0 + (tmp_path / "0001_init.sql").write_text(CLI_SAFE_0001.replace("cli_mt", "cli_mt_ed")) + r1 = runner.invoke(app, ["stamp", *DSN_ARG, "--dir", str(tmp_path)]) + assert r1.exit_code == 1 + assert isinstance(r1.exception, SqlpushError) + assert "0001_init.sql" in str(r1.exception) + r2 = runner.invoke(app, ["stamp", *DSN_ARG, "--dir", str(tmp_path), "--force"]) + assert r2.exit_code == 0 + + # --- unit tests below: api.push monkeypatched, no live PostgreSQL --- diff --git a/tests/test_diff.py b/tests/test_diff.py index 6b0a87d..e2f095d 100644 --- a/tests/test_diff.py +++ b/tests/test_diff.py @@ -574,3 +574,49 @@ def test_timescale_born_time_index_equals_declared(pg_engine): finally: with pg_engine.begin() as conn: conn.execute(text("DROP TABLE IF EXISTS tsborn")) + + +@pytest.mark.timescale +def test_timescale_born_time_index_different_columns_still_drift(pg_engine): + # Boundary of the S1 equal-case pin above: same qualified index name + # and owning hypertable, but a DIFFERENT column sequence (m_cols != + # r_cols) must NOT be pruned as birth state — the drop+add pair is + # genuine drift and has to surface, exactly the way a reordered or + # extra-column declaration does. + with pg_engine.begin() as conn: + conn.execute(text("DROP TABLE IF EXISTS tsborn2")) + conn.execute( + text( + "CREATE TABLE tsborn2 (" + "id integer, ts timestamptz NOT NULL, payload integer, " + "PRIMARY KEY (id, ts))" + ) + ) + conn.execute(text("SELECT create_hypertable('tsborn2', 'ts')")) + try: + with pg_engine.connect() as conn: + idxdef = conn.execute( + text("SELECT indexdef FROM pg_indexes WHERE indexname = 'tsborn2_ts_idx'") + ).scalar() + # sanity: the auto time index exists (on ts, born DESC) — the + # metadata declaration below deliberately diverges from it + assert idxdef is not None and "DESC" in idxdef, idxdef + + md = MetaData() + Table( + "tsborn2", + md, + Column("id", Integer, primary_key=True), + Column("ts", DateTime(timezone=True), nullable=False, primary_key=True), + Column("payload", Integer), + # same NAME as the auto time index, different column set + Index("tsborn2_ts_idx", "payload"), + ) + plan = DiffEngine().plan(md, pg_engine) + assert plan.drift + # the reported drift is exactly the index pair — not pruned + # (silently equal), not anything else + assert sorted(op.type for op in plan.operations) == ["add_index", "drop_index"] + finally: + with pg_engine.begin() as conn: + conn.execute(text("DROP TABLE IF EXISTS tsborn2")) diff --git a/tests/test_enum_dedup.py b/tests/test_enum_dedup.py index 0b27287..9f98855 100644 --- a/tests/test_enum_dedup.py +++ b/tests/test_enum_dedup.py @@ -4,14 +4,16 @@ import pytest from sqlalchemy import Column, Enum, Integer, MetaData, Table, text -from sqlpush.api import migrate, plan, revision - -pytestmark = pytest.mark.pg +from sqlpush.api import check, migrate, plan, push, revision +from sqlpush.core.diff import _dedup_enum_types +from sqlpush.types import PlannedOperation, RiskClass # S2 (cycle-6 switchover): two tables sharing a native enum each embed a # verbatim CREATE TYPE in their add_table render — executing both dies with # DuplicateObject. Same defect family as F1/F2 (embedded-index dedup), but # statement-level and cross-table. +# NB: no module-level pg mark — the identity test below is DB-free and must +# never require the server; the DB tests carry the mark individually. _DIRTY_TABLES = ("enum_t1", "enum_t2") _ENUM_NAME = "shared_enum_type" @@ -43,6 +45,39 @@ def _create_type_count(sql_text: str) -> int: return sql_text.count(f"CREATE TYPE {_ENUM_NAME}") +def test_dedup_enum_types_returns_original_when_nothing_dropped(): + # S3 identity contract, unit-level (DB-free): an op whose type + # statement is NOT a duplicate passes through as the SAME object — + # only an op that actually lost a statement comes back rebuilt. + first = PlannedOperation( + type="add_table", + risk=RiskClass.SAFE, + sql=( + "CREATE TYPE public.shared_enum_type AS ENUM ('A', 'B');\n\n" + "CREATE TABLE public.enum_t1 (\n id SERIAL PRIMARY KEY\n)" + ), + table="enum_t1", + ) + dup = PlannedOperation( + type="add_table", + risk=RiskClass.SAFE, + sql=( + "CREATE TYPE public.shared_enum_type AS ENUM ('A', 'B');\n\n" + "CREATE TABLE public.enum_t2 (\n id SERIAL PRIMARY KEY\n)" + ), + table="enum_t2", + ) + assert _dedup_enum_types([first]) == [first] + assert _dedup_enum_types([first])[0] is first + + out = _dedup_enum_types([first, dup]) + assert out[0] is first # first occurrence keeps its embedded copy + assert out[1] is not dup # the duplicate is rebuilt... + assert "CREATE TYPE" not in out[1].sql # ...without the dropped statement... + assert "CREATE TABLE public.enum_t2" in out[1].sql # ...and with the rest + + +@pytest.mark.pg def test_plan_renders_shared_enum_create_type_once(enum_db): # Plan level: across ALL ops, the enum's CREATE TYPE appears exactly # once (first occurrence keeps it embedded in its add_table render). @@ -50,6 +85,7 @@ def test_plan_renders_shared_enum_create_type_once(enum_db): assert _create_type_count("".join(op.sql for op in p.operations)) == 1 +@pytest.mark.pg def test_migrate_shared_enum_succeeds_single_type(enum_db, tmp_path): """End-to-end S2 repro: revision → migrate on a fresh DB must succeed (was DuplicateObject) and leave exactly ONE pg_type row for the enum.""" @@ -71,3 +107,25 @@ def test_migrate_shared_enum_succeeds_single_type(enum_db, tmp_path): ).scalar() assert n == 1 # exactly one type, not one per table assert tables == 2 + + +@pytest.mark.pg +def test_push_shared_enum_applies_single_type(enum_db): + """S2 apply-path pin: push() must apply the deduped plan cleanly and + leave exactly ONE pg_type row — the dedup has to survive the apply + path (advisory-lock re-plan included), not just planning/replay.""" + rep = push(_shared_enum_md(), enum_db) + assert rep.applied and not rep.blocked and not rep.partial_failure, rep + with enum_db.connect() as conn: + n = conn.execute( + text("SELECT count(*) FROM pg_type WHERE typname = :n"), {"n": _ENUM_NAME} + ).scalar() + tables = conn.execute( + text( + "SELECT count(*) FROM pg_tables " + "WHERE schemaname='public' AND tablename IN ('enum_t1','enum_t2')" + ) + ).scalar() + assert n == 1 # deduped at apply time, not just in the planned SQL + assert tables == 2 + assert check(_shared_enum_md(), enum_db).clean diff --git a/tests/test_migrate.py b/tests/test_migrate.py index 6a700c7..567421d 100644 --- a/tests/test_migrate.py +++ b/tests/test_migrate.py @@ -4,7 +4,9 @@ from sqlalchemy import Column, Integer, MetaData, Table, text from sqlpush.api import check, migrate +from sqlpush.apply.executor import advisory_key from sqlpush.chain.format import MigrationFileError +from sqlpush.types import SqlpushError pytestmark = pytest.mark.pg @@ -121,3 +123,22 @@ def test_versions_table_pruned_from_public_check(migrate_db, tmp_path): Table("mt1", md, Column("id", Integer, primary_key=True)) # exactly the migrated state result = check(md, migrate_db) # PUBLIC scope — no schemas= assert result.clean, f"sqlpush_versions leaked into the diff: {result.drift}" + + +def test_migrate_advisory_wait_bounded(migrate_db, tmp_path): + # B3: the chain session's advisory lock wait is BOUNDED — a second + # connection holding the same key plus advisory_wait=0 must raise + # SqlpushError promptly instead of blocking on pg_advisory_lock + # forever (mirrors with_advisory_lock's wait-exhaustion contract). + _write(tmp_path, "0001_init.sql", SAFE_0001) + holder = migrate_db.connect() + key: int | None = None + try: + key = advisory_key(holder) + holder.execute(text("SELECT pg_advisory_lock(:k)"), {"k": key}) + with pytest.raises(SqlpushError, match="advisory lock"): + migrate(migrate_db, chain_dir=tmp_path, advisory_wait=0) + finally: + if key is not None: + holder.execute(text("SELECT pg_advisory_unlock(:k)"), {"k": key}) + holder.close() diff --git a/tests/test_stamp.py b/tests/test_stamp.py index 1fb1b60..fde93cc 100644 --- a/tests/test_stamp.py +++ b/tests/test_stamp.py @@ -5,6 +5,7 @@ from sqlpush.api import migrate, stamp from sqlpush.chain.format import checksum +from sqlpush.types import SqlpushError # helpers inlined: `from tests.test_migrate import ...` doesn't resolve — # tests/ is not a package and repo root is not on sys.path under pytest @@ -70,3 +71,49 @@ def test_stamp_fail_loud_on_bad_header(stamp_db, tmp_path): assert rep.blocked == ("0001_bad.sql",) assert rep.skipped == () # orden estricto: nada posterior se registra assert any(n.startswith("0001_bad.sql: ") for n in rep.notes) + + +def test_stamp_refuses_edited_file_without_force(stamp_db, tmp_path): + # B4: re-stamping after editing an applied/stamped file must NOT + # silently refresh the checksum — that wipes the edit-detection the + # whole chain integrity rides on. Refusal leaves the recorded + # checksum untouched; force=True accepts the new content. + _write(tmp_path, "0001_init.sql", SAFE_0001) + stamp(stamp_db, chain_dir=tmp_path) + edited = SAFE_0001.replace("mt1", "mt1_edited") + _write(tmp_path, "0001_init.sql", edited) + with pytest.raises(SqlpushError, match="0001_init.sql"): + stamp(stamp_db, chain_dir=tmp_path) + with stamp_db.connect() as conn: + sha = conn.execute( + text("SELECT sha256 FROM public.sqlpush_versions WHERE name = '0001_init.sql'") + ).scalar() + assert sha == checksum(SAFE_0001) # la negativa NO refrescó el registro + + rep = stamp(stamp_db, chain_dir=tmp_path, force=True) + assert rep.skipped == ("0001_init.sql",) + with stamp_db.connect() as conn: + sha2 = conn.execute( + text("SELECT sha256 FROM public.sqlpush_versions WHERE name = '0001_init.sql'") + ).scalar() + assert sha2 == checksum(edited) # force acepta el contenido nuevo + + +def test_stamp_mismatch_stops_the_walk(stamp_db, tmp_path): + # strict order on refusal too: the FIRST mismatch raises and nothing + # after it registers (same contract as migrate's blocked files) + _write(tmp_path, "0001_init.sql", SAFE_0001) + stamp(stamp_db, chain_dir=tmp_path) + _write(tmp_path, "0001_init.sql", SAFE_0001.replace("mt1", "mt1x")) # edit post-stamp + _write( + tmp_path, + "0002_later.sql", + "-- sqlpush: revision=0002 risk=SAFE\nCREATE TABLE mt2 (id int);\n", + ) + with pytest.raises(SqlpushError, match="0001_init.sql"): + stamp(stamp_db, chain_dir=tmp_path) + with stamp_db.connect() as conn: + later = conn.execute( + text("SELECT count(*) FROM public.sqlpush_versions WHERE name = '0002_later.sql'") + ).scalar() + assert later == 0 # nada posterior al mismatch se registra diff --git a/tests/test_validation.py b/tests/test_validation.py index c130dde..87116d8 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -76,6 +76,16 @@ def test_with_advisory_lock_rejects_negative_budgets(kwargs): ) +# --- B3/S2 (0.4.2): migrate budget validation (validation runs +# before any connection attempt, so DB-free like I4) ------------------------ + + +@pytest.mark.parametrize("kwargs", [{"advisory_wait": -1}, {"lock_timeout": -1}]) +def test_migrate_rejects_negative_budgets(kwargs, tmp_path): + with pytest.raises(SqlpushError, match=">= 0"): + api.migrate(_lazy_engine(), chain_dir=tmp_path, **kwargs) + + # --- I6: AlterColumnOp disambiguation (sentinel semantics per # docs/notes/alembic-notes.md Pattern C: False/None = do not touch) ---------