|
| 1 | +"""Lock in reflected column type-class identity (ISSUE-45). |
| 2 | +
|
| 3 | +ischema_names inherited from pysqlite covers the tokens dqlite emits |
| 4 | +at DDL time (INTEGER, TEXT, REAL, BLOB, NUMERIC, DATE, DATETIME, ...). |
| 5 | +If a future pysqlite change alters the mapping, a reflected column |
| 6 | +might silently land in NULLTYPE. This test is the canary. |
| 7 | +""" |
| 8 | + |
| 9 | +from collections.abc import Generator |
| 10 | + |
| 11 | +import pytest |
| 12 | +from sqlalchemy import ( |
| 13 | + BLOB, |
| 14 | + Boolean, |
| 15 | + Column, |
| 16 | + Date, |
| 17 | + DateTime, |
| 18 | + Float, |
| 19 | + Integer, |
| 20 | + MetaData, |
| 21 | + Numeric, |
| 22 | + String, |
| 23 | + Table, |
| 24 | + Text, |
| 25 | + create_engine, |
| 26 | + inspect, |
| 27 | + types, |
| 28 | +) |
| 29 | +from sqlalchemy.engine import Engine |
| 30 | + |
| 31 | +BOOLEAN_LIKE = (types.Boolean, types.Numeric, types.SmallInteger, types.Integer) |
| 32 | + |
| 33 | + |
| 34 | +@pytest.mark.integration |
| 35 | +class TestReflectColumnTypes: |
| 36 | + @pytest.fixture |
| 37 | + def engine(self, engine_url: str) -> Generator[Engine]: |
| 38 | + engine = create_engine(engine_url) |
| 39 | + md = MetaData() |
| 40 | + Table( |
| 41 | + "reflect_types_test", |
| 42 | + md, |
| 43 | + Column("id", Integer, primary_key=True), |
| 44 | + Column("name", String(32)), |
| 45 | + Column("body", Text), |
| 46 | + Column("blob_data", BLOB), |
| 47 | + Column("price", Numeric(10, 2)), |
| 48 | + Column("ratio", Float), |
| 49 | + Column("created", DateTime), |
| 50 | + Column("birthday", Date), |
| 51 | + Column("flag", Boolean), |
| 52 | + ) |
| 53 | + md.create_all(engine) |
| 54 | + try: |
| 55 | + yield engine |
| 56 | + finally: |
| 57 | + md.drop_all(engine) |
| 58 | + engine.dispose() |
| 59 | + |
| 60 | + def test_reflects_standard_types(self, engine: Engine) -> None: |
| 61 | + insp = inspect(engine) |
| 62 | + cols = {c["name"]: c for c in insp.get_columns("reflect_types_test")} |
| 63 | + |
| 64 | + assert isinstance(cols["id"]["type"], types.Integer) |
| 65 | + assert isinstance(cols["name"]["type"], types.String) |
| 66 | + assert isinstance(cols["body"]["type"], types.Text | types.String) |
| 67 | + assert isinstance(cols["blob_data"]["type"], types.LargeBinary | types.BLOB) |
| 68 | + assert isinstance(cols["price"]["type"], types.Numeric | types.Float | types.Integer) |
| 69 | + assert isinstance(cols["ratio"]["type"], types.Float | types.Numeric) |
| 70 | + assert isinstance(cols["created"]["type"], types.DateTime) |
| 71 | + assert isinstance(cols["birthday"]["type"], types.Date) |
| 72 | + assert isinstance(cols["flag"]["type"], BOOLEAN_LIKE) |
| 73 | + |
| 74 | + # No column landed in NullType. |
| 75 | + for name, c in cols.items(): |
| 76 | + assert not isinstance(c["type"], types.NullType), f"column {name} reflected to NullType" |
0 commit comments