gouki
9e358fed6b
审计发现两处「已声明但实际用不了」的缺口： - 阿里云/腾讯云国际版与国内版共用适配器，region 只能靠代码默认值（cn-hangzhou / ap-guangzhou），账号弹窗里无处填写，海外账号查不到资源。新增 optional_config 机制，region 按可选明文渲染；留空仍走原默认值，不阻塞保存与同步。 - NameSilo 有 getAccountBalance 接口但未实现，平台卡片不显示余额。补充余额 与未出账消费（available_balance / unbilled_amount 分开报告）。 顺带修掉一个会丢凭证的隐患：编辑账号时前端只提交填写过的字段，后端原先整段 覆盖 api_config，只补 region 会把已存 AK/SK 抹掉；改为与已存 JSON 合并。 测试：449 passed（新增合并、余额、optional_config 用例），vitest 62 passed。
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""补齐平台适配器测试：CloudCone / Zeabur / Dynadot / Agnes / 新网
|
||
|
||
这五家此前 sdk_type 为空，平台卡片只能显示「未对接」。测试重点：
|
||
- 归一化后的字段直接决定台账展示，不能把 MB 当 GB、毫核当核；
|
||
- 到期日必须是 date，否则写库时整批 commit 抛 TypeError；
|
||
- 余额/到期拿不到时如实报告不支持，不能伪造 0；
|
||
- 新网响应是 URL 编码文本且 checksum 有严格拼接顺序，错一位就是 auth-failure。
|
||
|
||
运行：.venv/bin/python -m pytest tests/test_platform_adapters.py -v
|
||
"""
|
||
|
||
import hashlib
|
||
from datetime import date
|
||
from unittest.mock import patch
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from app.adapters import registry
|
||
from app.adapters.ai import AgnesAdapter
|
||
from app.adapters.cloudcone import CloudConeAdapter
|
||
from app.adapters.dynadot import DynadotAdapter, _epoch_ms_to_date
|
||
from app.adapters.xinnet import XinnetAdapter
|
||
from app.adapters.zeabur import ZeaburAdapter
|
||
|
||
|
||
# ----------------------------- CloudCone ----------------------------- #
|
||
def test_cloudcone_headers_use_documented_names():
|
||
adapter = CloudConeAdapter({"api_key": "KEY", "api_hash": "HASH"})
|
||
assert adapter._headers() == {"App-Secret": "KEY", "Hash": "HASH"}
|
||
|
||
|
||
def test_cloudcone_list_vps_normalization():
|
||
adapter = CloudConeAdapter({"api_key": "k", "api_hash": "h"})
|
||
payload = {
|
||
"status": 1,
|
||
"__data": {
|
||
"instances": [
|
||
{
|
||
"id": 236441,
|
||
"nodename": "S64",
|
||
"state": {"name": "active"},
|
||
"ips": "10.10.10.1,10.10.10.2",
|
||
"hostname": "web-01",
|
||
"distro": "ubuntu",
|
||
"ram": 3,
|
||
"cpu": 1,
|
||
"disk": 30,
|
||
},
|
||
{
|
||
"id": 2,
|
||
"nodename": "S53",
|
||
"state": {"name": "stopped"},
|
||
"ips": "10.10.10.3",
|
||
"hostname": "off-01",
|
||
"ram": 7.5,
|
||
"cpu": 8,
|
||
"disk": 150,
|
||
},
|
||
]
|
||
},
|
||
}
|
||
with patch.object(CloudConeAdapter, "_get", return_value=payload):
|
||
vps = adapter.list_vps()
|
||
|
||
assert [v.external_id for v in vps] == ["236441", "2"]
|
||
assert vps[0].name == "web-01"
|
||
assert vps[0].ip_address == "10.10.10.1" # 多 IP 取第一个
|
||
assert vps[0].memory_gb == 3 # ram 已经是 GB，不能再 /1024
|
||
assert vps[0].disk_gb == 30
|
||
assert vps[0].cpu_cores == 1
|
||
assert vps[0].status == "active"
|
||
assert vps[1].status == "stopped"
|
||
assert vps[0].charge_type == "prepaid"
|
||
|
||
|
||
def test_cloudcone_non_success_status_raises():
|
||
"""status=0 时响应体是 HTTP 200，只能在 _request 里翻译成异常"""
|
||
adapter = CloudConeAdapter({"api_key": "k", "api_hash": "h"})
|
||
resp = httpx.Response(
|
||
200,
|
||
request=httpx.Request("GET", "https://api.cloudcone.com/api/v2/compute/instances"),
|
||
json={"status": 0, "message": "Invalid hash"},
|
||
)
|
||
with patch("httpx.Client.request", return_value=resp):
|
||
with pytest.raises(RuntimeError) as ei:
|
||
adapter.list_vps()
|
||
assert "Invalid hash" in str(ei.value)
|
||
|
||
|
||
def test_cloudcone_power_actions_hit_documented_paths():
|
||
adapter = CloudConeAdapter({"api_key": "k", "api_hash": "h"})
|
||
with patch.object(CloudConeAdapter, "_post", return_value={"status": 1}) as m:
|
||
assert adapter.start_instance("42")["ok"] is True
|
||
assert m.call_args.args[0] == "/compute/42/boot"
|
||
assert adapter.stop_instance("42")["ok"] is True
|
||
assert m.call_args.args[0] == "/compute/42/shutdown"
|
||
assert adapter.reboot_instance("42")["ok"] is True
|
||
assert m.call_args.args[0] == "/compute/42/reboot"
|
||
|
||
|
||
def test_cloudcone_has_no_balance_or_renew():
|
||
caps = CloudConeAdapter({"api_key": "k", "api_hash": "h"}).capabilities()
|
||
assert caps["list_vps"] is True
|
||
assert caps["get_account"] is False # 没有公开账单接口
|
||
assert caps["renew_instance"] is False # 拿不到到期日就不提供续费
|
||
assert caps["start_instance"] is True
|
||
|
||
|
||
# ----------------------------- Zeabur ----------------------------- #
|
||
def test_zeabur_list_vps_normalization():
|
||
adapter = ZeaburAdapter({"api_token": "t"})
|
||
data = {
|
||
"servers": [
|
||
{
|
||
"_id": "s1",
|
||
"name": "hk-01",
|
||
"ip": "1.2.3.4",
|
||
"country": "HK",
|
||
"city": "Hong Kong",
|
||
"providerInfo": {"name": "Tencent"},
|
||
"hasK3s": False,
|
||
"status": {"isOnline": True, "totalCPU": 4000, "totalMemory": 8192, "totalDisk": 102400},
|
||
},
|
||
{
|
||
"_id": "s2",
|
||
"name": "legacy",
|
||
"ip": "5.6.7.8",
|
||
"hasK3s": None,
|
||
"status": {"isOnline": False, "totalCPU": 0, "totalMemory": 0, "totalDisk": 0},
|
||
},
|
||
]
|
||
}
|
||
with patch.object(ZeaburAdapter, "_graphql", return_value=data):
|
||
vps = adapter.list_vps()
|
||
|
||
assert [v.external_id for v in vps] == ["s1", "s2"]
|
||
assert vps[0].region == "Hong Kong, HK"
|
||
assert vps[0].cpu_cores == 4 # 毫核 -> 核
|
||
assert vps[0].memory_gb == 8.0 # MB -> GB
|
||
assert vps[0].disk_gb == 100
|
||
assert vps[0].status == "active"
|
||
assert vps[0].os == "Ubuntu"
|
||
assert vps[1].status == "stopped"
|
||
assert vps[1].os == "ZeaburOS" # hasK3s=None 是旧机器，不能标成纯净 Ubuntu
|
||
assert vps[1].cpu_cores is None # 未采集到规格时为 None，不能伪造 1 核
|
||
|
||
|
||
def test_zeabur_graphql_error_is_readable():
|
||
adapter = ZeaburAdapter({"api_token": "t"})
|
||
with patch.object(ZeaburAdapter, "_graphql", side_effect=RuntimeError("Zeabur GraphQL 错误：unauthorized")):
|
||
res = adapter.test_connection()
|
||
assert res["ok"] is False
|
||
assert "unauthorized" in res["message"]
|
||
|
||
|
||
def test_zeabur_reboot_mutation():
|
||
adapter = ZeaburAdapter({"api_token": "t"})
|
||
with patch.object(ZeaburAdapter, "_graphql", return_value={"rebootServer": True}) as m:
|
||
res = adapter.reboot_instance("s1")
|
||
assert res["ok"] is True
|
||
assert "rebootServer" in m.call_args.args[0]
|
||
assert m.call_args.args[1] == {"id": "s1"}
|
||
|
||
|
||
def test_zeabur_capabilities():
|
||
caps = ZeaburAdapter({"api_token": "t"}).capabilities()
|
||
assert caps["list_vps"] is True
|
||
assert caps["reboot_instance"] is True
|
||
assert caps["start_instance"] is False
|
||
assert caps["get_account"] is False
|
||
|
||
|
||
# ----------------------------- Dynadot ----------------------------- #
|
||
def test_dynadot_epoch_millis_to_date():
|
||
assert _epoch_ms_to_date("1798761600000") == date(2027, 1, 1)
|
||
assert _epoch_ms_to_date("1798761600") == date(2027, 1, 1)
|
||
assert _epoch_ms_to_date("2027-01-01") == date(2027, 1, 1)
|
||
assert _epoch_ms_to_date(None) is None
|
||
assert _epoch_ms_to_date("garbage") is None
|
||
|
||
|
||
def test_dynadot_list_domains_nested_shape():
|
||
adapter = DynadotAdapter({"api_key": "k"})
|
||
root = {
|
||
"MainDomains": [
|
||
{
|
||
"Domain": {
|
||
"Name": "example.com",
|
||
"Expiration": "1798761600000",
|
||
"Status": "active",
|
||
}
|
||
}
|
||
]
|
||
}
|
||
with patch.object(DynadotAdapter, "_call", return_value=root):
|
||
domains = adapter.list_domains()
|
||
assert len(domains) == 1
|
||
assert domains[0].domain_name == "example.com"
|
||
assert isinstance(domains[0].expiry_date, date) # 写库前必须是 date
|
||
assert domains[0].expiry_date == date(2027, 1, 1)
|
||
assert domains[0].registrar == "dynadot"
|
||
|
||
|
||
def test_dynadot_list_domains_flat_shape():
|
||
"""JSON 版本可能把字段摊平到 MainDomains 元素上"""
|
||
adapter = DynadotAdapter({"api_key": "k"})
|
||
root = {"MainDomains": [{"Name": "flat.com", "Expiration": "1798761600000", "Status": "active"}]}
|
||
with patch.object(DynadotAdapter, "_call", return_value=root):
|
||
assert adapter.list_domains()[0].domain_name == "flat.com"
|
||
|
||
|
||
def test_dynadot_balance_prefers_usd():
|
||
adapter = DynadotAdapter({"api_key": "k"})
|
||
root = {"BalanceList": [{"Currency": "CNY", "Amount": "100.00"}, {"Currency": "USD", "Amount": "300.00"}]}
|
||
with patch.object(DynadotAdapter, "_call", return_value=root):
|
||
acc = adapter.get_account()
|
||
assert acc.balance == 300.0
|
||
assert acc.currency == "USD"
|
||
|
||
|
||
def test_dynadot_error_code_raises():
|
||
adapter = DynadotAdapter({"api_key": "bad"})
|
||
resp = httpx.Response(
|
||
200,
|
||
request=httpx.Request("GET", "https://api.dynadot.com/api3.json"),
|
||
json={"ListDomainInfoResponse": {"ResponseCode": -1, "Status": "error", "Error": "invalid key"}},
|
||
)
|
||
client = httpx.Client()
|
||
with patch("httpx.Client.get", return_value=resp):
|
||
with pytest.raises(RuntimeError) as ei:
|
||
adapter._call("list_domain")
|
||
assert "invalid key" in str(ei.value)
|
||
client.close()
|
||
|
||
|
||
def test_dynadot_capabilities_domain_only():
|
||
caps = DynadotAdapter({"api_key": "k"}).capabilities()
|
||
assert caps["list_domains"] is True
|
||
assert caps["get_account"] is True
|
||
assert caps["list_vps"] is False
|
||
assert caps["list_dns_records"] is False # set_dns2 位置参数无法安全做单条增删改
|
||
|
||
|
||
# ----------------------------- Agnes ----------------------------- #
|
||
def test_agnes_test_connection_lists_models():
|
||
adapter = AgnesAdapter({"api_key": "k"})
|
||
with patch.object(AgnesAdapter, "_get", return_value={"data": [{"id": "agnes-3.0-flash"}]}):
|
||
res = adapter.test_connection()
|
||
assert res["ok"] is True
|
||
assert "1" in res["message"]
|
||
|
||
|
||
def test_agnes_has_no_balance_api():
|
||
caps = AgnesAdapter({"api_key": "k"}).capabilities()
|
||
assert caps["get_account"] is False
|
||
assert caps["list_vps"] is False
|
||
|
||
|
||
def test_agnes_base_url_override():
|
||
adapter = AgnesAdapter({"api_key": "k", "base_url": "https://api.agnes-ai.com/v1/"})
|
||
assert adapter._base_url() == "https://api.agnes-ai.com/v1"
|
||
|
||
|
||
# ----------------------------- 新网 ----------------------------- #
|
||
def test_xinnet_checksum_order():
|
||
"""checksum=MD5("GetProductKey"+client+password+dn+enc)，顺序错一位就是 auth-failure"""
|
||
adapter = XinnetAdapter({"client": "agent1", "password": "secret", "domains": "example.com"})
|
||
expected = hashlib.md5("GetProductKeyagent1secretexample.comE".encode()).hexdigest()
|
||
assert adapter._sign("GetProductKey", ["example.com", "E"]) == expected
|
||
|
||
|
||
def test_xinnet_domain_list_parsing_dedupes():
|
||
adapter = XinnetAdapter({"client": "c", "password": "p", "domains": "a.com, b.com；a.com\nc.com"})
|
||
assert adapter._domains() == ["a.com", "b.com", "c.com"]
|
||
|
||
|
||
def test_xinnet_list_domains_uses_expire_date():
|
||
adapter = XinnetAdapter({"client": "c", "password": "p", "domains": "a.com"})
|
||
with patch.object(XinnetAdapter, "_post", return_value={"name": "a.com", "key": "2028-03-01"}) as m:
|
||
domains = adapter.list_domains()
|
||
assert domains[0].expiry_date == date(2028, 3, 1)
|
||
assert domains[0].registrar == "xinnet"
|
||
params = m.call_args.args[1]
|
||
assert params["name"] == "a.com"
|
||
assert params["keyname"] == "ExpireDate"
|
||
|
||
|
||
def test_xinnet_error_ret_raises_with_reason():
|
||
"""新网失败时 HTTP 仍是 200，只靠响应体里的 ret/err 区分"""
|
||
adapter = XinnetAdapter({"client": "c", "password": "p", "domains": "a.com"})
|
||
resp = httpx.Response(
|
||
200,
|
||
request=httpx.Request("POST", "https://api.xinnet.com/domain/api.gb"),
|
||
text="ret=0&err=auth-failure",
|
||
)
|
||
with patch("httpx.Client.post", return_value=resp):
|
||
with pytest.raises(RuntimeError) as ei:
|
||
adapter.list_domains()
|
||
assert "auth-failure" in str(ei.value)
|
||
|
||
|
||
def test_xinnet_capabilities_domain_only():
|
||
caps = XinnetAdapter({"client": "c", "password": "p", "domains": "a.com"}).capabilities()
|
||
assert caps["list_domains"] is True
|
||
assert caps["list_vps"] is False
|
||
assert caps["list_dns_records"] is False
|
||
|
||
|
||
# ----------------------------- 注册表 ----------------------------- #
|
||
def test_new_platform_sdk_types_are_registered():
|
||
for sdk_type in (
|
||
"cloudcone-api",
|
||
"zeabur-api",
|
||
"dynadot-api",
|
||
"agnes-api",
|
||
"xinnet-api",
|
||
):
|
||
assert registry.is_supported(sdk_type), f"{sdk_type} 未注册"
|
||
|
||
|
||
def test_adapter_meta_exposes_plain_config():
|
||
meta = {m["sdk_types"][0]: m for m in registry.adapter_meta()}
|
||
assert meta["xinnet-api"]["required_config"] == ["client", "password", "domains"]
|
||
assert set(meta["xinnet-api"]["plain_config"]) == {"client", "domains"}
|
||
|
||
|
||
def test_adapter_meta_exposes_optional_region():
|
||
"""阿里云/腾讯云国际版必须能填 region，否则默认国内区域查不到资源"""
|
||
meta = {m["sdk_types"][0]: m for m in registry.adapter_meta()}
|
||
for sdk_type in ("aliyun-sdk", "tencent-sdk"):
|
||
item = meta[sdk_type]
|
||
assert item["optional_config"] == ["region"]
|
||
assert "region" not in item["required_config"] # 国内账号不强制填
|
||
assert "region" in item["plain_config"] # 区域不是密钥，明文显示
|