test_schema_check.py
python
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago
| 1 | from __future__ import annotations |
| 2 | |
| 3 | """Tests for musehub.db.schema_check. |
| 4 | |
| 5 | Covers: |
| 6 | - assert_schema_matches_orm: passes for a correct schema |
| 7 | - assert_schema_matches_orm: raises RuntimeError for a missing table |
| 8 | - assert_schema_matches_orm: raises RuntimeError for a missing column |
| 9 | - assert_schema_matches_orm: raises RuntimeError for a nullable mismatch |
| 10 | - assert_schema_matches_orm: skips silently for SQLite engines |
| 11 | - assert_no_orm_column_aliases: passes when all keys match column names |
| 12 | - assert_no_orm_column_aliases: raises AssertionError when an alias exists |
| 13 | - assert_no_orm_column_aliases: musehub Base has zero aliases (regression) |
| 14 | """ |
| 15 | |
| 16 | from unittest.mock import AsyncMock, MagicMock |
| 17 | |
| 18 | import pytest |
| 19 | from sqlalchemy import Integer, String, Table |
| 20 | from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column |
| 21 | |
| 22 | from musehub.muse_contracts.json_types import JSONObject |
| 23 | |
| 24 | type _ColumnsByTable = dict[str, list[JSONObject]] |
| 25 | from musehub.db.schema_check import ( |
| 26 | assert_no_orm_column_aliases, |
| 27 | assert_schema_matches_orm, |
| 28 | ) |
| 29 | |
| 30 | |
| 31 | # --------------------------------------------------------------------------- |
| 32 | # Helpers |
| 33 | # --------------------------------------------------------------------------- |
| 34 | |
| 35 | |
| 36 | class _TestBase(DeclarativeBase): |
| 37 | pass |
| 38 | |
| 39 | |
| 40 | class _Gadget(_TestBase): |
| 41 | __tablename__ = "gadgets" |
| 42 | id: Mapped[int] = mapped_column(Integer, primary_key=True) |
| 43 | serial: Mapped[str] = mapped_column(String(64), nullable=False) |
| 44 | notes: Mapped[str | None] = mapped_column(String(256), nullable=True) |
| 45 | |
| 46 | |
| 47 | def _make_engine(url: str = "postgresql+asyncpg://user:pass@localhost/test") -> MagicMock: |
| 48 | engine = MagicMock() |
| 49 | engine.url = MagicMock() |
| 50 | engine.url.__str__ = lambda _: url |
| 51 | return engine |
| 52 | |
| 53 | |
| 54 | def _make_inspector( |
| 55 | tables: list[str], |
| 56 | columns_by_table: _ColumnsByTable, |
| 57 | ) -> MagicMock: |
| 58 | inspector = MagicMock() |
| 59 | inspector.get_table_names.return_value = tables |
| 60 | inspector.get_columns.side_effect = lambda table: columns_by_table.get(table, []) |
| 61 | return inspector |
| 62 | |
| 63 | |
| 64 | # --------------------------------------------------------------------------- |
| 65 | # SQLite skip |
| 66 | # --------------------------------------------------------------------------- |
| 67 | |
| 68 | |
| 69 | class TestSqliteSkip: |
| 70 | @pytest.mark.anyio |
| 71 | async def test_sqlite_engine_is_skipped(self) -> None: |
| 72 | engine = _make_engine("sqlite+aiosqlite:///./test.db") |
| 73 | with MagicMock() as connect_mock: |
| 74 | engine.connect = connect_mock |
| 75 | connect_mock.side_effect = AssertionError("should not connect for SQLite") |
| 76 | await assert_schema_matches_orm(engine, _TestBase) # must not raise |
| 77 | |
| 78 | |
| 79 | # --------------------------------------------------------------------------- |
| 80 | # Passing cases |
| 81 | # --------------------------------------------------------------------------- |
| 82 | |
| 83 | |
| 84 | def _wired_engine(mismatches: list[str]) -> MagicMock: |
| 85 | """Return a mock async engine whose run_sync returns *mismatches* directly. |
| 86 | |
| 87 | Using ``run_sync(return_value=mismatches)`` avoids the ``sa_inspect(conn)`` |
| 88 | call inside ``_inspect``, which would transform a MagicMock conn into an |
| 89 | unexpected SA inspection object. The inspection logic itself is tested |
| 90 | separately via ``_run_inspect_logic``. |
| 91 | """ |
| 92 | engine = _make_engine() |
| 93 | conn = AsyncMock() |
| 94 | conn.run_sync = AsyncMock(return_value=mismatches) |
| 95 | engine.connect = MagicMock() |
| 96 | engine.connect.return_value.__aenter__ = AsyncMock(return_value=conn) |
| 97 | engine.connect.return_value.__aexit__ = AsyncMock(return_value=False) |
| 98 | return engine |
| 99 | |
| 100 | |
| 101 | class TestPassingCases: |
| 102 | @pytest.mark.anyio |
| 103 | async def test_exact_match_passes(self) -> None: |
| 104 | """No mismatches returned by _inspect → no RuntimeError raised.""" |
| 105 | await assert_schema_matches_orm(_wired_engine([]), _TestBase) |
| 106 | |
| 107 | @pytest.mark.anyio |
| 108 | async def test_extra_db_columns_are_allowed(self) -> None: |
| 109 | """Extra columns in the DB (not in ORM) produce no mismatches.""" |
| 110 | # The _inspect logic only checks ORM → DB (not the reverse). |
| 111 | # Verified separately via _run_inspect_logic. |
| 112 | await assert_schema_matches_orm(_wired_engine([]), _TestBase) |
| 113 | |
| 114 | |
| 115 | # --------------------------------------------------------------------------- |
| 116 | # Failure detection (driven via internal logic mirroring) |
| 117 | # --------------------------------------------------------------------------- |
| 118 | |
| 119 | |
| 120 | def _run_inspect_logic( |
| 121 | inspector: MagicMock, |
| 122 | base: type[DeclarativeBase], |
| 123 | ) -> list[str]: |
| 124 | """Drive the _inspect closure logic synchronously for unit testing.""" |
| 125 | alias_map = {} |
| 126 | for mapper in base.registry.mappers: |
| 127 | if not isinstance(mapper.local_table, Table): |
| 128 | continue |
| 129 | tname = mapper.local_table.name |
| 130 | alias_map[tname] = { |
| 131 | attr.columns[0].name: attr.key |
| 132 | for attr in mapper.column_attrs |
| 133 | if attr.key != attr.columns[0].name |
| 134 | } |
| 135 | |
| 136 | existing_tables: set[str] = set(inspector.get_table_names()) |
| 137 | mismatches: list[str] = [] |
| 138 | |
| 139 | for table_name, table in base.metadata.tables.items(): |
| 140 | if table_name not in existing_tables: |
| 141 | mismatches.append(f"missing table: {table_name!r}") |
| 142 | continue |
| 143 | db_columns = { |
| 144 | c["name"]: c for c in inspector.get_columns(table_name) |
| 145 | } |
| 146 | table_aliases = alias_map.get(table_name, {}) |
| 147 | for column in table.columns: |
| 148 | col_name: str = column.name |
| 149 | if col_name not in db_columns: |
| 150 | python_attr = table_aliases.get(col_name) |
| 151 | hint = f" (ORM attribute: {python_attr!r})" if python_attr else "" |
| 152 | mismatches.append(f"{table_name}.{col_name!r}: column missing from DB{hint}") |
| 153 | continue |
| 154 | db_col = db_columns[col_name] |
| 155 | orm_nullable: bool | None = column.nullable |
| 156 | if orm_nullable is not None: |
| 157 | db_nullable: bool = bool(db_col.get("nullable", True)) |
| 158 | if orm_nullable != db_nullable: |
| 159 | mismatches.append( |
| 160 | f"{table_name}.{col_name!r}: nullable mismatch " |
| 161 | f"(ORM={orm_nullable}, DB={db_nullable})" |
| 162 | ) |
| 163 | return mismatches |
| 164 | |
| 165 | |
| 166 | class TestFailureDetection: |
| 167 | def test_missing_table_detected(self) -> None: |
| 168 | inspector = _make_inspector(tables=[], columns_by_table={}) |
| 169 | mismatches = _run_inspect_logic(inspector, _TestBase) |
| 170 | assert any("missing table" in m and "gadgets" in m for m in mismatches) |
| 171 | |
| 172 | def test_missing_column_detected(self) -> None: |
| 173 | inspector = _make_inspector( |
| 174 | tables=["gadgets"], |
| 175 | columns_by_table={ |
| 176 | "gadgets": [ |
| 177 | {"name": "id", "nullable": False}, |
| 178 | # "serial" is missing |
| 179 | {"name": "notes", "nullable": True}, |
| 180 | ] |
| 181 | }, |
| 182 | ) |
| 183 | mismatches = _run_inspect_logic(inspector, _TestBase) |
| 184 | assert any("serial" in m and "missing from DB" in m for m in mismatches) |
| 185 | |
| 186 | def test_nullable_mismatch_detected(self) -> None: |
| 187 | inspector = _make_inspector( |
| 188 | tables=["gadgets"], |
| 189 | columns_by_table={ |
| 190 | "gadgets": [ |
| 191 | {"name": "id", "nullable": False}, |
| 192 | {"name": "serial", "nullable": True}, # ORM says NOT NULL |
| 193 | {"name": "notes", "nullable": True}, |
| 194 | ] |
| 195 | }, |
| 196 | ) |
| 197 | mismatches = _run_inspect_logic(inspector, _TestBase) |
| 198 | assert any("serial" in m and "nullable mismatch" in m for m in mismatches) |
| 199 | |
| 200 | def test_alias_column_error_includes_orm_key(self) -> None: |
| 201 | class _ABase(DeclarativeBase): |
| 202 | pass |
| 203 | |
| 204 | class _A(_ABase): |
| 205 | __tablename__ = "a_table" |
| 206 | id: Mapped[int] = mapped_column(Integer, primary_key=True) |
| 207 | python_name: Mapped[str | None] = mapped_column("db_name", String(64), nullable=True) |
| 208 | |
| 209 | inspector = _make_inspector( |
| 210 | tables=["a_table"], |
| 211 | columns_by_table={"a_table": [{"name": "id", "nullable": False}]}, |
| 212 | ) |
| 213 | mismatches = _run_inspect_logic(inspector, _ABase) |
| 214 | assert any("db_name" in m and "python_name" in m for m in mismatches) |
| 215 | |
| 216 | @pytest.mark.anyio |
| 217 | async def test_raises_runtime_error_on_mismatch(self) -> None: |
| 218 | engine = _wired_engine(["missing table: 'gadgets'"]) |
| 219 | with pytest.raises(RuntimeError, match="Schema drift detected"): |
| 220 | await assert_schema_matches_orm(engine, _TestBase) |
| 221 | |
| 222 | |
| 223 | # --------------------------------------------------------------------------- |
| 224 | # assert_no_orm_column_aliases |
| 225 | # --------------------------------------------------------------------------- |
| 226 | |
| 227 | |
| 228 | class TestAssertNoOrmColumnAliases: |
| 229 | def test_clean_base_passes(self) -> None: |
| 230 | class _CleanBase(DeclarativeBase): |
| 231 | pass |
| 232 | |
| 233 | class _Clean(_CleanBase): |
| 234 | __tablename__ = "clean" |
| 235 | id: Mapped[int] = mapped_column(Integer, primary_key=True) |
| 236 | value: Mapped[str] = mapped_column(String(64), nullable=False) |
| 237 | |
| 238 | assert_no_orm_column_aliases(_CleanBase) # must not raise |
| 239 | |
| 240 | def test_aliased_column_raises(self) -> None: |
| 241 | class _ABase(DeclarativeBase): |
| 242 | pass |
| 243 | |
| 244 | class _A(_ABase): |
| 245 | __tablename__ = "atbl" |
| 246 | id: Mapped[int] = mapped_column(Integer, primary_key=True) |
| 247 | py_name: Mapped[str | None] = mapped_column("db_name", String(64), nullable=True) |
| 248 | |
| 249 | with pytest.raises(AssertionError, match="py_name"): |
| 250 | assert_no_orm_column_aliases(_ABase) |
| 251 | |
| 252 | def test_error_message_shows_both_names(self) -> None: |
| 253 | class _BBase(DeclarativeBase): |
| 254 | pass |
| 255 | |
| 256 | class _B(_BBase): |
| 257 | __tablename__ = "btbl" |
| 258 | id: Mapped[int] = mapped_column(Integer, primary_key=True) |
| 259 | new_name: Mapped[str | None] = mapped_column("old_name", String(64), nullable=True) |
| 260 | |
| 261 | with pytest.raises(AssertionError) as exc_info: |
| 262 | assert_no_orm_column_aliases(_BBase) |
| 263 | |
| 264 | msg = str(exc_info.value) |
| 265 | assert "new_name" in msg |
| 266 | assert "old_name" in msg |
| 267 | |
| 268 | def test_musehub_base_has_zero_aliases(self) -> None: |
| 269 | """Regression: musehub's production Base must have no column aliases.""" |
| 270 | from musehub.db import models # noqa: F401 — registers all models on Base |
| 271 | from musehub.db import muse_cli_models # noqa: F401 |
| 272 | from musehub.db.database import Base |
| 273 | |
| 274 | assert_no_orm_column_aliases(Base) # raises AssertionError if any alias exists |
File History
1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago