75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
"""Idempotency contract for the Issue #26 strategy-catalog migration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
|
|
_MIGRATION_PATH = (
|
|
Path(__file__).parents[1]
|
|
/ "app/db/alembic/versions/20260814_0029_strategy_catalog_v1.py"
|
|
)
|
|
|
|
|
|
def _load_migration():
|
|
spec = importlib.util.spec_from_file_location("strategy_catalog_0029", _MIGRATION_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
|
|
|
|
|
|
class _Bind:
|
|
def __init__(self) -> None:
|
|
self.executed: list[str] = []
|
|
self.params: list[dict | None] = []
|
|
|
|
def execute(self, statement, params=None):
|
|
self.executed.append(str(statement))
|
|
self.params.append(params)
|
|
return SimpleNamespace()
|
|
|
|
|
|
class _Inspector:
|
|
def __init__(self, columns: set[str]) -> None:
|
|
self.columns = columns
|
|
|
|
def get_table_names(self) -> list[str]:
|
|
return ["strategy_sets"]
|
|
|
|
def get_columns(self, table_name: str) -> list[dict[str, str]]:
|
|
assert table_name == "strategy_sets"
|
|
return [{"name": name} for name in sorted(self.columns)]
|
|
|
|
|
|
def test_upgrade_is_repeatable_and_uses_data_and_constraint_guards() -> None:
|
|
migration = _load_migration()
|
|
columns = {"id", "name", "config", "nick", "created_at"}
|
|
inspector = _Inspector(columns)
|
|
bind = _Bind()
|
|
added: list[str] = []
|
|
|
|
def add_column(table_name: str, column) -> None:
|
|
assert table_name == "strategy_sets"
|
|
columns.add(column.name)
|
|
added.append(column.name)
|
|
|
|
with (
|
|
patch.object(migration.op, "get_bind", return_value=bind),
|
|
patch.object(migration.op, "add_column", side_effect=add_column),
|
|
patch.object(migration, "inspect", return_value=inspector),
|
|
):
|
|
migration.upgrade()
|
|
migration.upgrade()
|
|
|
|
assert added == ["status", "revision", "updated_at"]
|
|
sql = "\n".join(bind.executed)
|
|
assert "WHERE NOT EXISTS" in sql
|
|
assert "IF NOT EXISTS" in sql
|
|
assert "ck_strategy_sets_status" in sql
|
|
assert "ck_strategy_sets_revision_positive" in sql
|
|
assert any(params and params.get("nick") == "target_value:bf21a9e8fbc5" for params in bind.params)
|