# -*- coding: utf-8 -*- """推理后端抽象接口（backends，Issue #57，PRD 5.6）单元测试。 覆盖： - 抽象基类不可直接实例化（必须由子类实现四个生命周期方法）； - 值对象 BackendCapabilities / BackendHealth / InferResult 的字段与序列化； - LocalBackend / CloudBackend 占位实现的生命周期（load/infer/health/unload）与幂等； - 向后兼容：``generate`` 转发到 ``infer`` 并返回 ``text``； - 能力声明差异（本地出厂内闭环 / 云端出厂外）； - 注册表与 ``build_backend`` 的配置驱动构造 + 未知后端报错。 """ import os import sys import unittest from abc import ABC sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import _bootstrap # noqa: F401 from llm_gateway.backends import ( # noqa: E402 BackendCapabilities, BackendHealth, CloudBackend, InferResult, InferenceBackend, LocalBackend, _PlaceholderBackend, build_backend, default_registry, ) # --------------------------------------------------------------------------- # 抽象基类契约 # --------------------------------------------------------------------------- class AbstractionContractTest(unittest.TestCase): """PRD 5.6：InferenceBackend 是抽象接口，业务代码只依赖它。""" def test_cannot_instantiate_abstract_base(self): # 缺少四个抽象方法 → 不能实例化 with self.assertRaises(TypeError): InferenceBackend() # noqa: E721 def test_is_abc_subclass(self): self.assertTrue(issubclass(InferenceBackend, ABC)) def test_required_abstract_methods(self): # PRD 5.6 明列的生命周期动作 abstract = InferenceBackend.__abstractmethods__ for name in ("load_model", "infer", "health_check", "unload"): self.assertIn(name, abstract) def test_concrete_backends_are_inference_backends(self): for cls in (LocalBackend, CloudBackend): self.assertTrue(issubclass(cls, InferenceBackend), f"{cls.__name__} 必须实现 InferenceBackend") # --------------------------------------------------------------------------- # 值对象 # --------------------------------------------------------------------------- class BackendCapabilitiesTest(unittest.TestCase): def test_defaults(self): cap = BackendCapabilities() self.assertFalse(cap.streaming) self.assertIsNone(cap.max_concurrency) self.assertFalse(cap.on_premises) self.assertEqual(cap.modalities, ("text",)) def test_supports_modality(self): cap = BackendCapabilities(modalities=("text", "image")) self.assertTrue(cap.supports("text")) self.assertTrue(cap.supports("image")) self.assertFalse(cap.supports("audio")) def test_to_dict_roundtrip(self): cap = BackendCapabilities(streaming=True, max_concurrency=4, on_premises=False, modalities=("text",)) d = cap.to_dict() self.assertEqual(d["streaming"], True) self.assertEqual(d["max_concurrency"], 4) self.assertEqual(d["modalities"], ["text"]) class BackendHealthTest(unittest.TestCase): def test_fields(self): h = BackendHealth(healthy=True, detail="ok") self.assertTrue(h.healthy) self.assertEqual(h.detail, "ok") self.assertTrue(h.checked_at) # 自动生成时间戳 def test_to_dict(self): d = BackendHealth(healthy=False, detail="down").to_dict() self.assertEqual(d["healthy"], False) self.assertIn("checked_at", d) class InferResultTest(unittest.TestCase): def test_required_fields(self): r = InferResult(text="hello", backend_name="local-70b") self.assertEqual(r.text, "hello") self.assertEqual(r.backend_name, "local-70b") self.assertIsNone(r.prompt_tokens) def test_to_dict(self): r = InferResult(text="a", backend_name="b", model_id="m", prompt_tokens=3, completion_tokens=5) d = r.to_dict() self.assertEqual(d["text"], "a") self.assertEqual(d["prompt_tokens"], 3) self.assertEqual(d["completion_tokens"], 5) # --------------------------------------------------------------------------- # 占位实现生命周期 # --------------------------------------------------------------------------- class PlaceholderLifecycleTest(unittest.TestCase): def setUp(self): self.b = LocalBackend() def test_health_reflects_load_state(self): # 未加载 → 不健康 self.assertFalse(self.b.health_check().healthy) self.b.load_model("local-70b-base") self.assertTrue(self.b.health_check().healthy) def test_load_is_idempotent(self): self.b.load_model("local-70b-base") # 重复加载同一 model_id 不报错 self.b.load_model("local-70b-base") self.assertTrue(self.b.health_check().healthy) def test_infer_lazy_loads_when_not_loaded(self): # 演示态：未显式 load_model 也能 infer（惰性自加载） r = self.b.infer("炉温是多少", context=["SOP-炉温"]) self.assertIsInstance(r, InferResult) self.assertEqual(r.backend_name, "local-70b") self.assertIn("炉温是多少", r.text) self.assertIn("[来源: SOP-炉温]", r.text) def test_infer_after_explicit_load(self): self.b.load_model("local-70b-base") r = self.b.infer("hello") self.assertEqual(r.model_id, "local-70b-base") self.assertIn("hello", r.text) def test_unload_is_idempotent(self): self.b.load_model("local-70b-base") self.b.unload() self.assertFalse(self.b.health_check().healthy) # 未加载再 unload 也不报错 self.b.unload() def test_echo_context_disabled(self): b = LocalBackend(echo_context=False) b.load_model("m") r = b.infer("q", context=["src1", "src2"]) self.assertNotIn("[来源:", r.text) # --------------------------------------------------------------------------- # 向后兼容：generate 转发到 infer # --------------------------------------------------------------------------- class BackwardCompatGenerateTest(unittest.TestCase): def test_generate_returns_text_of_infer(self): b = CloudBackend() b.load_model("cloud-qwen-plus") txt = b.generate("海绵钛是什么", context=["科普手册"]) # 与 infer().text 一致 self.assertEqual(txt, b.infer("海绵钛是什么", context=["科普手册"]).text) self.assertIn("云端API占位", txt) self.assertIn("[来源: 科普手册]", txt) def test_gateway_still_works_with_new_backends(self): # 集成校验：LLMGateway.ask() 经 generate 路径仍正常（不导入失败）。 # 复用 test_gateway.py 的模板配置加载 prompts，避免默认空注册表 KeyError。 from llm_gateway.dlp import DlpEngine from llm_gateway.gateway import LLMGateway from llm_gateway.prompts import PromptRegistry from llm_gateway.router import SensitivityRouter cfg_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) prompts = PromptRegistry.from_template_config( os.path.join(cfg_dir, "config", "prompts.template.yaml")) router = SensitivityRouter.from_template_config( os.path.join(cfg_dir, "config", "router.template.yaml")) gw = LLMGateway( dlp=DlpEngine(), router=router, prompts=prompts, local=LocalBackend(), cloud=CloudBackend()) result = gw.ask("海绵钛是什么", rag_context=["科普手册"]) self.assertTrue(result.answer) # 后端占位回显特征仍在（证明走的是新 backends 的 generate 路径） self.assertIn("云端API占位", result.answer) # --------------------------------------------------------------------------- # 能力声明差异（本地 vs 云端） # --------------------------------------------------------------------------- class CapabilitiesDifferenceTest(unittest.TestCase): def test_local_is_on_premises(self): cap = LocalBackend().capabilities self.assertTrue(cap.on_premises) self.assertTrue(cap.streaming) self.assertGreater(cap.max_concurrency, 0) def test_cloud_is_off_premises(self): cap = CloudBackend().capabilities self.assertFalse(cap.on_premises) self.assertTrue(cap.streaming) def test_local_and_cloud_differ_on_premises(self): # 关键差异：本地出厂内闭环，云端数据出厂 self.assertNotEqual( LocalBackend().capabilities.on_premises, CloudBackend().capabilities.on_premises, ) # --------------------------------------------------------------------------- # 注册表与配置驱动构造 # --------------------------------------------------------------------------- class RegistryTest(unittest.TestCase): def test_default_registry_has_known_backends(self): reg = default_registry() self.assertIn("local-70b", reg) self.assertIn("cloud-api", reg) self.assertIs(reg["local-70b"], LocalBackend) self.assertIs(reg["cloud-api"], CloudBackend) def test_build_backend_by_name(self): b = build_backend("local-70b") self.assertIsInstance(b, LocalBackend) self.assertIsInstance(b, InferenceBackend) self.assertEqual(b.name, "local-70b") def test_build_unknown_backend_raises_with_hint(self): with self.assertRaises(ValueError) as ctx: build_backend("npu-cann") # 尚未实现（#59 才接入） self.assertIn("npu-cann", str(ctx.exception)) self.assertIn("local-70b", str(ctx.exception)) # 提示已知项 def test_build_passes_kwargs(self): b = build_backend("cloud-api", echo_context=False) self.assertIsInstance(b, CloudBackend) self.assertFalse(b.echo_context) # --------------------------------------------------------------------------- # 自定义后端通过实现接口接入（证明「业务代码不感知硬件」） # --------------------------------------------------------------------------- class CustomBackendImplementationTest(unittest.TestCase): """模拟 #59 昇腾后端：只需实现四个方法即可被当作 InferenceBackend 使用。""" def test_custom_backend_satisfies_interface(self): class NpuCannBackend(InferenceBackend): name = "npu-cann" def __init__(self): self._loaded = False def load_model(self, model_id): self._loaded = True def infer(self, prompt, context=None): if not self._loaded: self.load_model("ascend-cann") return InferResult(text=f"[NPU] {prompt}", backend_name=self.name) def health_check(self): return BackendHealth(healthy=self._loaded) def unload(self): self._loaded = False b = NpuCannBackend() self.assertIsInstance(b, InferenceBackend) self.assertFalse(b.health_check().healthy) b.load_model("ascend-cann") self.assertTrue(b.health_check().healthy) self.assertEqual(b.infer("q").text, "[NPU] q") # generate 兼容路径 self.assertEqual(b.generate("q", context=[]), "[NPU] q") b.unload() self.assertFalse(b.health_check().healthy) if __name__ == "__main__": unittest.main()