vps-manager/tests/test_adapters.py

"""云厂商适配器单元测试

通过 mock 各平台 API 响应，验证适配器的标准化转换逻辑、签名算法正确性
与注册表路由。确保用户填入真实凭证后，解析与签名链路可靠。
运行：.venv/bin/pytest tests/ -v
"""

import base64
from unittest.mock import patch

from app.adapters import registry
from app.adapters.ai import DeepSeekAdapter, KimiAdapter, OpenAIAdapter
from app.adapters.aliyun import AliyunAdapter
from app.adapters.cloudflare import CloudflareAdapter
from app.adapters.digitalocean import DigitalOceanAdapter
from app.adapters.tencent import (
    BILLING_HOST,
    CVM_HOST,
    DNSPOD_HOST,
    TencentAdapter,
    _service_of,
)
from app.adapters.vultr import VultrAdapter


# ----------------------------- Vultr ----------------------------- #
def test_vultr_list_vps_normalization():
    adapter = VultrAdapter({"api_key": "x"})
    mock_resp = {
        "instances": [
            {
                "id": "vps-1",
                "label": "my-vps",
                "main_ip": "1.2.3.4",
                "region": "tokyo",
                "os": "Debian 12",
                "vcpu_count": 2,
                "ram": 4096,
                "disk": 80,
                "status": "active",
                "monthly_cost": 10.0,
            }
        ]
    }
    with patch.object(VultrAdapter, "_get", return_value=mock_resp):
        result = adapter.list_vps()
    assert len(result) == 1
    v = result[0]
    assert v.external_id == "vps-1"
    assert v.name == "my-vps"
    assert v.memory_gb == 4.0  # 4096 MB -> 4 GB
    assert v.ip_address == "1.2.3.4"
    assert v.cpu_cores == 2
    assert v.monthly_cost == 10.0
    assert v.status == "active"


def test_vultr_get_account():
    adapter = VultrAdapter({"api_key": "x"})
    with patch.object(
        VultrAdapter, "_get", return_value={"account": {"balance": -50.0, "pending_charges": 12.3}}
    ):
        acc = adapter.get_account()
    assert acc.balance == -50.0
    assert acc.pending_charges == 12.3
    assert acc.currency == "USD"


# --------------------------- DigitalOcean --------------------------- #
def test_do_list_vps_normalization():
    adapter = DigitalOceanAdapter({"api_key": "x"})
    mock_resp = {
        "droplets": [
            {
                "id": 123,
                "name": "droplet-1",
                "vcpus": 1,
                "memory": 2048,
                "disk": 25,
                "status": "active",
                "region": {"slug": "sgp1"},
                "image": {"distribution": "Debian", "name": "12 x64"},
                "networks": {"v4": [{"ip_address": "5.6.7.8"}]},
            }
        ]
    }
    with patch.object(DigitalOceanAdapter, "_get", return_value=mock_resp):
        result = adapter.list_vps()
    v = result[0]
    assert v.external_id == "123"
    assert v.memory_gb == 2.0  # 2048 MB -> 2 GB
    assert v.ip_address == "5.6.7.8"
    assert v.region == "sgp1"
    assert "Debian" in v.os


# --------------------------- Cloudflare --------------------------- #
def test_cloudflare_list_domains():
    adapter = CloudflareAdapter({"api_token": "x"})
    mock_resp = {"result": [{"id": "z1", "name": "example.com", "status": "active"}]}
    with patch.object(CloudflareAdapter, "_get", return_value=mock_resp):
        domains = adapter.list_domains()
    assert len(domains) == 1
    assert domains[0].domain_name == "example.com"
    assert domains[0].external_id == "z1"
    assert domains[0].registrar == "cloudflare"


def test_cloudflare_capabilities_no_vps():
    caps = CloudflareAdapter({"api_token": "x"}).capabilities()
    assert caps["list_domains"] is True
    assert caps["list_vps"] is False  # Cloudflare 无传统 VPS


# --------------------------- 阿里云签名 --------------------------- #
def test_aliyun_sign_deterministic_and_valid():
    adapter = AliyunAdapter({"access_key_id": "id", "access_key_secret": "secret"})
    params = {
        "Action": "DescribeRegions",
        "RegionId": "cn-hangzhou",
        "Timestamp": "2024-01-01T00:00:00Z",
        "SignatureNonce": "fixed-nonce",
    }
    sig1 = adapter._sign(params)
    sig2 = adapter._sign(params)
    assert sig1 == sig2  # 确定性
    base64.b64decode(sig1)  # 合法 base64（HMAC-SHA1 -> 28 字符 base64）
    assert len(sig1) == 28


def test_aliyun_sign_sensitive_to_secret():
    params = {"Action": "DescribeRegions", "Timestamp": "2024-01-01T00:00:00Z", "SignatureNonce": "n"}
    a1 = AliyunAdapter({"access_key_id": "id", "access_key_secret": "secret1"})
    a2 = AliyunAdapter({"access_key_id": "id", "access_key_secret": "secret2"})
    assert a1._sign(params) != a2._sign(params)


# --------------------------- 腾讯云签名 --------------------------- #
def test_tencent_authorization_format():
    adapter = TencentAdapter({"secret_id": "SID", "secret_key": "SKEY"})
    auth = adapter._build_authorization(
        "DescribeInstances", "{}", 1704067200, "2024-01-01", CVM_HOST, "cvm"
    )
    assert auth.startswith("TC3-HMAC-SHA256 Credential=SID/")
    assert "2024-01-01/cvm/tc3_request" in auth
    assert "SignedHeaders=content-type;host;x-tc-action" in auth
    assert "Signature=" in auth
    # 签名为 64 位十六进制（HMAC-SHA256）
    signature = auth.split("Signature=")[1]
    assert len(signature) == 64
    int(signature, 16)


def test_tencent_sign_sensitive_to_payload():
    adapter = TencentAdapter({"secret_id": "SID", "secret_key": "SKEY"})
    a1 = adapter._build_authorization(
        "DescribeInstances", '{"Limit":1}', 1704067200, "2024-01-01", CVM_HOST, "cvm"
    )
    a2 = adapter._build_authorization(
        "DescribeInstances", '{"Limit":2}', 1704067200, "2024-01-01", CVM_HOST, "cvm"
    )
    assert a1 != a2


def test_tencent_credential_scope_follows_service():
    """多产品复用同一套签名：credential scope 与 host 中的 canonical host 必须随 service 变化

    若 scope 写死成 cvm，调用 billing/domain/dnspod 会全部返回 AuthFailure.SignatureFailure。
    """
    adapter = TencentAdapter({"secret_id": "SID", "secret_key": "SKEY"})
    cvm = adapter._build_authorization(
        "DescribeInstances", "{}", 1704067200, "2024-01-01", CVM_HOST, "cvm"
    )
    billing = adapter._build_authorization(
        "DescribeAccountBalance", "{}", 1704067200, "2024-01-01", BILLING_HOST, "billing"
    )
    assert "2024-01-01/cvm/tc3_request" in cvm
    assert "2024-01-01/billing/tc3_request" in billing
    assert cvm != billing


def test_tencent_service_of_host():
    assert _service_of(CVM_HOST) == "cvm"
    assert _service_of(BILLING_HOST) == "billing"
    assert _service_of(DNSPOD_HOST) == "dnspod"


# --------------------------- 注册表路由 --------------------------- #
def test_registry_supported_types():
    for t in [
        "vultr-api",
        "do-api",
        "cloudflare-api",
        "aliyun-sdk",
        "alibabacloud-sdk",
        "tencent-sdk",
        "tencent-intl-sdk",
    ]:
        assert registry.is_supported(t), f"{t} 应被支持"
    assert not registry.is_supported("nonexistent-sdk")


def test_registry_get_adapter_routing():
    assert isinstance(registry.get_adapter("vultr-api", {}), VultrAdapter)
    assert isinstance(registry.get_adapter("do-api", {}), DigitalOceanAdapter)
    assert isinstance(registry.get_adapter("cloudflare-api", {}), CloudflareAdapter)
    assert isinstance(registry.get_adapter("aliyun-sdk", {}), AliyunAdapter)
    assert isinstance(registry.get_adapter("alibabacloud-sdk", {}), AliyunAdapter)
    assert isinstance(registry.get_adapter("tencent-sdk", {}), TencentAdapter)
    assert isinstance(registry.get_adapter("tencent-intl-sdk", {}), TencentAdapter)


# --------------------------- AI 平台适配器 --------------------------- #
def test_deepseek_balance_normalization():
    adapter = DeepSeekAdapter({"api_key": "x"})
    mock = {"data": {"total_balance": "88.50", "granted_balance": "10.00"}}
    with patch.object(DeepSeekAdapter, "_get", return_value=mock):
        acc = adapter.get_account()
    assert acc.balance == 88.5  # 字符串 -> float
    assert acc.currency == "CNY"


def test_deepseek_test_connection_includes_balance():
    adapter = DeepSeekAdapter({"api_key": "x"})
    with patch.object(DeepSeekAdapter, "_get", return_value={"data": {"total_balance": "88.50"}}):
        result = adapter.test_connection()
    assert result["ok"] is True
    assert "88.5" in result["message"]


def test_kimi_balance_normalization():
    adapter = KimiAdapter({"api_key": "x"})
    with patch.object(KimiAdapter, "_get", return_value={"data": {"available_balance": 123.45}}):
        acc = adapter.get_account()
    assert acc.balance == 123.45
    assert acc.currency == "CNY"


def test_openai_test_connection_models():
    adapter = OpenAIAdapter({"api_key": "x"})
    mock = {"data": [{"id": "gpt-4"}, {"id": "gpt-3.5-turbo"}]}
    with patch.object(OpenAIAdapter, "_get", return_value=mock):
        result = adapter.test_connection()
    assert result["ok"] is True
    assert "2" in result["message"]


def test_ai_capabilities_account_only():
    caps = DeepSeekAdapter({"api_key": "x"}).capabilities()
    assert caps["get_account"] is True
    assert caps["list_vps"] is False
    assert caps["list_domains"] is False


def test_registry_supports_ai_types():
    for t in ["deepseek-api", "moonshot-api", "openai-api", "minimax-api"]:
        assert registry.is_supported(t), f"{t} 应被支持"


# --------------------------- 分页处理 --------------------------- #
def test_vultr_pagination_cursor():
    adapter = VultrAdapter({"api_key": "x"})
    page1 = {"instances": [{"id": "a", "label": "a", "ram": 1024}], "meta": {"links": {"next": "cursor2"}}}
    page2 = {"instances": [{"id": "b", "label": "b", "ram": 2048}], "meta": {"links": {"next": ""}}}
    with patch.object(VultrAdapter, "_get", side_effect=[page1, page2]):
        result = adapter.list_vps()
    assert len(result) == 2
    assert {v.external_id for v in result} == {"a", "b"}


def test_do_pagination_page():
    adapter = DigitalOceanAdapter({"api_key": "x"})
    page1 = {"droplets": [{"id": 1, "name": "d1", "memory": 1024, "networks": {"v4": []}}], "links": {"pages": {"next": "x"}}}
    page2 = {"droplets": [{"id": 2, "name": "d2", "memory": 1024, "networks": {"v4": []}}], "links": {}}
    with patch.object(DigitalOceanAdapter, "_get", side_effect=[page1, page2]):
        result = adapter.list_vps()
    assert len(result) == 2
    assert {v.external_id for v in result} == {"1", "2"}


def test_cloudflare_pagination_result_info():
    adapter = CloudflareAdapter({"api_token": "x"})
    page1 = {"result": [{"id": "z1", "name": "a.com", "status": "active"}], "result_info": {"total_pages": 2, "page": 1}}
    page2 = {"result": [{"id": "z2", "name": "b.com", "status": "active"}], "result_info": {"total_pages": 2, "page": 2}}
    with patch.object(CloudflareAdapter, "_registrar_expiry_map", return_value={}), \
         patch.object(CloudflareAdapter, "_get", side_effect=[page1, page2]):
        result = adapter.list_domains()
    assert len(result) == 2
    assert {d.domain_name for d in result} == {"a.com", "b.com"}