From 28ebf723c8cd722a6bca306c9f1549169698ab8e Mon Sep 17 00:00:00 2001 From: modusensus Date: Sun, 20 Sep 2026 11:55:28 +0800 Subject: [PATCH] =?UTF-8?q?test(health):=20=E5=BE=AA=E7=8E=AF=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E4=B8=8D=E5=86=8D=E6=96=AD=E8=A8=80=E5=A2=99=E9=92=9F?= =?UTF-8?q?=E9=80=9F=E7=8E=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main 上 Python 3.11 (macos-latest) 变红：test_run_loop_stops_cleanly 的 assert len(seen) >= 3 拿到 2。原因不在产品代码，而是这个断言考的是"这台机器够快"： run_loop 每轮是 check() + wait(interval)，测试睡 0.3 秒就要求至少 3 次，而 CI 记录里两次检查相隔 0.12 秒——带 coverage 插桩的 runner 每轮比标称慢 2.4 倍，2 倍的 余量被吃掉了。同文件的 test_run_loop_callback_error_swallowed 是同类（睡 0.15 秒 后断言回调异常已记录）。 改法： - 次数改为带截止时间的等待（新增 _wait_for，_wait_for_values 复用它），不再睡固定秒数； - "停下来"改用 join 证明循环线程真的退出，而不是睡一段时间再看计数没变；停止后至多 一次在途回调是实现的承诺，也写进断言； - isinstance 检查挪到 join 之后：线程死后计数不再变，才不是"边遍历边被改"。 验证：用插件拖慢每轮 check 模拟慢 runner——0.1 秒/轮时旧版本复现出 CI 的同一条 AssertionError: 2，0.2 秒/轮时两条测试都挂；修好后 0.2 与 0.5 秒/轮均全过，本地 连跑 10 次稳定。 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- tests/test_health.py | 62 ++++++++++++++++++++++++++++++++++---------- 1 file changed, 48 insertions(+), 14 deletions(-) diff --git a/tests/test_health.py b/tests/test_health.py index 5e1df05..9ae03d4 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -4,6 +4,7 @@ import threading import time +from collections.abc import Callable import pytest @@ -158,32 +159,48 @@ def test_remote_check_disabled() -> None: def test_run_loop_stops_cleanly() -> None: - # Use a healthy manager: with backoff enabled an unhealthy one would grow - # the interval and could starve the ">= 3 checks" assertion below. + """循环反复检查，而设置停止事件后线程真的结束了。 + + 不按墙钟断言："睡 0.3 秒后至少 3 次检查"考的是**这台机器够快**，而 CI 上带 + coverage 插桩的 macOS runner 把 ``run_loop`` 的一轮从标称 50ms 拉到 120ms， + 于是等到 2 次就断言失败（main 上真实发生过）。这里改成：次数用带截止时间的 + 等待，"停下来"则用 join 证明——那才是测试名字里的 stops cleanly。 + """ + # 用健康的 manager：不健康会触发退避，间隔指数增长，把"若干次检查"饿死。 hc = HealthChecker(_TM(alive=True, ports="list"), _hc()) seen: list[HealthStatus] = [] + before_threads = set(threading.enumerate()) stop = hc.run_loop(interval=0.05, callback=lambda st: seen.append(st)) assert isinstance(stop, threading.Event) - time.sleep(0.3) + monitor = next((t for t in threading.enumerate() if t not in before_threads), None) + assert monitor is not None, "run_loop 没有起后台线程" + + _wait_for_values(seen, 3) + + checks_at_stop = len(seen) stop.set() - time.sleep(0.15) - assert len(seen) >= 3, len(seen) + monitor.join(timeout=5) + assert not monitor.is_alive(), "设置停止事件后循环没有退出" + # 已经进入 check() 的那一轮仍会回调一次（实现只保证在 wait 之前查一次事件）， + # 再多就说明停止没生效。线程已 join，计数此后不会再变，所以这是确定性的。 + late = len(seen) - checks_at_stop + assert late <= 1, f"停止后又回调了 {late} 次" assert all(isinstance(st, HealthStatus) for st in seen) - n = len(seen) - time.sleep(0.15) - assert len(seen) == n, (len(seen), n) def test_run_loop_callback_error_swallowed() -> None: + """回调抛出的异常被记下来，而不是把监控线程弄死。""" hc = HealthChecker(_TM(alive=True, ports="dict"), _hc()) def bad_cb(_st: HealthStatus) -> None: raise ValueError("cb boom") stop = hc.run_loop(interval=0.02, callback=bad_cb) - time.sleep(0.15) + _wait_for( + lambda: isinstance(hc.last_callback_error, ValueError), + lambda: f"回调异常没被记录：{hc.last_callback_error!r}", + ) stop.set() - assert isinstance(hc.last_callback_error, ValueError) def test_health_status_str() -> None: @@ -281,12 +298,29 @@ def test_backoff_interval_formula() -> None: assert HealthChecker._backoff_interval(60.0, 10, 300.0) == 300.0 -def _wait_for_values(values: list, n: int, timeout: float = 3.0) -> None: - """Busy-wait until ``values`` has at least ``n`` entries.""" +def _wait_for( + predicate: Callable[[], bool], + describe: Callable[[], str], + timeout: float = 5.0, +) -> None: + """Wait for *predicate* on a deadline, instead of napping a fixed time. + + 固定秒数的 ``sleep`` 断言的是"这台机器够快"，而不是被测行为。``describe`` + 是个函数：失败信息要在**等过之后**才取，否则它记的是等待前那一刻的状态。 + """ deadline = time.time() + timeout - while len(values) < n and time.time() < deadline: + while not predicate() and time.time() < deadline: time.sleep(0.01) - assert len(values) >= n, f"captured only {len(values)} values, need {n}" + assert predicate(), describe() + + +def _wait_for_values(values: list, n: int, timeout: float = 5.0) -> None: + """Busy-wait until ``values`` has at least ``n`` entries.""" + _wait_for( + lambda: len(values) >= n, + lambda: f"captured only {len(values)} values, need {n}", + timeout, + ) def _fake_wait_recorder(