gabriel / musehub public
test_schema_check.py python
270 lines 10.1 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 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.types.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 async def test_sqlite_engine_is_skipped(self) -> None:
71 engine = _make_engine("sqlite+aiosqlite:///./test.db")
72 with MagicMock() as connect_mock:
73 engine.connect = connect_mock
74 connect_mock.side_effect = AssertionError("should not connect for SQLite")
75 await assert_schema_matches_orm(engine, _TestBase) # must not raise
76
77
78 # ---------------------------------------------------------------------------
79 # Passing cases
80 # ---------------------------------------------------------------------------
81
82
83 def _wired_engine(mismatches: list[str]) -> MagicMock:
84 """Return a mock async engine whose run_sync returns *mismatches* directly.
85
86 Using ``run_sync(return_value=mismatches)`` avoids the ``sa_inspect(conn)``
87 call inside ``_inspect``, which would transform a MagicMock conn into an
88 unexpected SA inspection object. The inspection logic itself is tested
89 separately via ``_run_inspect_logic``.
90 """
91 engine = _make_engine()
92 conn = AsyncMock()
93 conn.run_sync = AsyncMock(return_value=mismatches)
94 engine.connect = MagicMock()
95 engine.connect.return_value.__aenter__ = AsyncMock(return_value=conn)
96 engine.connect.return_value.__aexit__ = AsyncMock(return_value=False)
97 return engine
98
99
100 class TestPassingCases:
101 async def test_exact_match_passes(self) -> None:
102 """No mismatches returned by _inspect → no RuntimeError raised."""
103 await assert_schema_matches_orm(_wired_engine([]), _TestBase)
104
105 async def test_extra_db_columns_are_allowed(self) -> None:
106 """Extra columns in the DB (not in ORM) produce no mismatches."""
107 # The _inspect logic only checks ORM → DB (not the reverse).
108 # Verified separately via _run_inspect_logic.
109 await assert_schema_matches_orm(_wired_engine([]), _TestBase)
110
111
112 # ---------------------------------------------------------------------------
113 # Failure detection (driven via internal logic mirroring)
114 # ---------------------------------------------------------------------------
115
116
117 def _run_inspect_logic(
118 inspector: MagicMock,
119 base: type[DeclarativeBase],
120 ) -> list[str]:
121 """Drive the _inspect closure logic synchronously for unit testing."""
122 alias_map = {}
123 for mapper in base.registry.mappers:
124 if not isinstance(mapper.local_table, Table):
125 continue
126 tname = mapper.local_table.name
127 alias_map[tname] = {
128 attr.columns[0].name: attr.key
129 for attr in mapper.column_attrs
130 if attr.key != attr.columns[0].name
131 }
132
133 existing_tables: set[str] = set(inspector.get_table_names())
134 mismatches: list[str] = []
135
136 for table_name, table in base.metadata.tables.items():
137 if table_name not in existing_tables:
138 mismatches.append(f"missing table: {table_name!r}")
139 continue
140 db_columns = {
141 c["name"]: c for c in inspector.get_columns(table_name)
142 }
143 table_aliases = alias_map.get(table_name, {})
144 for column in table.columns:
145 col_name: str = column.name
146 if col_name not in db_columns:
147 python_attr = table_aliases.get(col_name)
148 hint = f" (ORM attribute: {python_attr!r})" if python_attr else ""
149 mismatches.append(f"{table_name}.{col_name!r}: column missing from DB{hint}")
150 continue
151 db_col = db_columns[col_name]
152 orm_nullable: bool | None = column.nullable
153 if orm_nullable is not None:
154 db_nullable: bool = bool(db_col.get("nullable", True))
155 if orm_nullable != db_nullable:
156 mismatches.append(
157 f"{table_name}.{col_name!r}: nullable mismatch "
158 f"(ORM={orm_nullable}, DB={db_nullable})"
159 )
160 return mismatches
161
162
163 class TestFailureDetection:
164 def test_missing_table_detected(self) -> None:
165 inspector = _make_inspector(tables=[], columns_by_table={})
166 mismatches = _run_inspect_logic(inspector, _TestBase)
167 assert any("missing table" in m and "gadgets" in m for m in mismatches)
168
169 def test_missing_column_detected(self) -> None:
170 inspector = _make_inspector(
171 tables=["gadgets"],
172 columns_by_table={
173 "gadgets": [
174 {"name": "id", "nullable": False},
175 # "serial" is missing
176 {"name": "notes", "nullable": True},
177 ]
178 },
179 )
180 mismatches = _run_inspect_logic(inspector, _TestBase)
181 assert any("serial" in m and "missing from DB" in m for m in mismatches)
182
183 def test_nullable_mismatch_detected(self) -> None:
184 inspector = _make_inspector(
185 tables=["gadgets"],
186 columns_by_table={
187 "gadgets": [
188 {"name": "id", "nullable": False},
189 {"name": "serial", "nullable": True}, # ORM says NOT NULL
190 {"name": "notes", "nullable": True},
191 ]
192 },
193 )
194 mismatches = _run_inspect_logic(inspector, _TestBase)
195 assert any("serial" in m and "nullable mismatch" in m for m in mismatches)
196
197 def test_alias_column_error_includes_orm_key(self) -> None:
198 class _ABase(DeclarativeBase):
199 pass
200
201 class _A(_ABase):
202 __tablename__ = "a_table"
203 id: Mapped[int] = mapped_column(Integer, primary_key=True)
204 python_name: Mapped[str | None] = mapped_column("db_name", String(64), nullable=True)
205
206 inspector = _make_inspector(
207 tables=["a_table"],
208 columns_by_table={"a_table": [{"name": "id", "nullable": False}]},
209 )
210 mismatches = _run_inspect_logic(inspector, _ABase)
211 assert any("db_name" in m and "python_name" in m for m in mismatches)
212
213 async def test_raises_runtime_error_on_mismatch(self) -> None:
214 engine = _wired_engine(["missing table: 'gadgets'"])
215 with pytest.raises(RuntimeError, match="Schema drift detected"):
216 await assert_schema_matches_orm(engine, _TestBase)
217
218
219 # ---------------------------------------------------------------------------
220 # assert_no_orm_column_aliases
221 # ---------------------------------------------------------------------------
222
223
224 class TestAssertNoOrmColumnAliases:
225 def test_clean_base_passes(self) -> None:
226 class _CleanBase(DeclarativeBase):
227 pass
228
229 class _Clean(_CleanBase):
230 __tablename__ = "clean"
231 id: Mapped[int] = mapped_column(Integer, primary_key=True)
232 value: Mapped[str] = mapped_column(String(64), nullable=False)
233
234 assert_no_orm_column_aliases(_CleanBase) # must not raise
235
236 def test_aliased_column_raises(self) -> None:
237 class _ABase(DeclarativeBase):
238 pass
239
240 class _A(_ABase):
241 __tablename__ = "atbl"
242 id: Mapped[int] = mapped_column(Integer, primary_key=True)
243 py_name: Mapped[str | None] = mapped_column("db_name", String(64), nullable=True)
244
245 with pytest.raises(AssertionError, match="py_name"):
246 assert_no_orm_column_aliases(_ABase)
247
248 def test_error_message_shows_both_names(self) -> None:
249 class _BBase(DeclarativeBase):
250 pass
251
252 class _B(_BBase):
253 __tablename__ = "btbl"
254 id: Mapped[int] = mapped_column(Integer, primary_key=True)
255 new_name: Mapped[str | None] = mapped_column("old_name", String(64), nullable=True)
256
257 with pytest.raises(AssertionError) as exc_info:
258 assert_no_orm_column_aliases(_BBase)
259
260 msg = str(exc_info.value)
261 assert "new_name" in msg
262 assert "old_name" in msg
263
264 def test_musehub_base_has_zero_aliases(self) -> None:
265 """Regression: musehub's production Base must have no column aliases."""
266 from musehub.db import models # noqa: F401 — registers all models on Base
267 from musehub.db import muse_cli_models # noqa: F401
268 from musehub.db.database import Base
269
270 assert_no_orm_column_aliases(Base) # raises AssertionError if any alias exists
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago