feat: 完成 issue #45 ④ 云端 API（Qwen/DeepSeek）接入与安全网关

core/llm-gateway/tests/test_cloudapi.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+"""云端 API 推理后端 + 安全网关测试（issue #45）。
+
+覆盖：
+1. 参数化与配置资产（provider/endpoint/model/api_key_env）；
+2. API Key 从环境变量读取（不硬编码、health 不泄露密钥）；
+3. dry-run（未配置 endpoint）占位输出；
+4. OpenAI 兼容调用：Bearer 认证头 + mock 响应提取；
+5. 安全网关：safety_checker 拒绝 → 拦截占位、不调用上游；
+6. 坏响应 → RuntimeError。
+"""
+import os
+import sys
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401
+
+from llm_gateway.backends import CloudApiBackend  # noqa: E402
+
+CONFIG = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+    "config", "cloudapi.template.yaml",
+)
+
+
+class TestParameterization(unittest.TestCase):
+    """参数化与配置资产。"""
+
+    def test_config_asset_parses(self):
+        import yaml
+        with open(CONFIG, "r", encoding="utf-8") as fh:
+            cfg = yaml.safe_load(fh)["cloudapi"]
+        b = CloudApiBackend(**{k: v for k, v in cfg.items() if k != "provider"})
+        self.assertEqual(b.endpoint, "https://api.deepseek.com/v1")
+        self.assertEqual(b.model, "deepseek-chat")
+        self.assertEqual(b.api_key_env, "DEEPSEEK_API_KEY")
+
+    def test_api_key_from_env(self):
+        with mock.patch.dict(os.environ, {"MY_KEY": "sk-secret-123"}):
+            b = CloudApiBackend(endpoint="http://x/v1", api_key_env="MY_KEY")
+            self.assertEqual(b._api_key, "sk-secret-123")
+
+    def test_health_does_not_leak_key(self):
+        with mock.patch.dict(os.environ, {"K": "sk-secret-123"}):
+            b = CloudApiBackend(api_key_env="K")
+            health = b.health()
+        self.assertEqual(health["api_key_configured"], True)
+        self.assertNotIn("sk-secret", str(health))
+
+    def test_name(self):
+        self.assertEqual(CloudApiBackend().name, "cloud-api")
+
+
+class TestDryRun(unittest.TestCase):
+    """未配置 endpoint：占位输出。"""
+
+    def setUp(self):
+        self.b = CloudApiBackend()
+
+    def test_dry_run_placeholder(self):
+        out = self.b.generate("海绵钛是什么", ["公开资料"])
+        self.assertIn("[云端API占位]", out)
+        self.assertIn("[来源: 公开资料]", out)
+
+
+class TestOpenAICompat(unittest.TestCase):
+    """OpenAI 兼容调用（Bearer 认证）。"""
+
+    def setUp(self):
+        with mock.patch.dict(os.environ, {"DK": "sk-abc"}):
+            self.b = CloudApiBackend(
+                endpoint="https://api.deepseek.com/v1", api_key_env="DK")
+
+    def test_generate_with_bearer_auth(self):
+        fake = {"choices": [{"message": {"content": "海绵钛是钛的一种形态"}}]}
+        with mock.patch.object(self.b, "_post_json", return_value=fake) as post:
+            out = self.b.generate("海绵钛是什么", ["公开资料"])
+        path, payload = post.call_args[0]
+        self.assertEqual(path, "/v1/chat/completions")
+        self.assertEqual(payload["model"], "deepseek-chat")
+        self.assertEqual(out, "海绵钛是钛的一种形态")
+
+    def test_bearer_header_sent(self):
+        # 验证 _post_json 实际携带 Authorization: Bearer
+        opened = []
+        def fake_urlopen(req, timeout=None):
+            opened.append(req)
+            resp = mock.MagicMock()
+            resp.read.return_value = b'{"choices":[]}'
+            cm = mock.MagicMock()
+            cm.__enter__.return_value = resp
+            return cm
+        with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
+            self.b._post_json("/v1/chat/completions", {"model": "x"})
+        self.assertEqual(opened[0].get_header("Authorization"), "Bearer sk-abc")
+
+    def test_bad_response_raises(self):
+        with mock.patch.object(self.b, "_post_json", return_value={}):
+            with self.assertRaises(RuntimeError):
+                self.b.generate("x", [])
+
+
+class TestSafetyGateway(unittest.TestCase):
+    """安全网关：safety_checker 复查（fail-closed）。"""
+
+    def test_checker_reject_blocks(self):
+        b = CloudApiBackend(endpoint="http://x/v1",
+                            safety_checker=lambda p: "脱敏" in p)
+        with mock.patch.object(b, "_post_json") as post:
+            out = b.generate("含敏感内容", [])
+        post.assert_not_called()   # 不调用上游
+        self.assertIn("拦截", out)
+
+    def test_checker_allow_passes(self):
+        b = CloudApiBackend(endpoint="http://x/v1",
+                            safety_checker=lambda p: "脱敏" in p)
+        fake = {"choices": [{"message": {"content": "ok"}}]}
+        with mock.patch.object(b, "_post_json", return_value=fake) as post:
+            out = b.generate("脱敏后的公开问题", [])
+        post.assert_called_once()
+        self.assertEqual(out, "ok")
+
+    def test_health_reports_safety(self):
+        b = CloudApiBackend(safety_checker=lambda p: True)
+        self.assertTrue(b.health()["safety_checker"])
+
+
+if __name__ == "__main__":
+    unittest.main()