255 lines
9.9 KiB
Python
255 lines
9.9 KiB
Python
"""云厂商适配器单元测试
|
||
|
||
通过 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 TencentAdapter
|
||
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")
|
||
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")
|
||
a2 = adapter._build_authorization("DescribeInstances", '{"Limit":2}', 1704067200, "2024-01-01")
|
||
assert a1 != a2
|
||
|
||
|
||
# --------------------------- 注册表路由 --------------------------- #
|
||
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"}
|