test: 适配器单元测试（标准化解析+签名算法+registry路由，11项通过）

.gitignore

@@ -4,3 +4,4 @@ __pycache__/
 data/
 .env
 .DS_Store
+.pytest_cache/

requirements-dev.txt

@@ -0,0 +1,2 @@
+# 开发/测试依赖（生产部署不需要）
+pytest>=8.0

tests/test_adapters.py

@@ -0,0 +1,174 @@
+"""云厂商适配器单元测试
+
+通过 mock 各平台 API 响应，验证适配器的标准化转换逻辑、签名算法正确性
+与注册表路由。确保用户填入真实凭证后，解析与签名链路可靠。
+运行：.venv/bin/pytest tests/ -v
+"""
+
+import base64
+from unittest.mock import patch
+
+from app.adapters import registry
+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)