guke
47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
"""updated_at 列存在、回填 = created_at、onupdate 在 ORM 更新时刷新。"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from sqlalchemy import inspect
|
|
|
|
from app.db.session import SessionLocal, engine
|
|
from app.models.comparison import ComparisonRecord
|
|
|
|
|
|
def test_updated_at_column_and_index_exist() -> None:
|
|
insp = inspect(engine)
|
|
cols = {c["name"] for c in insp.get_columns("comparison_record")}
|
|
assert "updated_at" in cols
|
|
idx_names = {i["name"] for i in insp.get_indexes("comparison_record")}
|
|
# 迁移环境手动建索引名为 ix_comparison_updated;
|
|
# 测试环境 create_all() 按 SQLAlchemy 命名约定生成 ix_comparison_record_updated_at。
|
|
# 两种环境都验通过即可。
|
|
assert (
|
|
"ix_comparison_updated" in idx_names
|
|
or "ix_comparison_record_updated_at" in idx_names
|
|
), f"updated_at index not found; available indexes: {idx_names}"
|
|
|
|
|
|
def test_onupdate_refreshes_updated_at() -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
rec = ComparisonRecord(trace_id="alert-updated-at-onupdate", status="running")
|
|
db.add(rec)
|
|
db.commit()
|
|
db.refresh(rec)
|
|
first = rec.updated_at
|
|
assert first is not None
|
|
time.sleep(1.1) # SQLite CURRENT_TIMESTAMP 秒级精度,睡过 1 秒才看得出变化
|
|
rec.status = "failed"
|
|
db.commit()
|
|
db.refresh(rec)
|
|
assert rec.updated_at > first
|
|
finally:
|
|
db.rollback()
|
|
db.query(ComparisonRecord).filter(
|
|
ComparisonRecord.trace_id == "alert-updated-at-onupdate"
|
|
).delete()
|
|
db.commit()
|
|
db.close()
|