"""补齐平台适配器测试：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"]         # 区域不是密钥，明文显示