""" test_bootstrap.py — U1 地基验证:跨包 import + 代理旁路 + key 注入 + M3 model 装配。 须用 venv python 跑(import agentscope): cheap-worker/.venv/bin/python cheap-worker/tests/test_bootstrap.py """ import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/ import _bootstrap def test_cross_package_import(): """加 sys.path 后 from worker import config, client 不抛 ImportError,且拿到关键 API。""" fw = _bootstrap.load_framework() assert {"config", "client", "middleware"} <= set(fw) from worker import config, client # noqa: E402 assert hasattr(config, "build_model_openai") assert hasattr(client, "install_proxy_bypass") assert hasattr(client, "get_api_key") def test_proxy_bypass(): """install_proxy_bypass 把内网网关 host 并入 NO_PROXY / no_proxy 两个 env。""" from worker import client # noqa: E402 host = client.install_proxy_bypass("http://100.64.0.8:3000") assert host == "100.64.0.8" assert "100.64.0.8" in os.environ.get("NO_PROXY", "") assert "100.64.0.8" in os.environ.get("no_proxy", "") def test_api_key_from_doc(): """key 缺 env 时从 docs/内网凭据与端点.md 解析注入(内网单一事实源)。""" # 清掉 env 模拟"未设",强制走文档解析路径 os.environ.pop("NEWAPI_KEY", None) key = _bootstrap.ensure_api_key_env() assert key.startswith("sk-") assert os.environ.get("NEWAPI_KEY", "").startswith("sk-") def test_model_assembly(): """build_cheap_model 返回 OpenAIChatModel(base_url 带 /v1 由 build_model_openai 保证),model 名为 MiniMax-M3。""" model = _bootstrap.build_cheap_model(max_tokens=8000) from agentscope.model import OpenAIChatModel # noqa: E402 assert isinstance(model, OpenAIChatModel) # 模型名固定 M3(KTD6)。属性名在不同小版本可能差异,宽松双路断言。 assert getattr(model, "model", None) == _bootstrap.SPIKE_MODEL or _bootstrap.SPIKE_MODEL in repr(model) if __name__ == "__main__": _fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] _failed = 0 for _fn in _fns: try: _fn() print(f" PASS {_fn.__name__}") except Exception as e: # noqa: BLE001 地基测试要看到任何异常类型 _failed += 1 print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}") print(f"\n{len(_fns) - _failed}/{len(_fns)} passed") sys.exit(1 if _failed else 0)