diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index d6256c3..cb0626b 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -1930,7 +1930,9 @@ class TestSQLiteStatusVisibility: class TestPostgresStatusVisibility: """PG 侧的降级必须能被下游查到(计划 T2 建立可见性,T5 改判据)。""" - def _recorder(self, conn): + def _recorder(self, conn, now=None): + """假时钟是缺省: 快照里的 `retry_after_s`/`degraded_for_s` 是**时间差**, + 用真实时钟断言就等于断言"这几行代码零耗时",是设计上就会间歇红的用例。""" from polygateway.telemetry.postgres import PostgresRecorder return PostgresRecorder( @@ -1939,6 +1941,7 @@ class TestPostgresStatusVisibility: auto_migrate=True, pool_max=_TEST_POOL_MAX, write_timeout_s=_TEST_WRITE_TIMEOUT_S, + now=now or _FakeClock(), ) async def test_unusable_table_shows_up_in_the_status(self, captured_warnings): @@ -1950,7 +1953,7 @@ class TestPostgresStatusVisibility: status = recorder.telemetry_status assert status.degraded is True and status.fatal is False # 环境级: 建了表就该自愈 assert status.dropped_rows == 1 # 降级那一次调用本身也丢了一行 - assert status.retry_after_s == pytest.approx(_DEGRADE_COOLDOWN_S) + assert status.retry_after_s == _DEGRADE_COOLDOWN_S # 假时钟不动,余额恰是整个冷却期 async def test_healthy_recorder_is_not_degraded(self): recorder = self._recorder(_FakePgConn(list(_EXPECTED_COLUMNS)))