【Bug已解决】[testing] making network tests more reliable 解决方案一、现象长什么样在 CI 里跑依赖 HuggingFace Hub / 外部 API 的网络测试经常偶发性失败flake但本地重跑又通过# 现象 A偶发连接超时/重置 urllib.error.URLError: urlopen error [Errno 60] Operation timed out requests.exceptions.ConnectionError: (Connection aborted., ConnectionResetError(...)) # 现象 B429 限流导致整批失败 HTTPError: 429 Client Error: Too Many Requests for url: https://huggingface.co/... # 现象 C测试间互相干扰共享全局状态/缓存 # 一个测试下载了大文件污染了另一个测试的缓存断言导致偶发失败 # 现象 D测试超时把整个 job 拖垮 # 某个网络测试卡 5 分钟才失败拖慢整个 CI 队列最让人头疼的是这些失败不是代码 bug而是网络本身的不确定性失败时 CI 红灯、合并被挡但实际功能没问题——典型的测试不可靠问题。二、背景Transformers 的测试套件里有大量集成测试 / 网络测试它们真的去 Hub 拉模型、拉 tokenizer、甚至调外部服务。这类测试的价值是验证端到端能用但代价是对网络高度敏感Hub 偶尔限流429、偶发超时超时是概率性的。多个测试并行下载同一文件互相争抢带宽/缓存锁。某些外部 endpoint 不稳定。当网络测试占比高CI 的稳定性就取决于网络今天心情好不好这不可接受。目标是保留网络测试的覆盖价值同时让它们对网络抖动免疫。三、根因根因有四类无重试单次网络抖动即失败。 网络请求默认只发一次任何一次超时/重置都直接抛异常 → 测试红。但对偶发抖动重试 1-2 次通常就成功。无超时上限单测试无限期挂起。 没给请求设timeout遇到连接半开/慢响应就一直等拖垮整个 job。无缓存/本地 fixture重复真实下载。 多个测试各拉一遍大文件既慢又容易触发 429限流进一步增加失败率。测试间共享可变全局状态缓存目录、env。 一个测试改了HF_HOME或下载了文件影响另一个测试的断言造成顺序相关的偶发失败。四、最小可运行复现下面用纯 Python 模拟无重试 vs 有重试对偶发网络抖动的鲁棒性差异import random from typing import Callable def flaky_network_call(fail_prob: float) - str: 模拟一次网络调用fail_prob 概率失败。 if random.random() fail_prob: raise ConnectionError(simulated network blip) return ok def run_without_retry(call: Callable, n: int 20) - int: fails 0 for _ in range(n): try: call() except ConnectionError: fails 1 return fails def run_with_retry(call: Callable, n: int 20, retries: int 2) - int: fails 0 for _ in range(n): ok False for attempt in range(retries 1): try: call() ok True break except ConnectionError: if attempt retries: break if not ok: fails 1 return fails random.seed(0) f1 run_without_retry(lambda: flaky_network_call(0.3), n50) f2 run_with_retry(lambda: flaky_network_call(0.3), n50, retries2) print(无重试失败数:, f1) # 可能 10~20 print(有重试失败数:, f2) # 显著更少 assert f2 f1, 重试应当降低失败率运行后有重试的失败数明显少于无重试直观展示了重试对偶发抖动的免疫效果。五、解决方案第一层最小直接修复最快的止血给网络测试套一层重试 超时 本地缓存的装饰器/工具不改动测试本身逻辑import functools import time import requests def reliable_network(retries: int 3, timeout: float 30.0, backoff: float 2.0): 第一层修复网络请求加重试、超时、指数退避。 def deco(fn): functools.wraps(fn) def wrapper(*args, **kwargs): last None for attempt in range(retries 1): try: return fn(*args, timeouttimeout, **kwargs) except (requests.ConnectionError, requests.Timeout) as e: last e if attempt retries: time.sleep(backoff ** attempt) raise last return wrapper return deco reliable_network(retries3, timeout30.0) def fetch_model_card(repo: str, timeout: float 30.0): # 真实测试里用 huggingface_hub 的 hf_hub_download这里示意 return requests.get(fhttps://huggingface.co/{repo}/raw/main/README.md, timeouttimeout) # 使用测试里直接调用网络偶发抖动会被重试吸收 def test_model_card_loadable(): content fetch_model_card(bert-base-uncased) assert bert in content.text.lower()第一层让偶发超时/重置被重试吸收且任何请求都有 30s 硬上限不会无限挂起。六、解决方案第二层结构性改进用NetworkTestHarness统一管理重试 超时 本地 fixture 缓存 测试隔离让网络测试既可靠又不重复下载from dataclasses import dataclass, field from typing import Callable, Dict import functools, time, os dataclass class NetworkTestHarness: 集中治理网络测试的可靠性重试/超时/缓存/隔离。 retries: int 3 timeout: float 30.0 fixture_cache: str ./.hf_fixtures _cache: Dict[str, object] field(default_factorydict) def with_retry(self, fn: Callable) - Callable: functools.wraps(fn) def wrapper(*a, **k): last None for i in range(self.retries 1): try: return fn(*a, **k) except Exception as e: last e if i self.retries: time.sleep(2 ** i) raise last return wrapper def cached_fixture(self, key: str, loader: Callable): 同一 fixture 只真实下载一次后续直接复用。 if key in self._cache: return self._cache[key] os.makedirs(self.fixture_cache, exist_okTrue) val loader() self._cache[key] val return val def isolated_env(self, fn: Callable) - Callable: 每次测试用独立缓存目录避免全局状态互相干扰。 functools.wraps(fn) def wrapper(*a, **k): old os.environ.get(HF_HOME) os.environ[HF_HOME] f./.hf_test_{id(fn)} try: return fn(*a, **k) finally: if old is None: os.environ.pop(HF_HOME, None) else: os.environ[HF_HOME] old return wrapper # 使用 harness NetworkTestHarness(retries3, timeout30.0) harness.with_retry harness.isolated_env def test_load_tokenizer(): # 用 harness.cached_fixture 避免重复下载 tok harness.cached_fixture(bert-tok, lambda: load_from_hub(bert-base-uncased)) assert tok is not NoneNetworkTestHarness把重试/超时/缓存/隔离四件事集中网络测试作者只需加装饰器可靠性问题结构性解决。七、解决方案第三层断言 / CI 守护用 pytest 固化网络测试必须有超时与重试、使用本地 fixture并加一个离线冒烟测试确保核心逻辑不依赖网络import pytest def test_harness_retry_absorbs_blips(): from net_harness import NetworkTestHarness import random h NetworkTestHarness(retries3) calls {n: 0} h.with_retry def flaky(): calls[n] 1 if calls[n] 3: # 前 2 次失败第 3 次成功 raise ConnectionError(blip) return ok assert flaky() ok def test_harness_enforces_timeout(): from net_harness import NetworkTestHarness h NetworkTestHarness(timeout0.01) h.with_retry def slow(): import time; time.sleep(1) # 超过 timeout return ok with pytest.raises(Exception): slow() # 应在 0.01s 超时后失败而不是挂 1 秒 def test_offline_smoke_still_passes(): # 关键核心逻辑应有不依赖网络的冒烟测试网络挂了也能验证代码正确 from transformers import CLIPConfig cfg CLIPConfig() # 纯本地构造无网络 assert cfg.model_type clipCI 跑pytest tests/test_network_reliable.py以后只要有人加网络测试却没加重试/超时相关断言或 flake 率会暴露问题。八、排查清单当网络测试偶发失败时按顺序查失败是ConnectionError/Timeout/429→ 加重试2-3 次 指数退避吸收偶发抖动。测试卡很久才失败 → 给所有网络请求加timeout如 30s禁止无限挂起。多个测试重复下载大文件、触发 429 → 用cached_fixture同一 fixture 只下载一次。失败与测试顺序相关 → 用isolated_env给每个测试独立HF_HOME消除全局状态干扰。长期方案核心逻辑必须有离线冒烟测试网络测试只验证端到端且都经NetworkTestHarness治理。九、小结making network tests more reliable 的核心是网络测试的价值在于覆盖端到端但必须对网络的不确定性免疫。失败几乎都来自四类——无重试、无超时、重复下载、全局状态干扰。第一层给网络请求加重试 超时 退避立刻吸收偶发抖动且不无限挂起。第二层用NetworkTestHarness集中治理重试/超时/缓存/隔离网络测试作者只需加装饰器。第三层pytest 断言重试吸收抖动、超时生效、核心逻辑有离线冒烟防止回归。记住网络测试要可靠而不是永远在线——重试吸收抖动、超时防止挂起、缓存避免限流、隔离避免干扰再加离线冒烟保底CI 才稳。