Skip to content

Commit 0766383

Browse files
Add Time-column and reflection integration tests
ISSUE-20 — Column(Time) round-trips through the dialect. Confirms the current decision to *not* override Time in colspecs: the C server tags only DATETIME/DATE/TIMESTAMP as DQLITE_ISO8601, so TIME columns come back as plain TEXT and the inherited SQLAlchemy Time processor parses them correctly. ISSUE-22 — MetaData.reflect and inspect() round-trip over the wire. Covers columns / primary keys / indexes / foreign keys — the common PRAGMA paths the SQLiteDialect inherits. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 545e439 commit 0766383

1 file changed

Lines changed: 126 additions & 0 deletions

File tree

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
"""Coverage for Time columns (ISSUE-20) and metadata reflection (ISSUE-22).
2+
3+
- ISSUE-20: ``Column(Time)`` is not overridden in the dialect's
4+
``colspecs`` because the C server tags only DATETIME/DATE/TIMESTAMP
5+
as ``DQLITE_ISO8601``; TIME columns come back as plain TEXT and the
6+
inherited SQLAlchemy ``Time`` processor parses the string. Verified
7+
here end-to-end.
8+
9+
- ISSUE-22: the dialect inherits SQLite's reflection methods, which
10+
lean on PRAGMA queries. dqlite wraps SQLite but we had zero proof
11+
that reflection round-trips over the wire. These tests exercise the
12+
common paths.
13+
"""
14+
15+
import datetime
16+
from collections.abc import Generator
17+
18+
import pytest
19+
from sqlalchemy import (
20+
Column,
21+
Date,
22+
DateTime,
23+
ForeignKey,
24+
Index,
25+
Integer,
26+
MetaData,
27+
String,
28+
Table,
29+
Time,
30+
create_engine,
31+
inspect,
32+
)
33+
from sqlalchemy.engine import Engine
34+
from sqlalchemy.orm import Session, declarative_base
35+
36+
Base = declarative_base()
37+
38+
39+
class TimeTestModel(Base): # type: ignore[valid-type,misc]
40+
__tablename__ = "time_test"
41+
id = Column(Integer, primary_key=True)
42+
t = Column(Time)
43+
44+
45+
@pytest.mark.integration
46+
class TestTimeColumn:
47+
@pytest.fixture
48+
def engine(self, engine_url: str) -> Generator[Engine]:
49+
engine = create_engine(engine_url)
50+
Base.metadata.create_all(engine)
51+
yield engine
52+
Base.metadata.drop_all(engine)
53+
engine.dispose()
54+
55+
def test_time_roundtrip(self, engine: Engine) -> None:
56+
value = datetime.time(10, 30, 45)
57+
with Session(engine) as s:
58+
s.add(TimeTestModel(t=value))
59+
s.commit()
60+
result = s.query(TimeTestModel).order_by(TimeTestModel.id.desc()).first()
61+
assert result is not None
62+
assert isinstance(result.t, datetime.time)
63+
assert result.t == value
64+
65+
66+
@pytest.mark.integration
67+
class TestReflection:
68+
"""Metadata reflection via PRAGMA round-trips through the wire."""
69+
70+
def test_reflect_round_trip(self, engine_url: str) -> None:
71+
engine = create_engine(engine_url)
72+
try:
73+
# Create a schema with a handful of features.
74+
src = MetaData()
75+
Table(
76+
"authors",
77+
src,
78+
Column("id", Integer, primary_key=True),
79+
Column("name", String(64), nullable=False),
80+
)
81+
Table(
82+
"books",
83+
src,
84+
Column("id", Integer, primary_key=True),
85+
Column("title", String(255), nullable=False),
86+
Column("author_id", Integer, ForeignKey("authors.id")),
87+
Column("published", Date),
88+
Column("updated_at", DateTime),
89+
Index("ix_books_title", "title"),
90+
)
91+
src.create_all(engine)
92+
try:
93+
# Reflect into a fresh MetaData and check the shape.
94+
dst = MetaData()
95+
dst.reflect(bind=engine)
96+
assert {"authors", "books"}.issubset(dst.tables.keys())
97+
98+
authors = dst.tables["authors"]
99+
assert "id" in authors.columns
100+
assert "name" in authors.columns
101+
assert list(authors.primary_key.columns.keys()) == ["id"]
102+
103+
books = dst.tables["books"]
104+
assert {"id", "title", "author_id", "published", "updated_at"}.issubset(
105+
books.columns.keys()
106+
)
107+
108+
# Inspector-level checks.
109+
insp = inspect(engine)
110+
cols = {c["name"] for c in insp.get_columns("books")}
111+
assert cols == {"id", "title", "author_id", "published", "updated_at"}
112+
113+
indexes = insp.get_indexes("books")
114+
index_names = {ix["name"] for ix in indexes}
115+
assert "ix_books_title" in index_names
116+
117+
fks = insp.get_foreign_keys("books")
118+
assert any(
119+
fk["referred_table"] == "authors"
120+
and fk["constrained_columns"] == ["author_id"]
121+
for fk in fks
122+
)
123+
finally:
124+
src.drop_all(engine)
125+
finally:
126+
engine.dispose()

0 commit comments

Comments
 (0)