From 69a5b5fadba577e470628c842a482cd22f8a22ba Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 24 Aug 2026 11:03:33 -0400 Subject: [PATCH] test: pin the cooldown assertion to a fake clock The status snapshot reports elapsed time, so asserting retry_after_s against the real monotonic clock was really asserting that a few lines of code take zero time; it failed at 59.99993 vs 60.0. The recorder already accepts an injected clock for exactly this reason. --- tests/unit/test_telemetry.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) 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)))