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。
739 lines
31 KiB
Python
739 lines
31 KiB
Python
"""新增适配器测试：AWS（SigV4）/ Linode / Namesilo
|
||
|
||
重点覆盖三类真实会炸的问题：
|
||
|
||
1. AWS SigV4 的派生签名密钥必须与官方测试向量逐字节一致——手写签名没有 SDK 兜底，
|
||
错一步就是满屏 SignatureDoesNotMatch，且无法从错误信息反推是哪一步错了。
|
||
2. credential scope / canonical host 必须随 service 变化（ec2 / route53 / route53domains
|
||
共用一套签名代码），写死成 ec2 会让 Route53 全线鉴权失败。
|
||
3. EC2 查询协议转 JSON 后的驼峰包装：单元素时 AWS 返回 dict 而非 list，
|
||
不做归一会直接 TypeError。
|
||
|
||
运行：.venv/bin/pytest tests/test_new_adapters.py -v
|
||
"""
|
||
|
||
from datetime import date
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from app.adapters.aws import AWSAdapter, _signing_key
|
||
from app.adapters.linode import LinodeAdapter
|
||
from app.adapters.namesilo import NamesiloAdapter
|
||
|
||
|
||
# ----------------------------- AWS SigV4 ----------------------------- #
|
||
def test_aws_signing_key_matches_official_vector():
|
||
"""AWS 官方文档测试向量（Signature Version 4 派生签名密钥示例）
|
||
|
||
这是手写 SigV4 唯一可靠的正确性锚点：向量对得上，说明
|
||
AWS4 前缀、四级 HMAC 链、aws4_request 终止符全部正确。
|
||
"""
|
||
key = _signing_key(
|
||
"wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
|
||
"20120215",
|
||
"us-east-1",
|
||
"iam",
|
||
)
|
||
assert key.hex() == "f4780e2d9f65fa895f9c67b32ce1baf0b0d8a43505a000a1a9e090d414db404d"
|
||
|
||
|
||
def test_aws_signing_key_sensitive_to_every_input():
|
||
"""任一分量变化都必须产出不同密钥（漏拼 region/service 是常见错法）"""
|
||
base = _signing_key("secret", "20240101", "us-east-1", "ec2")
|
||
assert base != _signing_key("secret2", "20240101", "us-east-1", "ec2")
|
||
assert base != _signing_key("secret", "20240102", "us-east-1", "ec2")
|
||
assert base != _signing_key("secret", "20240101", "eu-west-1", "ec2")
|
||
assert base != _signing_key("secret", "20240101", "us-east-1", "route53")
|
||
|
||
|
||
def _capture_signed_headers(adapter: AWSAdapter, **kwargs) -> dict:
|
||
"""拦掉 httpx，捕获 _signed_request 实际发出的请求头与 body"""
|
||
captured = {}
|
||
|
||
def fake_request(method, url, content=None, headers=None):
|
||
captured.update(method=method, url=url, content=content, headers=headers)
|
||
resp = MagicMock()
|
||
resp.content = b"{}"
|
||
resp.json.return_value = {}
|
||
resp.raise_for_status.return_value = None
|
||
return resp
|
||
|
||
client = MagicMock()
|
||
client.request.side_effect = fake_request
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = client
|
||
with patch("httpx.Client", return_value=cm):
|
||
adapter._signed_request(**kwargs)
|
||
return captured
|
||
|
||
|
||
def test_aws_authorization_header_format():
|
||
adapter = AWSAdapter({"access_key_id": "AKIDEXAMPLE", "secret_access_key": "SECRET"})
|
||
cap = _capture_signed_headers(
|
||
adapter, service="ec2", host="ec2.us-east-1.amazonaws.com",
|
||
region="us-east-1", method="POST", path="/", body="Action=DescribeRegions",
|
||
)
|
||
auth = cap["headers"]["Authorization"]
|
||
assert auth.startswith("AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/")
|
||
assert "/us-east-1/ec2/aws4_request" in auth
|
||
assert "SignedHeaders=content-type;host;x-amz-date" in auth
|
||
signature = auth.split("Signature=")[1]
|
||
assert len(signature) == 64
|
||
int(signature, 16)
|
||
|
||
|
||
def test_aws_credential_scope_follows_service():
|
||
"""ec2 / route53 / route53domains 共用签名代码，scope 必须随 service 变
|
||
|
||
写死成 ec2 的话 Route53 全部返回 SignatureDoesNotMatch。
|
||
"""
|
||
adapter = AWSAdapter({"access_key_id": "AKID", "secret_access_key": "SECRET"})
|
||
ec2 = _capture_signed_headers(
|
||
adapter, service="ec2", host="ec2.us-east-1.amazonaws.com",
|
||
region="us-east-1", method="POST", path="/", body="",
|
||
)["headers"]["Authorization"]
|
||
r53 = _capture_signed_headers(
|
||
adapter, service="route53", host="route53.amazonaws.com",
|
||
region="us-east-1", method="GET", path="/2013-04-01/hostedzone/Z1/rrset", body="",
|
||
)["headers"]["Authorization"]
|
||
assert "/us-east-1/ec2/aws4_request" in ec2
|
||
assert "/us-east-1/route53/aws4_request" in r53
|
||
assert ec2 != r53
|
||
|
||
|
||
def test_aws_route53_uses_rest_json_content_type():
|
||
"""Route53 的 /2013-04-01/* 是 REST-JSON（按路径路由，无 X-Amz-Target），
|
||
用 x-amz-json-1.1 会被当成 JSON 协议服务而报 MissingAuthenticationToken"""
|
||
adapter = AWSAdapter({"access_key_id": "AKID", "secret_access_key": "SECRET"})
|
||
with patch.object(AWSAdapter, "_signed_request", return_value={"ResourceRecordSets": []}) as m:
|
||
adapter.list_dns_records("Z123")
|
||
assert m.call_args.kwargs["content_type"] == "application/json"
|
||
assert m.call_args.kwargs["service"] == "route53"
|
||
|
||
|
||
def test_aws_host_header_excluded_from_send():
|
||
"""host 参与签名但不能手工发送，否则与 httpx 自动设置的值冲突"""
|
||
adapter = AWSAdapter({"access_key_id": "AKID", "secret_access_key": "SECRET"})
|
||
cap = _capture_signed_headers(
|
||
adapter, service="ec2", host="ec2.us-east-1.amazonaws.com",
|
||
region="us-east-1", method="POST", path="/", body="",
|
||
)
|
||
assert "host" not in cap["headers"]
|
||
# 但签名串里含 host（SignedHeaders 列出了它）
|
||
assert "host" in cap["headers"]["Authorization"]
|
||
|
||
|
||
def test_aws_signature_sensitive_to_payload():
|
||
adapter = AWSAdapter({"access_key_id": "AKID", "secret_access_key": "SECRET"})
|
||
a = _capture_signed_headers(
|
||
adapter, service="ec2", host="h", region="us-east-1",
|
||
method="POST", path="/", body="Action=A",
|
||
)["headers"]["Authorization"]
|
||
b = _capture_signed_headers(
|
||
adapter, service="ec2", host="h", region="us-east-1",
|
||
method="POST", path="/", body="Action=B",
|
||
)["headers"]["Authorization"]
|
||
assert a != b
|
||
|
||
|
||
def test_aws_region_defaults_and_override():
|
||
assert AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})._region() == "us-east-1"
|
||
assert AWSAdapter({"region": "ap-southeast-1"})._region() == "ap-southeast-1"
|
||
|
||
|
||
# ----------------------------- AWS EC2 响应解析 ----------------------------- #
|
||
def _ec2_response(instances, token=None):
|
||
return {
|
||
"reservationSet": {"item": [{"instancesSet": {"item": instances}}]},
|
||
**({"nextToken": token} if token else {}),
|
||
}
|
||
|
||
|
||
def test_aws_list_vps_unwraps_camelcase():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = _ec2_response([
|
||
{
|
||
"instanceId": "i-123",
|
||
"instanceState": {"name": "running"},
|
||
"ipAddress": "1.2.3.4",
|
||
"privateIpAddress": "10.0.0.5",
|
||
"placement": {"availabilityZone": "us-east-1a"},
|
||
"instanceLifecycle": "spot",
|
||
"tagSet": {"item": [{"key": "Name", "value": "web-01"}]},
|
||
}
|
||
])
|
||
with patch.object(AWSAdapter, "_ec2_call", return_value=payload):
|
||
vps = adapter.list_vps()
|
||
assert len(vps) == 1
|
||
v = vps[0]
|
||
assert v.external_id == "i-123"
|
||
assert v.name == "web-01" # 取 tagSet 的 Name，不是 instanceId
|
||
assert v.ip_address == "1.2.3.4" # 公网优先
|
||
assert v.region == "us-east-1a"
|
||
assert v.status == "active"
|
||
assert v.charge_type == "spot"
|
||
assert v.currency == "USD"
|
||
|
||
|
||
def test_aws_list_vps_single_reservation_is_dict():
|
||
"""EC2 查询协议转 JSON 后，单元素返回 dict 而不是 list——不归一会 TypeError"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = {
|
||
"reservationSet": {
|
||
"item": { # 注意：单个 reservation 时是 dict
|
||
"instancesSet": {
|
||
"item": { # 单个 instance 时也是 dict
|
||
"instanceId": "i-only",
|
||
"instanceState": {"name": "stopped"},
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
with patch.object(AWSAdapter, "_ec2_call", return_value=payload):
|
||
vps = adapter.list_vps()
|
||
assert len(vps) == 1
|
||
assert vps[0].external_id == "i-only"
|
||
assert vps[0].status == "stopped"
|
||
# 无 tagSet 时回退到 instanceId
|
||
assert vps[0].name == "i-only"
|
||
# 无公网 IP 时回退到私网
|
||
assert vps[0].ip_address is None
|
||
|
||
|
||
def test_aws_list_vps_filters_terminated():
|
||
"""terminated/shutting-down 不入台账，否则台账会越堆越多已销毁实例"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = _ec2_response([
|
||
{"instanceId": "i-live", "instanceState": {"name": "running"}},
|
||
{"instanceId": "i-dead", "instanceState": {"name": "terminated"}},
|
||
{"instanceId": "i-dying", "instanceState": {"name": "shutting-down"}},
|
||
{"instanceId": "i-pending", "instanceState": {"name": "pending"}},
|
||
])
|
||
with patch.object(AWSAdapter, "_ec2_call", return_value=payload):
|
||
vps = adapter.list_vps()
|
||
ids = {v.external_id for v in vps}
|
||
assert ids == {"i-live", "i-pending"}
|
||
|
||
|
||
def test_aws_list_vps_pagination():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
pages = [
|
||
_ec2_response([{"instanceId": "i-1", "instanceState": {"name": "running"}}], token="tok2"),
|
||
_ec2_response([{"instanceId": "i-2", "instanceState": {"name": "running"}}]),
|
||
]
|
||
with patch.object(AWSAdapter, "_ec2_call", side_effect=pages) as m:
|
||
vps = adapter.list_vps()
|
||
assert [v.external_id for v in vps] == ["i-1", "i-2"]
|
||
assert m.call_count == 2
|
||
# 第二页必须带上 NextToken
|
||
assert m.call_args_list[1].args[0]["NextToken"] == "tok2"
|
||
|
||
|
||
def test_aws_instance_action_sends_instance_id_list():
|
||
"""EC2 的启停接口参数是 InstanceId.1（列表编码），不是 InstanceId"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
with patch.object(AWSAdapter, "_ec2_call", return_value={}) as m:
|
||
assert adapter.start_instance("i-9")["ok"] is True
|
||
params = m.call_args.args[0]
|
||
assert params["Action"] == "StartInstances"
|
||
assert params["InstanceId.1"] == "i-9"
|
||
|
||
|
||
def test_aws_action_failure_returns_not_raises():
|
||
"""动作失败要回 ok=False + 原因，不能抛异常（上层按结果渲染提示）"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
with patch.object(AWSAdapter, "_ec2_call", side_effect=RuntimeError("boom")):
|
||
res = adapter.reboot_instance("i-1")
|
||
assert res["ok"] is False
|
||
assert "boom" in res["message"]
|
||
|
||
|
||
def test_aws_has_no_account_balance():
|
||
"""刻意不实现 get_account：AWS 没有「账户余额」概念，硬凑数字比报不支持更误导"""
|
||
caps = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"}).capabilities()
|
||
assert caps["get_account"] is False
|
||
assert caps["list_vps"] is True
|
||
assert caps["renew_instance"] is False # 按需计费无续费概念
|
||
|
||
|
||
# ----------------------------- AWS Route53 ----------------------------- #
|
||
def test_aws_hosted_zone_id_strips_prefix():
|
||
assert AWSAdapter._hosted_zone_id("/hostedzone/Z123ABC") == "Z123ABC"
|
||
assert AWSAdapter._hosted_zone_id("Z123ABC") == "Z123ABC"
|
||
|
||
|
||
def test_aws_epoch_to_date():
|
||
# 2027-01-01T00:00:00Z == 1798761600
|
||
assert AWSAdapter._epoch_to_date(1798761600) == date(2027, 1, 1)
|
||
assert AWSAdapter._epoch_to_date("1798761600") == date(2027, 1, 1)
|
||
assert AWSAdapter._epoch_to_date(None) is None
|
||
assert AWSAdapter._epoch_to_date("garbage") is None
|
||
|
||
|
||
def test_aws_list_domains_from_route53domains():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = {"Domains": [{"DomainName": "example.com", "Expiry": 1798761600}]}
|
||
with patch.object(AWSAdapter, "_r53domains_call", return_value=payload):
|
||
domains = adapter.list_domains()
|
||
assert len(domains) == 1
|
||
d = domains[0]
|
||
assert d.domain_name == "example.com"
|
||
assert isinstance(d.expiry_date, date) # 必须是 date，否则写库抛 TypeError 整批回滚
|
||
assert d.expiry_date == date(2027, 1, 1)
|
||
assert d.registrar == "aws-route53"
|
||
|
||
|
||
def test_aws_list_dns_records_normalization():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = {
|
||
"ResourceRecordSets": [
|
||
{
|
||
"Name": "www.example.com.", # Route53 总是带尾点
|
||
"Type": "A",
|
||
"TTL": 300,
|
||
"ResourceRecords": [{"Value": "1.2.3.4"}, {"Value": "5.6.7.8"}],
|
||
}
|
||
]
|
||
}
|
||
with patch.object(AWSAdapter, "_signed_request", return_value=payload):
|
||
records = adapter.list_dns_records("Z1")
|
||
assert len(records) == 1
|
||
r = records[0]
|
||
assert r.name == "www.example.com" # 尾点被剥掉
|
||
assert r.record_id == "www.example.com.|A" # record_id 用原始带尾点的 Name
|
||
assert r.content == "1.2.3.4, 5.6.7.8" # 多值合并
|
||
assert r.ttl == 300
|
||
|
||
|
||
def test_aws_dns_write_wraps_values_in_array():
|
||
"""Route53 的 ResourceRecords 必须是数组，单值也要包一层"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
with patch.object(AWSAdapter, "_signed_request", return_value={}) as m:
|
||
res = adapter.create_dns_record("Z1", {"name": "a.example.com", "type": "A", "content": "1.2.3.4", "ttl": 600})
|
||
assert res["ok"] is True
|
||
import json as _json
|
||
body = _json.loads(m.call_args.kwargs["body"])
|
||
rrs = body["Changes"][0]["ResourceRecordSet"]
|
||
assert rrs["ResourceRecords"] == [{"Value": "1.2.3.4"}]
|
||
assert rrs["TTL"] == 600
|
||
assert body["Changes"][0]["Action"] == "CREATE"
|
||
|
||
|
||
def test_aws_delete_requires_name_type_record_id():
|
||
"""Route53 删除必须提交完整记录集，record_id 缺 TYPE 无法构造"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
res = adapter.delete_dns_record("Z1", "no-pipe-here")
|
||
assert res["ok"] is False
|
||
|
||
|
||
def test_aws_delete_reads_existing_record_first():
|
||
"""DELETE 需要带上现存的值与 TTL，必须先读回记录"""
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = {
|
||
"ResourceRecordSets": [
|
||
{"Name": "a.example.com.", "Type": "A", "TTL": 300, "ResourceRecords": [{"Value": "1.2.3.4"}]}
|
||
]
|
||
}
|
||
with patch.object(AWSAdapter, "_signed_request", side_effect=[payload, {}]) as m:
|
||
res = adapter.delete_dns_record("Z1", "a.example.com.|A")
|
||
assert res["ok"] is True
|
||
import json as _json
|
||
body = _json.loads(m.call_args_list[1].kwargs["body"])
|
||
assert body["Changes"][0]["Action"] == "DELETE"
|
||
assert body["Changes"][0]["ResourceRecordSet"]["ResourceRecords"] == [{"Value": "1.2.3.4"}]
|
||
|
||
|
||
def test_aws_delete_missing_record():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
with patch.object(AWSAdapter, "_signed_request", return_value={"ResourceRecordSets": []}):
|
||
res = adapter.delete_dns_record("Z1", "gone.example.com.|A")
|
||
assert res["ok"] is False
|
||
assert "不存在" in res["message"]
|
||
|
||
|
||
def test_aws_test_connection_reports_region_count():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
payload = {"regionInfo": {"item": [{"regionName": "us-east-1"}, {"regionName": "eu-west-1"}]}}
|
||
with patch.object(AWSAdapter, "_ec2_call", return_value=payload):
|
||
res = adapter.test_connection()
|
||
assert res["ok"] is True
|
||
assert "2" in res["message"]
|
||
|
||
|
||
def test_aws_test_connection_failure_is_graceful():
|
||
adapter = AWSAdapter({"access_key_id": "a", "secret_access_key": "b"})
|
||
resp = httpx.Response(403, request=httpx.Request("POST", "https://ec2.us-east-1.amazonaws.com/"))
|
||
with patch.object(AWSAdapter, "_ec2_call", side_effect=httpx.HTTPStatusError("denied", request=resp.request, response=resp)):
|
||
res = adapter.test_connection()
|
||
assert res["ok"] is False
|
||
assert "403" in res["message"]
|
||
|
||
|
||
# ----------------------------- Linode ----------------------------- #
|
||
def test_linode_list_vps_normalization():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
payload = {
|
||
"pages": 1,
|
||
"data": [
|
||
{
|
||
"id": 12345,
|
||
"label": "prod-web",
|
||
"status": "running",
|
||
"region": "us-east",
|
||
"image": "linode/ubuntu22.04",
|
||
"ipv4": ["1.2.3.4", "10.0.0.1"],
|
||
"specs": {"memory": 4096, "vcpus": 2, "disk": 81920},
|
||
}
|
||
],
|
||
}
|
||
with patch.object(LinodeAdapter, "_get", return_value=payload):
|
||
vps = adapter.list_vps()
|
||
assert len(vps) == 1
|
||
v = vps[0]
|
||
assert v.external_id == "12345"
|
||
assert v.ip_address == "1.2.3.4" # 取第一个 IPv4
|
||
assert v.memory_gb == 4.0 # MB -> GB
|
||
assert v.cpu_cores == 2
|
||
assert v.disk_gb == 81920
|
||
assert v.status == "active"
|
||
assert v.charge_type == "postpaid" # 按小时计费，不会有到期日
|
||
assert v.expiry_date is None
|
||
|
||
|
||
def test_linode_status_mapping():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
for raw, expected in (("offline", "stopped"), ("shutting_down", "stopped"), ("booting", "active")):
|
||
payload = {"pages": 1, "data": [{"id": 1, "label": "n", "status": raw, "specs": {}}]}
|
||
with patch.object(LinodeAdapter, "_get", return_value=payload):
|
||
assert adapter.list_vps()[0].status == expected
|
||
|
||
|
||
def test_linode_negative_balance_is_credit():
|
||
"""Linode 的 balance 为负表示账户有预付余额，正数才是欠款——不要取绝对值"""
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
with patch.object(LinodeAdapter, "_get", return_value={"balance": -25.5, "email": "me@x.com"}):
|
||
acc = adapter.get_account()
|
||
assert acc.balance == -25.5
|
||
assert acc.currency == "USD"
|
||
|
||
|
||
def test_linode_pagination_uses_pages():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
pages = [
|
||
{"pages": 2, "data": [{"id": 1, "label": "a", "status": "running", "specs": {}}]},
|
||
{"pages": 2, "data": [{"id": 2, "label": "b", "status": "running", "specs": {}}]},
|
||
]
|
||
with patch.object(LinodeAdapter, "_get", side_effect=pages) as m:
|
||
vps = adapter.list_vps()
|
||
assert [v.external_id for v in vps] == ["1", "2"]
|
||
assert "page=2" in m.call_args_list[1].args[0]
|
||
|
||
|
||
def test_linode_domains_expiry_nullable():
|
||
"""仅 Linode 注册的域名有 expiry；纯 DNS 托管的为 null，不能炸"""
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
payload = {
|
||
"pages": 1,
|
||
"data": [
|
||
{"id": 1, "domain": "reg.com", "status": "active", "expiry": "2027-05-06T04:00:00"},
|
||
{"id": 2, "domain": "dns-only.com", "status": "active", "expiry": None},
|
||
],
|
||
}
|
||
with patch.object(LinodeAdapter, "_get", return_value=payload):
|
||
domains = adapter.list_domains()
|
||
assert domains[0].expiry_date == date(2027, 5, 6)
|
||
assert domains[1].expiry_date is None
|
||
|
||
|
||
def test_linode_error_message_reads_body_reason():
|
||
"""Linode 把原因放在 body.errors[0].reason，裸状态码对用户没用"""
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
resp = httpx.Response(
|
||
403,
|
||
request=httpx.Request("GET", "https://api.linode.com/v4/account"),
|
||
json={"errors": [{"reason": "Invalid token"}]},
|
||
)
|
||
err = httpx.HTTPStatusError("denied", request=resp.request, response=resp)
|
||
assert "Invalid token" in adapter._error_message(err)
|
||
|
||
|
||
def test_linode_error_message_survives_unparseable_body():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
resp = httpx.Response(
|
||
502,
|
||
request=httpx.Request("GET", "https://api.linode.com/v4/account"),
|
||
content=b"<html>bad gateway</html>",
|
||
)
|
||
err = httpx.HTTPStatusError("boom", request=resp.request, response=resp)
|
||
assert adapter._error_message(err) == "HTTP 502"
|
||
|
||
|
||
def test_linode_capabilities_no_renew():
|
||
"""按小时计费无「续费」概念，不能声明 renew_instance"""
|
||
caps = LinodeAdapter({"api_key": "x"}).capabilities()
|
||
assert caps["renew_instance"] is False
|
||
assert caps["start_instance"] is True
|
||
assert caps["list_dns_records"] is True
|
||
assert caps["get_account"] is True
|
||
|
||
|
||
def test_linode_instance_actions_hit_right_verbs():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
for fn, verb in (("start_instance", "boot"), ("stop_instance", "shutdown"), ("reboot_instance", "reboot")):
|
||
with patch.object(LinodeAdapter, "_request", return_value={}) as m:
|
||
assert getattr(adapter, fn)("999")["ok"] is True
|
||
assert m.call_args.args[1] == f"/linode/instances/999/{verb}"
|
||
|
||
|
||
def test_linode_record_body_mapping():
|
||
"""Linode 用 target/ttl_sec，不是 content/ttl"""
|
||
body = LinodeAdapter._record_body({"type": "MX", "name": "mail", "content": "mx.example.com", "ttl": 600, "priority": 10})
|
||
assert body == {"type": "MX", "name": "mail", "target": "mx.example.com", "ttl_sec": 600, "priority": 10}
|
||
# priority 为 None 时不应出现在 body 里（Linode 会校验类型）
|
||
assert "priority" not in LinodeAdapter._record_body({"type": "A", "name": "a", "content": "1.2.3.4"})
|
||
|
||
|
||
def test_linode_dns_crud_paths():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
with patch.object(LinodeAdapter, "_request", return_value={"id": 555}) as m:
|
||
assert adapter.create_dns_record("7", {"type": "A", "name": "a", "content": "1.2.3.4"})["record_id"] == "555"
|
||
assert m.call_args.args[0] == "POST"
|
||
assert m.call_args.args[1] == "/domains/7/records"
|
||
|
||
assert adapter.update_dns_record("7", "555", {"type": "A", "name": "a", "content": "1.2.3.5"})["ok"] is True
|
||
assert m.call_args.args[0] == "PUT"
|
||
assert m.call_args.args[1] == "/domains/7/records/555"
|
||
|
||
assert adapter.delete_dns_record("7", "555")["ok"] is True
|
||
assert m.call_args.args[0] == "DELETE"
|
||
|
||
|
||
def test_linode_list_dns_records_maps_target():
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
payload = {"pages": 1, "data": [{"id": 1, "type": "A", "name": "www", "target": "1.2.3.4", "ttl_sec": 300}]}
|
||
with patch.object(LinodeAdapter, "_get", return_value=payload):
|
||
records = adapter.list_dns_records("7")
|
||
assert records[0].content == "1.2.3.4"
|
||
assert records[0].ttl == 300
|
||
assert records[0].zone_id == "7"
|
||
|
||
|
||
def test_linode_empty_body_handled():
|
||
"""POST 动作类接口返回 200 + 空 body，直接 .json() 会炸"""
|
||
adapter = LinodeAdapter({"api_key": "x"})
|
||
resp = MagicMock()
|
||
resp.content = b""
|
||
resp.raise_for_status.return_value = None
|
||
client = MagicMock()
|
||
client.request.return_value = resp
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = client
|
||
with patch("httpx.Client", return_value=cm):
|
||
assert adapter.start_instance("1")["ok"] is True
|
||
|
||
|
||
# ----------------------------- Namesilo ----------------------------- #
|
||
def _namesilo_reply(reply: dict) -> dict:
|
||
return {"request": {"command": "x"}, "reply": reply}
|
||
|
||
|
||
def test_namesilo_success_code_300():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
payload = _namesilo_reply({"code": 300, "detail": "success", "domains": []})
|
||
resp = MagicMock()
|
||
resp.json.return_value = payload
|
||
resp.raise_for_status.return_value = None
|
||
client = MagicMock()
|
||
client.get.return_value = resp
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = client
|
||
with patch("httpx.Client", return_value=cm):
|
||
assert adapter._call("listDomains")["code"] == 300
|
||
|
||
|
||
def test_namesilo_string_code_accepted():
|
||
"""Namesilo 的 code 可能是 str，直接 != 300 会把成功误判成失败"""
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
resp = MagicMock()
|
||
resp.json.return_value = _namesilo_reply({"code": "300", "detail": "success"})
|
||
resp.raise_for_status.return_value = None
|
||
client = MagicMock()
|
||
client.get.return_value = resp
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = client
|
||
with patch("httpx.Client", return_value=cm):
|
||
assert adapter._call("checkApiKey")["code"] == "300"
|
||
|
||
|
||
def test_namesilo_failure_raises_with_detail():
|
||
"""301 未授权 / 400 参数错必须抛出并带上 detail，否则用户只看到一个空失败"""
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
resp = MagicMock()
|
||
resp.json.return_value = _namesilo_reply({"code": 301, "detail": "Invalid or missing API key"})
|
||
resp.raise_for_status.return_value = None
|
||
client = MagicMock()
|
||
client.get.return_value = resp
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = client
|
||
with patch("httpx.Client", return_value=cm):
|
||
with pytest.raises(RuntimeError) as ei:
|
||
adapter._call("listDomains")
|
||
assert "301" in str(ei.value)
|
||
assert "Invalid or missing API key" in str(ei.value)
|
||
|
||
|
||
def test_namesilo_api_key_sent_as_query_param():
|
||
adapter = NamesiloAdapter({"api_key": "SECRETKEY"})
|
||
resp = MagicMock()
|
||
resp.json.return_value = _namesilo_reply({"code": 300, "username": "me"})
|
||
resp.raise_for_status.return_value = None
|
||
client = MagicMock()
|
||
client.get.return_value = resp
|
||
cm = MagicMock()
|
||
cm.__enter__.return_value = client
|
||
with patch("httpx.Client", return_value=cm):
|
||
adapter.test_connection()
|
||
query = client.get.call_args.kwargs["params"]
|
||
assert query["key"] == "SECRETKEY"
|
||
assert query["version"] == "1"
|
||
assert query["type"] == "json"
|
||
|
||
|
||
def test_namesilo_test_connection_reports_username():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", return_value={"code": 300, "username": "neatstudio"}):
|
||
res = adapter.test_connection()
|
||
assert res["ok"] is True
|
||
assert "neatstudio" in res["message"]
|
||
|
||
|
||
def test_namesilo_list_domains_expiry():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
reply = {"code": 300, "domains": [{"domain": "example.com", "expires": "2027-01-01"}]}
|
||
with patch.object(NamesiloAdapter, "_call", return_value=reply):
|
||
domains = adapter.list_domains()
|
||
assert len(domains) == 1
|
||
assert isinstance(domains[0].expiry_date, date)
|
||
assert domains[0].expiry_date == date(2027, 1, 1)
|
||
assert domains[0].registrar == "namesilo"
|
||
|
||
|
||
def test_namesilo_unparseable_expiry_is_none():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
reply = {"code": 300, "domains": [{"domain": "x.com", "expires": "not-a-date"}]}
|
||
with patch.object(NamesiloAdapter, "_call", return_value=reply):
|
||
assert adapter.list_domains()[0].expiry_date is None
|
||
|
||
|
||
def test_namesilo_balance_separates_available_and_unbilled():
|
||
"""可用余额与未出账消费必须分开报告，合并会把可花金额说多"""
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
reply = {
|
||
"code": 300,
|
||
"available_balance": "12.34",
|
||
"unbilled_amount": "1.99",
|
||
}
|
||
with patch.object(NamesiloAdapter, "_call", return_value=reply):
|
||
acc = adapter.get_account()
|
||
assert acc.balance == 12.34
|
||
assert acc.pending_charges == 1.99
|
||
assert acc.currency == "USD"
|
||
|
||
|
||
def test_namesilo_balance_unparseable_is_none():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", return_value={"code": 300, "available_balance": "N/A"}):
|
||
acc = adapter.get_account()
|
||
assert acc.balance is None
|
||
|
||
|
||
def test_namesilo_fqdn_host_roundtrip():
|
||
"""Namesilo 存子域（空/@ 表示裸域），台账展示需要完整域名"""
|
||
assert NamesiloAdapter._fqdn("", "example.com") == "example.com"
|
||
assert NamesiloAdapter._fqdn("@", "example.com") == "example.com"
|
||
assert NamesiloAdapter._fqdn("www", "example.com") == "www.example.com"
|
||
|
||
assert NamesiloAdapter._host("example.com", "example.com") == ""
|
||
assert NamesiloAdapter._host("www.example.com", "example.com") == "www"
|
||
assert NamesiloAdapter._host("", "example.com") == ""
|
||
|
||
|
||
def test_namesilo_dns_record_normalization():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
reply = {
|
||
"code": 300,
|
||
"resource_record": [
|
||
{"record_id": 111, "host": "www", "type": "A", "value": "1.2.3.4", "ttl": 3600},
|
||
{"record_id": 222, "host": "", "type": "MX", "value": "mail.example.com", "ttl": 3600, "distance": 10},
|
||
],
|
||
}
|
||
with patch.object(NamesiloAdapter, "_call", return_value=reply):
|
||
records = adapter.list_dns_records("example.com")
|
||
assert records[0].name == "www.example.com"
|
||
assert records[0].record_id == "111"
|
||
assert records[0].ttl == 3600
|
||
assert records[1].name == "example.com" # 裸域
|
||
assert records[1].priority == 10
|
||
|
||
|
||
def test_namesilo_create_omits_rr_id():
|
||
"""dnsRecordUpdate 同时承担新增与修改：新增绝不能带 rr_id，否则改到别的记录上"""
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", return_value={"code": 300}) as m:
|
||
res = adapter.create_dns_record("example.com", {"type": "A", "name": "www.example.com", "content": "1.2.3.4", "ttl": 600})
|
||
assert res["ok"] is True
|
||
assert m.call_args.args[0] == "dnsRecordUpdate"
|
||
params = m.call_args.args[1]
|
||
assert "rr_id" not in params
|
||
assert params["rr_host"] == "www"
|
||
assert params["rr_type"] == "A"
|
||
assert params["rr_value"] == "1.2.3.4"
|
||
assert params["rr_ttl"] == 600
|
||
|
||
|
||
def test_namesilo_update_sends_rr_id():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", return_value={"code": 300}) as m:
|
||
res = adapter.update_dns_record("example.com", "111", {"type": "A", "name": "example.com", "content": "9.9.9.9"})
|
||
assert res["ok"] is True
|
||
params = m.call_args.args[1]
|
||
assert params["rr_id"] == "111"
|
||
assert params["rr_host"] == "" # 裸域 -> 空串
|
||
assert params["rr_ttl"] == 3600 # 未给 ttl 时的默认值
|
||
|
||
|
||
def test_namesilo_mx_sends_distance():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", return_value={"code": 300}) as m:
|
||
adapter.create_dns_record("example.com", {"type": "MX", "name": "example.com", "content": "mx.x.com", "priority": 10})
|
||
assert m.call_args.args[1]["rr_distance"] == 10
|
||
|
||
|
||
def test_namesilo_delete_passes_rr_id():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", return_value={"code": 300}) as m:
|
||
assert adapter.delete_dns_record("example.com", "111")["ok"] is True
|
||
assert m.call_args.args[0] == "dnsRecordDelete"
|
||
assert m.call_args.args[1] == {"domain": "example.com", "rr_id": "111"}
|
||
|
||
|
||
def test_namesilo_write_failure_returns_not_raises():
|
||
adapter = NamesiloAdapter({"api_key": "k"})
|
||
with patch.object(NamesiloAdapter, "_call", side_effect=RuntimeError("code=301: bad key")):
|
||
assert adapter.create_dns_record("x.com", {"type": "A", "name": "x.com", "content": "1.1.1.1"})["ok"] is False
|
||
assert adapter.update_dns_record("x.com", "1", {"type": "A", "name": "x.com", "content": "1.1.1.1"})["ok"] is False
|
||
assert adapter.delete_dns_record("x.com", "1")["ok"] is False
|
||
|
||
|
||
def test_namesilo_capabilities_domain_only():
|
||
"""纯域名注册商：无 VPS、无实例动作；余额走 getAccountBalance"""
|
||
caps = NamesiloAdapter({"api_key": "k"}).capabilities()
|
||
assert caps["list_domains"] is True
|
||
assert caps["list_dns_records"] is True
|
||
assert caps["get_account"] is True
|
||
assert caps["list_vps"] is False
|
||
assert caps["start_instance"] is False
|