test(c2): 补齐导入次序回归,注明环状导入为预期且可自愈
test_bzzoiro_source_registry.py 增至 11 项: - 新增 TestImportOrderSelfHeal:用独立子进程验证两种真实导入次序 (sources 先 / bzzoiro 先)下注册表均可用。该缺陷只在真实模块初始化 时序下成立,mock 照不出来,故不 mock。 - 新增未知名字仍抛 ValueError 的契约守卫。 src/data/sources.py:补全 _load_sources 文档,说明首次装载失败属正常的 环状导入(bzzoiro 仍在初始化),失败时保持 _loaded=False 使后续 get_source()/list_sources() 调用可重试装载(自愈)。 验证: - tests/test_bzzoiro_source_registry.py:11 passed - 两种导入次序实测均输出 bzzoiro ['bzzoiro']
This commit is contained in:
@@ -78,6 +78,58 @@ class TestIngestContract:
|
||||
assert name in params, f"fetch_bzzoiro_events() 缺少 {name!r}"
|
||||
|
||||
|
||||
class TestImportOrderSelfHeal:
|
||||
"""导入次序不得影响注册表 —— 这是本缺陷的第二个隐藏面。
|
||||
|
||||
背景:src.api.app 先 `import src.data.sources`,其模块级预热在
|
||||
`src.data.bzzoiro` 尚未初始化时发起,导入链在 sources.py 内成环,
|
||||
首次导入必然失败(bzzoiro 仍在加载中)。修复前 `except Exception: pass`
|
||||
把这次失败同时变得「无声」且「不可恢复」,注册表就永久空掉了。
|
||||
|
||||
这里用独立子进程验证真实导入次序,不 mock —— 因为该缺陷只在
|
||||
真实的模块初始化时序下才成立,单元测试里的 mock 反而照不出来。
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _run(import_lines: list[str]) -> str:
|
||||
"""在全新解释器中按指定次序导入,返回 get_source/list_sources 结果。"""
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
root = Path(__file__).resolve().parent.parent
|
||||
code = (
|
||||
"import sys; sys.path.insert(0, r'%s')\n" % root
|
||||
+ "\n".join(import_lines)
|
||||
+ "\nfrom src.data.sources import get_source, list_sources\n"
|
||||
"print('RESULT', get_source('bzzoiro').name, list_sources())\n"
|
||||
)
|
||||
proc = subprocess.run(
|
||||
[sys.executable, "-c", code],
|
||||
capture_output=True, text=True, timeout=120, cwd=str(root),
|
||||
)
|
||||
assert proc.returncode == 0, (
|
||||
f"导入次序 {import_lines} 下 get_source('bzzoiro') 失败:\n{proc.stderr[-1500:]}"
|
||||
)
|
||||
return proc.stdout.strip()
|
||||
|
||||
def test_sources_first_then_bzzoiro(self):
|
||||
"""次序 B:sources 先导入(正常应用路径)。"""
|
||||
out = self._run(["import src.data.sources"])
|
||||
assert out == "RESULT bzzoiro ['bzzoiro']", out
|
||||
|
||||
def test_bzzoiro_first_then_sources(self):
|
||||
"""次序 A:bzzoiro 先导入 —— 注册表必须仍然可用(自愈)。"""
|
||||
out = self._run(["import src.data.bzzoiro"])
|
||||
assert out == "RESULT bzzoiro ['bzzoiro']", (
|
||||
out + " —— 注册表为空说明首次装载失败后没有再重试(静默失效回归)"
|
||||
)
|
||||
|
||||
def test_unknown_name_still_raises(self):
|
||||
"""真正未知的名字仍须抛 ValueError —— 修复不应放宽这一契约。"""
|
||||
with pytest.raises(ValueError, match="未知数据源"):
|
||||
get_source("no_such_source_xyz")
|
||||
|
||||
|
||||
class TestNoSilentImportSwallow:
|
||||
"""sources.py 不许再用 `except Exception: pass` 吞掉导入失败。"""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user