iAOP/core/llm-gateway/tests/test_cloudapi.py

# -*- 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()