feat: BAT-025 add portfolio ledger manual metadata migration

tests/test_bat025_portfolio_ledger_manual_metadata_migration.py

@@ -0,0 +1,133 @@
+"""BAT-025: portfolio_ledger manual metadata migration behavior tests.
+
+Focus areas:
+- fresh schema: add required columns safely
+- runtime-precreated schema: source backfill + source constraints 보정
+- forward-only downgrade policy
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import importlib.util
+
+
+_MIGRATION_PATH = Path(__file__).resolve().parents[1] / "migrations" / "versions" / "20260715_0028_portfolio_ledger_manual_metadata.py"
+_MIGRATION_NAME = "m_20260715_0028_portfolio_ledger_manual_metadata"
+_spec = importlib.util.spec_from_file_location(_MIGRATION_NAME, _MIGRATION_PATH)
+assert _spec is not None and _spec.loader is not None
+_migration = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_migration)
+
+
+class _FakeInspector:
+    def __init__(self, columns: list[str]) -> None:
+        self._columns = columns
+
+    def get_table_names(self) -> list[str]:
+        return ["portfolio_ledger"]
+
+    def get_columns(self, table_name: str) -> list[dict[str, Any]]:
+        assert table_name == "portfolio_ledger"
+        return [{"name": name} for name in self._columns]
+
+
+class _MissingLedgerInspector(_FakeInspector):
+    def get_table_names(self) -> list[str]:
+        return ["other_table"]
+
+
+class _FakeResult:
+    def __init__(self, scalar_value: Any = None) -> None:
+        self._scalar_value = scalar_value
+
+    def scalar(self) -> Any:
+        return self._scalar_value
+
+
+class _FakeBind:
+    def __init__(self) -> None:
+        self.queries: list[str] = []
+
+    def execute(self, statement: object, *args: object, **kwargs: object) -> _FakeResult:
+        self.queries.append(str(statement).strip())
+        return _FakeResult(0)
+
+
+def _sql_contains(queries: list[str], needle: str) -> bool:
+    return any(needle in q for q in queries)
+
+
+def test_upgrade_adds_all_columns_on_fresh_portfolio_ledger() -> None:
+    fake_bind = _FakeBind()
+    fake_inspector = _FakeInspector(columns=[])
+
+    with (
+        patch.object(_migration.op, "get_bind", return_value=fake_bind),
+        patch.object(_migration, "inspect", return_value=fake_inspector),
+        patch.object(_migration.op, "add_column") as add_column,
+        patch.object(_migration.op, "alter_column") as alter_column,
+    ):
+        _migration.upgrade()
+
+        assert add_column.call_count == 4
+        called = [call.args[1].name for call in add_column.call_args_list]
+        assert called == [
+            "source",
+            "target_ledger_entry_id",
+            "confirmation_evidence",
+            "actor_user_id",
+        ]
+        alter_column.assert_not_called()
+
+
+def test_upgrade_preserves_data_and_enforces_source_when_precreated() -> None:
+    fake_bind = _FakeBind()
+    fake_inspector = _FakeInspector(columns=[
+        "source",
+        "target_ledger_entry_id",
+        "confirmation_evidence",
+        "actor_user_id",
+    ])
+
+    with (
+        patch.object(_migration.op, "get_bind", return_value=fake_bind),
+        patch.object(_migration, "inspect", return_value=fake_inspector),
+        patch.object(_migration.op, "add_column") as add_column,
+        patch.object(_migration.op, "alter_column") as alter_column,
+    ):
+        _migration.upgrade()
+
+        add_column.assert_not_called()
+        alter_column.assert_called_once()
+        alter_kwargs = alter_column.call_args.kwargs
+        assert str(alter_kwargs["existing_type"]).upper().startswith("TEXT")
+        assert alter_kwargs["server_default"].text == "'broker'"
+        assert alter_kwargs["nullable"] is False
+        assert _sql_contains(fake_bind.queries, "UPDATE portfolio_ledger SET source = 'broker' WHERE source IS NULL")
+
+
+def test_upgrade_noop_when_portfolio_ledger_missing() -> None:
+    fake_bind = _FakeBind()
+    fake_inspector = _MissingLedgerInspector(columns=[])
+
+    with (
+        patch.object(_migration.op, "get_bind", return_value=fake_bind),
+        patch.object(_migration, "inspect", return_value=fake_inspector),
+        patch.object(_migration.op, "add_column") as add_column,
+        patch.object(_migration.op, "alter_column") as alter_column,
+    ):
+        _migration.upgrade()
+
+        add_column.assert_not_called()
+        alter_column.assert_not_called()
+        assert fake_bind.queries == []
+
+
+def test_downgrade_is_forward_only_noop() -> None:
+    with patch.object(_migration.op, "get_bind", return_value=MagicMock()):
+        result = _migration.downgrade()
+    assert result is None