From f657bfb10b823245bc2c4bd7be4460ede93fdd53 Mon Sep 17 00:00:00 2001 From: gouki Date: Sun, 2 Aug 2026 16:42:45 +0000 Subject: [PATCH] =?UTF-8?q?test:=20=E9=80=82=E9=85=8D=E5=99=A8=E5=8D=95?= =?UTF-8?q?=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=88=E6=A0=87=E5=87=86=E5=8C=96?= =?UTF-8?q?=E8=A7=A3=E6=9E=90+=E7=AD=BE=E5=90=8D=E7=AE=97=E6=B3=95+registr?= =?UTF-8?q?y=E8=B7=AF=E7=94=B1=EF=BC=8C11=E9=A1=B9=E9=80=9A=E8=BF=87?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + requirements-dev.txt | 2 + tests/test_adapters.py | 174 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 177 insertions(+) create mode 100644 requirements-dev.txt create mode 100644 tests/test_adapters.py diff --git a/.gitignore b/.gitignore index e2b74fc..6a964db 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ __pycache__/ data/ .env .DS_Store +.pytest_cache/ diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..f817dc5 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,2 @@ +# 开发/测试依赖（生产部署不需要） +pytest>=8.0 diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..55b6335 --- /dev/null +++ b/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)