vps-manager/tests/test_sync_service.py

"""sync_service 核心同步流程测试

覆盖 sync_account 的三个关键场景：
1. 正常同步入库：适配器返回 VPS + 域名，验证资产正确写入数据库
2. (provider_id, external_id) 去重：同 provider_id + external_id 的资产再次同步时更新而非新建
3. 适配器异常容错：VPS 同步抛异常时不中断整体流程，域名仍可同步，错误记录在结果中

全程 tmp_path + create_engine 临时库，mock 适配器，免凭证免网络。

运行：.venv/bin/python -m pytest tests/test_sync_service.py -v
"""

from datetime import date, timedelta

import pytest
from cryptography.fernet import Fernet
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.base import (
    AccountInfo,
    BaseAdapter,
    NormalizedDomain,
    NormalizedVPS,
)
from app.core import crypto
from app.core.config import settings
from app.models.asset import Account, Asset, AssetStatus, AssetType, DomainDetail, VPSDetail
from app.models.provider import Provider
from app.services import sync_service


# ----------------------------- fixtures ----------------------------- #


@pytest.fixture()
def engine(tmp_path, monkeypatch):
    """临时资产库"""
    eng = create_engine(
        f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False}
    )
    SQLModel.metadata.create_all(eng)
    monkeypatch.setattr("app.database.metrics_engine", eng)
    yield eng
    eng.dispose()


@pytest.fixture()
def master_key(monkeypatch):
    """临时 Fernet key，不依赖 .env 里的真实 MASTER_KEY"""
    monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode())
    crypto._get_fernet.cache_clear()
    yield
    crypto._get_fernet.cache_clear()


# ----------------------------- mock adapter ----------------------------- #


class MockAdapter(BaseAdapter):
    """可配置返回值的测试适配器"""

    required_config = ["api_key"]

    def __init__(self, config=None, *, vps_list=None, domain_list=None, account_info=None,
                 vps_raises=None, domain_raises=None):
        super().__init__(config)
        self._vps_list = vps_list or []
        self._domain_list = domain_list or []
        self._account_info = account_info
        self._vps_raises = vps_raises
        self._domain_raises = domain_raises

    def test_connection(self):
        return {"ok": True, "message": "mock"}

    def list_vps(self):
        if self._vps_raises:
            raise self._vps_raises
        return self._vps_list

    def list_domains(self):
        if self._domain_raises:
            raise self._domain_raises
        return self._domain_list

    def get_account(self):
        if self._account_info:
            return self._account_info
        return AccountInfo(balance=100.0, currency="USD")

    def capabilities(self):
        return {"list_vps": True, "list_domains": True, "get_account": True}


def _seed_provider(session, slug="vultr", name="Vultr", sdk_type="vultr-api"):
    """在库中种入一个 Provider 并返回"""
    p = Provider(slug=slug, name=name, sdk_type=sdk_type, category="vps")
    session.add(p)
    session.commit()
    session.refresh(p)
    return p


def _seed_account(session, platform="vultr", name="test@example.com", api_config=None):
    """在库中种入一个 Account 并返回"""
    from app.core.crypto import encrypt
    config_enc = encrypt(api_config or '{"api_key": "fake-key"}')
    acc = Account(name=name, platform=platform, api_config_encrypted=config_enc)
    session.add(acc)
    session.commit()
    session.refresh(acc)
    return acc


def _patch_adapter(monkeypatch, adapter):
    """让 _build_adapter_for_account 返回固定的 mock 适配器"""
    def _fake_build(session, account):
        provider = session.exec(select(Provider).where(Provider.slug == account.platform)).first()
        return provider, adapter

    monkeypatch.setattr(sync_service, "_build_adapter_for_account", _fake_build)


# ----------------------------- 场景 1：正常同步入库 ----------------------------- #


def test_sync_account_normal路径_vps_and_domains_created(engine, master_key, monkeypatch):
    """适配器返回 2 台 VPS + 1 个域名，同步后全部正确入库"""
    today = date.today()
    vps_list = [
        NormalizedVPS(
            external_id="vps-001", name="web-server",
            ip_address="1.2.3.4", region="Tokyo", os="Debian 12",
            cpu_cores=2, memory_gb=4, disk_gb=80,
            status="active", monthly_cost=12.5, currency="USD",
            expiry_date=today + timedelta(days=30),
        ),
        NormalizedVPS(
            external_id="vps-002", name="db-server",
            ip_address="5.6.7.8", region="Singapore",
            status="active",
        ),
    ]
    domain_list = [
        NormalizedDomain(
            external_id="dom-001", domain_name="example.com",
            registrar="namesilo", expiry_date=today + timedelta(days=365),
            status="active",
        ),
    ]

    adapter = MockAdapter(config={"api_key": "fake"}, vps_list=vps_list, domain_list=domain_list)
    _patch_adapter(monkeypatch, adapter)

    with Session(engine) as s:
        provider = _seed_provider(s)
        account = _seed_account(s)

        result = sync_service.sync_account(s, account.id)

        # 验证返回结构
        assert result["provider"] == "vultr"
        assert result["vps"]["created"] == 2
        assert result["vps"]["updated"] == 0
        assert result["domains"]["created"] == 1
        assert result["domains"]["updated"] == 0
        assert "vps_error" not in result
        assert "domains_error" not in result
        assert result["last_synced_at"] is not None

        # 验证数据库中的资产
        all_assets = s.exec(select(Asset)).all()
        assert len(all_assets) == 3  # 2 VPS + 1 domain

        vps_assets = [a for a in all_assets if a.asset_type == AssetType.VPS]
        dom_assets = [a for a in all_assets if a.asset_type == AssetType.DOMAIN]
        assert len(vps_assets) == 2
        assert len(dom_assets) == 1

        # 验证 VPS 详情
        web = next(a for a in vps_assets if a.external_id == "vps-001")
        assert web.name == "web-server"
        assert web.status == AssetStatus.ACTIVE
        assert web.cost == 12.5
        assert web.account_id == account.id

        detail = s.exec(select(VPSDetail).where(VPSDetail.asset_id == web.id)).first()
        assert detail is not None
        assert detail.ip_address == "1.2.3.4"
        assert detail.region == "Tokyo"
        assert detail.cpu_cores == 2

        # 验证域名详情
        dom = dom_assets[0]
        assert dom.external_id == "dom-001"
        dom_detail = s.exec(select(DomainDetail).where(DomainDetail.asset_id == dom.id)).first()
        assert dom_detail.domain_name == "example.com"
        assert dom_detail.registrar == "namesilo"


# ----------------------------- 场景 2：去重逻辑 ----------------------------- #


def test_sync_account_去重_provider_id_external_id(engine, master_key, monkeypatch):
    """相同 (provider_id, external_id, asset_type) 的资产再次同步时更新而非新建"""
    today = date.today()

    # 第一次同步：创建 1 台 VPS
    first_vps = [
        NormalizedVPS(
            external_id="vps-001", name="web-server",
            ip_address="1.2.3.4", status="active",
            expiry_date=today + timedelta(days=30),
        ),
    ]
    adapter1 = MockAdapter(config={"api_key": "fake"}, vps_list=first_vps)
    _patch_adapter(monkeypatch, adapter1)

    with Session(engine) as s:
        provider = _seed_provider(s)
        account = _seed_account(s)
        result1 = sync_service.sync_account(s, account.id)
        assert result1["vps"]["created"] == 1
        assert result1["vps"]["updated"] == 0

    # 第二次同步：同一 external_id，名称和 IP 变了
    second_vps = [
        NormalizedVPS(
            external_id="vps-001", name="web-server-renamed",
            ip_address="9.9.9.9", status="active",
            expiry_date=today + timedelta(days=60),
        ),
    ]
    adapter2 = MockAdapter(config={"api_key": "fake"}, vps_list=second_vps)
    _patch_adapter(monkeypatch, adapter2)

    with Session(engine) as s:
        result2 = sync_service.sync_account(s, account.id)
        assert result2["vps"]["created"] == 0
        assert result2["vps"]["updated"] == 1

        # 验证没有新建资产，而是更新了已有资产
        all_assets = s.exec(select(Asset)).all()
        assert len(all_assets) == 1  # 仍然只有 1 条

        asset = all_assets[0]
        assert asset.name == "web-server-renamed"
        assert asset.expiry_date == today + timedelta(days=60)

        # 验证 VPS 详情也更新了
        detail = s.exec(select(VPSDetail).where(VPSDetail.asset_id == asset.id)).first()
        assert detail.ip_address == "9.9.9.9"


def test_sync_keeps_account_attribution_when_multiple_subaccounts_see_same_asset(
    engine, master_key, monkeypatch
):
    """一条资源对多个子账号可见时，后同步的账号不能把归属改走。

    腾讯云 CAM 子账号常见场景：只读子账号也能看到全部实例。归属应在首次
    同步时确定，之后保持稳定，手工在资产里改过的归属也不能被同步冲掉。
    """
    vps_list = [NormalizedVPS(external_id="vps-shared", name="shared", status="active")]
    adapter = MockAdapter(config={"api_key": "fake"}, vps_list=vps_list)
    _patch_adapter(monkeypatch, adapter)

    with Session(engine) as s:
        _seed_provider(s)
        first = _seed_account(s, name="sub-a@example.com")
        second = _seed_account(s, name="sub-b@example.com")
        sync_service.sync_account(s, first.id)
        sync_service.sync_account(s, second.id)

        asset = s.exec(select(Asset).where(Asset.external_id == "vps-shared")).one()
        assert asset.account_id == first.id


# ----------------------------- 场景 3：适配器异常容错 ----------------------------- #

def test_sync_account_vps异常_域名仍可同步(engine, master_key, monkeypatch):
    """VPS 同步抛异常时，域名同步不受影响，错误记录在结果中"""
    today = date.today()
    domain_list = [
        NormalizedDomain(
            external_id="dom-001", domain_name="example.com",
            registrar="namesilo", expiry_date=today + timedelta(days=365),
            status="active",
        ),
    ]

    # VPS 同步会抛异常，但域名正常
    adapter = MockAdapter(
        config={"api_key": "fake"},
        vps_raises=RuntimeError("API rate limit exceeded"),
        domain_list=domain_list,
    )
    _patch_adapter(monkeypatch, adapter)

    with Session(engine) as s:
        provider = _seed_provider(s)
        account = _seed_account(s)

        result = sync_service.sync_account(s, account.id)

        # VPS 失败应记录错误
        assert "vps_error" in result
        assert "rate limit" in result["vps_error"].lower()
        assert "vps" not in result or result.get("vps") is None

        # 域名应正常同步
        assert result["domains"]["created"] == 1
        assert result["domains"]["updated"] == 0
        assert "domains_error" not in result

        # 数据库中只有域名，没有 VPS
        all_assets = s.exec(select(Asset)).all()
        assert len(all_assets) == 1
        assert all_assets[0].asset_type == AssetType.DOMAIN


def test_sync_account_domain异常_vps仍可同步(engine, master_key, monkeypatch):
    """域名同步抛异常时，VPS 同步不受影响（对称验证）"""
    vps_list = [
        NormalizedVPS(
            external_id="vps-001", name="web-server",
            ip_address="1.2.3.4", status="active",
        ),
    ]

    adapter = MockAdapter(
        config={"api_key": "fake"},
        vps_list=vps_list,
        domain_raises=ConnectionError("DNS resolution failed"),
    )
    _patch_adapter(monkeypatch, adapter)

    with Session(engine) as s:
        provider = _seed_provider(s)
        account = _seed_account(s)

        result = sync_service.sync_account(s, account.id)

        # VPS 正常
        assert result["vps"]["created"] == 1
        assert "vps_error" not in result

        # 域名失败
        assert "domains_error" in result
        assert "DNS resolution" in result["domains_error"]

        all_assets = s.exec(select(Asset)).all()
        assert len(all_assets) == 1
        assert all_assets[0].asset_type == AssetType.VPS