浅念
43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
"""5 个新鲜度索引:create_all 建出 + 模型迁移双声明一致(防 autogenerate 反 DROP)。
|
|
|
|
迁移侧(c3d4e5f6a7b8)由临时库 alembic upgrade head 验证;本文件锁定
|
|
模型侧——Base.metadata 与 models.py __table_args__ 声明一旦被误删,
|
|
create_all 建出的表就缺索引,此测试即红。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.db.base import Base
|
|
|
|
EXPECTED = {
|
|
"customer": "ix_customer_updated_at",
|
|
"chat_group": "ix_chat_group_updated_at",
|
|
"phone_call": "ix_phone_call_updated_at",
|
|
"media": "ix_media_updated_at",
|
|
"moments_post": "ix_moments_post_updated_at",
|
|
}
|
|
|
|
|
|
async def test_freshness_indexes_exist():
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
try:
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
names = await conn.run_sync(
|
|
lambda sync_conn: {
|
|
t: {ix["name"] for ix in sa.inspect(sync_conn).get_indexes(t)}
|
|
for t in EXPECTED
|
|
}
|
|
)
|
|
for table, index in EXPECTED.items():
|
|
assert index in names[table], f"{table}.{index} missing"
|
|
finally:
|
|
await engine.dispose()
|