vps-manager/tests/test_provider_alignment.py

"""平台注册表对齐测试

覆盖两处回归：

1. `_align_preset_providers`：seed_providers 只插入缺失 slug、不更新已有行，
   所以代码侧的能力变化必须靠迁移落地。迁移必须是「增量」的——线上 aliyun-intl
   的 services 被用户手工加过 ai_agent/other，按种子值整体覆盖会静默丢用户配置。
2. 种子与注册表的一致性：任何预设平台都不得声明未实现的 sdk_type。
   声明了前端就会显示「API 对接: xxx」，用户建好账号点同步才撞
   「暂不支持的 SDK 类型」——声明支持却不支持，比明说不支持更糟。

运行：.venv/bin/pytest tests/test_provider_alignment.py -v
"""

import sqlalchemy as sa
import pytest
from sqlmodel import Session, SQLModel, create_engine, select

import app.models.asset  # noqa: F401
import app.models.credential  # noqa: F401
import app.models.monitor  # noqa: F401
import app.models.provider  # noqa: F401
import app.models.ssl  # noqa: F401
from app.adapters import registry
from app.core.seed import PRESET_PROVIDERS, seed_providers
from app.database import _align_preset_providers
from app.models.provider import Provider

#: 线上 GET /api/providers 抓下来的真实状态（2026-10-11），
#: 用它做 fixture 才能保证迁移验证的是生产形态而不是理想形态
LIVE_PROVIDERS = [
    ("agnes", None, "ai_agent"),
    ("deepseek", "deepseek-api", "ai_agent"),
    ("kimi", "moonshot-api", "ai_agent"),
    ("minimax", "minimax-api", "ai_agent"),
    ("openai", "openai-api", "ai_agent"),
    ("cloudflare", "cloudflare-api", "cloudflare,dns,ssl_cert,cdn"),
    ("dynadot", "dynadot-api", "domain,dns"),
    ("namesilo", "namesilo-api", "domain,dns"),
    ("xinwang", None, "domain,dns"),
    ("aliyun", "aliyun-sdk", "vps,domain,ssl_cert,cdn,dns"),
    ("aliyun-intl", "alibabacloud-sdk", "vps,domain,ssl_cert,cdn,dns,ai_agent,other"),
    ("cloudcone", "cloudcone-api", "vps"),
    ("digitalocean", "do-api", "vps,domain"),
    ("linode", "linode-api", "vps"),
    ("tencent", "tencent-sdk", "vps,domain,ssl_cert,cdn,dns"),
    ("tencent-intl", "tencent-intl-sdk", "vps,domain,ssl_cert,cdn,dns"),
    ("tothost", None, "vps"),
    ("vultr", "vultr-api", "vps"),
    ("zeabur", "zeabur-api", "vps,domain"),
]


@pytest.fixture()
def engine(tmp_path):
    eng = create_engine(
        f"sqlite:///{tmp_path / 'providers.db'}", connect_args={"check_same_thread": False}
    )
    SQLModel.metadata.create_all(eng)
    yield eng
    eng.dispose()


@pytest.fixture()
def live_db(engine):
    """复刻线上 providers 表"""
    with Session(engine) as s:
        for slug, sdk, svc in LIVE_PROVIDERS:
            s.add(Provider(slug=slug, name=slug, category="vps", services=svc, sdk_type=sdk))
        s.commit()
    return engine


def _snapshot(engine) -> dict:
    with engine.begin() as conn:
        rows = conn.execute(sa.text("SELECT slug, sdk_type, services FROM providers")).fetchall()
    return {r[0]: (r[1], r[2]) for r in rows}


def _run_migration(engine) -> None:
    with engine.begin() as conn:
        _align_preset_providers(conn)


# ----------------------------- 种子 / 注册表一致性 ----------------------------- #
def test_no_preset_declares_unimplemented_sdk_type():
    """预设平台不得声明未实现的 sdk_type（AWS/Linode/Namesilo 落地后的不变量）"""
    supported = set(registry.supported_types())
    bogus = [(p["slug"], p["sdk_type"]) for p in PRESET_PROVIDERS
             if p.get("sdk_type") and p["sdk_type"] not in supported]
    assert bogus == [], f"这些平台声明了未实现的适配器：{bogus}"


def test_new_adapters_are_registered():
    """本轮新增的三家必须有对应注册项"""
    for sdk_type in ("aws-api", "linode-api", "namesilo-api"):
        assert registry.is_supported(sdk_type), f"{sdk_type} 未注册"


def test_seed_providers_inserts_missing_slug(live_db):
    """线上没有 aws 行，seed_providers 必须补上"""
    assert "aws" not in _snapshot(live_db)
    with Session(live_db) as s:
        added = seed_providers(s)
    snap = _snapshot(live_db)
    assert added >= 1
    assert snap["aws"][0] == "aws-api"
    assert "vps" in snap["aws"][1]


def test_seed_providers_is_idempotent(live_db):
    with Session(live_db) as s:
        first = seed_providers(s)
        second = seed_providers(s)
    assert second == 0
    assert len(_snapshot(live_db)) == len(LIVE_PROVIDERS) + first


# ----------------------------- 迁移行为 ----------------------------- #
def test_migration_clears_bogus_sdk_type(live_db):
    """cloudcone/zeabur/dynadot 声明的适配器从未实现，必须清空"""
    _run_migration(live_db)
    snap = _snapshot(live_db)
    for slug in ("cloudcone", "zeabur", "dynadot"):
        assert snap[slug][0] is None, f"{slug}.sdk_type 未清空：{snap[slug][0]!r}"


def test_migration_keeps_valid_sdk_type(live_db):
    """有效 sdk_type 不能被误伤，否则同步能力全丢"""
    _run_migration(live_db)
    snap = _snapshot(live_db)
    for slug, expected in (
        ("aliyun", "aliyun-sdk"), ("tencent", "tencent-sdk"), ("vultr", "vultr-api"),
        ("digitalocean", "do-api"), ("linode", "linode-api"), ("namesilo", "namesilo-api"),
        ("cloudflare", "cloudflare-api"), ("openai", "openai-api"),
    ):
        assert snap[slug][0] == expected


def test_migration_preserves_user_customized_services(live_db):
    """aliyun-intl 被用户手工加过 ai_agent/other，覆盖式迁移会静默丢配置"""
    _run_migration(live_db)
    tags = set(_snapshot(live_db)["aliyun-intl"][1].split(","))
    assert {"ai_agent", "other"} <= tags


def test_migration_adds_missing_service_tags(live_db):
    """适配器新增了域名/DNS 能力，services 必须跟上，
    否则 modals.js 按 services 过滤时选不到这些平台"""
    _run_migration(live_db)
    snap = _snapshot(live_db)
    assert {"domain", "dns"} <= set(snap["vultr"][1].split(","))
    assert {"domain", "dns"} <= set(snap["linode"][1].split(","))
    assert "dns" in snap["digitalocean"][1].split(",")


def test_migration_ignores_user_created_providers(live_db):
    """tothost 不在种子里，一个字都不能动"""
    before = _snapshot(live_db)["tothost"]
    _run_migration(live_db)
    assert _snapshot(live_db)["tothost"] == before


def test_migration_is_idempotent(live_db):
    """迁移跑两次结果必须一致（update.sh 每分钟可能触发重启）"""
    _run_migration(live_db)
    once = _snapshot(live_db)
    _run_migration(live_db)
    _run_migration(live_db)
    assert _snapshot(live_db) == once


def test_migration_leaves_no_unimplemented_declaration(live_db):
    """终态不变量：迁移 + seed 后，库里不存在「声明但未实现」的 sdk_type"""
    _run_migration(live_db)
    with Session(live_db) as s:
        seed_providers(s)
    _run_migration(live_db)

    supported = set(registry.supported_types())
    snap = _snapshot(live_db)
    bad = [(slug, v[0]) for slug, v in snap.items() if v[0] and v[0] not in supported]
    assert bad == []


def test_migration_result_covers_all_seed_services(live_db):
    """终态：每个预设平台的 services 至少包含种子声明的标签"""
    _run_migration(live_db)
    snap = _snapshot(live_db)
    for data in PRESET_PROVIDERS:
        slug = data["slug"]
        if slug not in snap:
            continue
        seed_tags = {t.strip() for t in (data.get("services") or "").split(",") if t.strip()}
        cur_tags = {t.strip() for t in (snap[slug][1] or "").split(",") if t.strip()}
        assert seed_tags <= cur_tags, f"{slug} 缺标签 {sorted(seed_tags - cur_tags)}"