yml2213
65a4590291
问题不是「UTC 还是本地时间」，而是这套口径有两个半吊子的地方： 1. 库里存 naive UTC、读出来也是裸时间，于是每一处比较/输出都得自己补时区。 项目里因此散了三份 _as_utc 和一堆 replace(tzinfo=UTC)，P1a 排下一轮时间 时还炸过一次 TypeError: can't compare offset-naive and offset-aware。 2. MySQL 会话时区跟随系统 +08:00，而库里存的是 UTC —— SQL 里的 NOW()/ CURDATE() 比库里的时间快 8 小时，查询会静默返回空（P1a 排查时就遇到过， 看着像「没有数据」）。 改动： - 新增 database.UTCDateTime：写时 aware→转 UTC 去 tzinfo、naive 视为 UTC； 读时一律补 UTC 时区。底层 DDL 仍是 DATETIME，**不需要迁移**。models.py 的 28 个时间列全部改用它，模型上的时间字段从此不再是裸值； - docker-compose 给 mysql 加 --default-time-zone=+00:00，让 SQL 时间函数与存储 口径一致。安全性已核实：全库 0 个 TIMESTAMP 列、0 个 CURRENT_TIMESTAMP 默认值，改这里不会改变任何已有数据的语义； - 新增 tests/test_utc_datetime.py：读出必须 aware、能直接和 datetime.now(UTC) 比较、带 +08:00 的写入要落成 UTC、naive 写入视为 UTC，并有一条元测试扫描 Base.metadata，防止以后新增时间列忘了用 UTCDateTime。 保留既有的 _as_utc 补丁（对 aware 值是幂等的，属防御性代码）；新代码不再需要。 「存 UTC、算 UTC、显示本地」三条线至此各自明确：存储与比较统一 UTC，日志按 容器 TZ 显示本地时间，前端 formatTime 负责转本地。
95 lines
4.1 KiB
Python
95 lines
4.1 KiB
Python
"""回归：时间列统一走 ``UTCDateTime`` —— 读出来必须带时区，写进去必须存 UTC。
|
||
|
||
背景：库里存的是 naive UTC，读出来也是裸时间，于是每一处比较/输出都要自己记得
|
||
补时区。项目里因此散了三份 ``_as_utc`` 和一堆 ``replace(tzinfo=UTC)``，还踩过一次
|
||
|
||
TypeError: can't compare offset-naive and offset-aware datetimes
|
||
|
||
（P1a 的 CK 巡检排下一轮时间时炸的）。现在把边界收在 ``UTCDateTime``：
|
||
写时 aware→UTC 裸值、naive 视为 UTC；读时一律补 UTC 时区。
|
||
|
||
这组测试盯三件事：读出来是 aware、能和 ``datetime.now(UTC)`` 直接比、写进去
|
||
落的是 UTC 而不是本地时间。
|
||
"""
|
||
|
||
from datetime import UTC, datetime, timedelta, timezone
|
||
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
from web.backend.database import Base, UTCDateTime
|
||
from web.backend.models import AutomationRun, LoginTask
|
||
|
||
|
||
class TestUtcDateTime:
|
||
def setup_method(self):
|
||
self.engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(self.engine)
|
||
self.session = sessionmaker(bind=self.engine)()
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def _row(self, started_at):
|
||
self.session.add(
|
||
AutomationRun(job_key="k", started_at=started_at, status="success")
|
||
)
|
||
self.session.commit()
|
||
return self.session.query(AutomationRun).one()
|
||
|
||
def test_read_back_is_timezone_aware(self):
|
||
row = self._row(datetime(2026, 10, 2, 12, 0, tzinfo=UTC))
|
||
assert row.started_at.tzinfo is not None
|
||
assert row.started_at.utcoffset() == timedelta(0)
|
||
|
||
def test_read_back_value_can_compare_with_now(self):
|
||
"""这条就是当初炸掉的写法：库里的时间直接和 datetime.now(UTC) 比。"""
|
||
row = self._row(datetime.now(UTC) - timedelta(minutes=1))
|
||
assert row.started_at < datetime.now(UTC) # 不再 TypeError
|
||
|
||
def test_write_converts_local_aware_to_utc(self):
|
||
"""带 +08:00 的时间要按 UTC 落库，不能把本地壁钟直接存进去。"""
|
||
local = datetime(2026, 10, 2, 20, 0, tzinfo=timezone(timedelta(hours=8)))
|
||
row = self._row(local)
|
||
assert row.started_at == datetime(2026, 10, 2, 12, 0, tzinfo=UTC)
|
||
raw = self.session.execute(
|
||
text("SELECT started_at FROM automation_runs")
|
||
).scalar()
|
||
assert str(raw).startswith("2026-10-02 12:00")
|
||
|
||
def test_naive_write_is_treated_as_utc(self):
|
||
"""库里读出来的裸值再写回去（常见于搬迁/复制）不能被当成本地时间再减 8 小时。"""
|
||
naive_utc = datetime(2026, 10, 2, 12, 0) # noqa: DTZ001 刻意构造裸 UTC
|
||
row = self._row(naive_utc)
|
||
assert row.started_at == datetime(2026, 10, 2, 12, 0, tzinfo=UTC)
|
||
|
||
def test_none_stays_none(self):
|
||
assert UTCDateTime().process_bind_param(None, None) is None
|
||
assert UTCDateTime().process_result_value(None, None) is None
|
||
|
||
def test_all_model_datetime_columns_use_utc_type(self):
|
||
"""新增时间列若忘了用 UTCDateTime，这里直接失败。"""
|
||
offenders = []
|
||
for table in Base.metadata.sorted_tables:
|
||
for column in table.columns:
|
||
if isinstance(column.type, UTCDateTime):
|
||
continue
|
||
if str(column.type).upper().startswith("DATETIME"):
|
||
offenders.append(f"{table.name}.{column.name}")
|
||
assert not offenders, f"这些时间列没用 UTCDateTime: {offenders}"
|
||
|
||
def test_login_task_attempt_at_round_trips_as_utc(self):
|
||
"""真实模型抽查：写 aware、读 aware、能直接比较。"""
|
||
task = LoginTask(
|
||
batch_id="b",
|
||
account_id=1,
|
||
created_by=1,
|
||
attempt_at=datetime(2026, 10, 2, 12, 0, tzinfo=UTC),
|
||
)
|
||
self.session.add(task)
|
||
self.session.commit()
|
||
self.session.refresh(task)
|
||
assert task.attempt_at == datetime(2026, 10, 2, 12, 0, tzinfo=UTC)
|
||
# 模型侧保证 aware，所以业务代码可以直接比较
|
||
assert task.attempt_at < datetime.now(UTC)
|