vps-manager/tests/test_domain_expiry.py

"""域名到期日链路与同步容错测试

覆盖三处回归：

1. 适配器产出的 expiry_date 必须是 date（NormalizedDomain 契约）。
   直接塞 ISO 字符串会在写库时抛 TypeError，而 _sync_domains 的 commit 在循环外，
   一个域名解析失败就会让整批域名同步全量回滚。
2. _sync_domains 写库前再归一一次，兼容仍返回字符串的旧/自定义适配器。
3. sync_account 单个能力失败后必须 rollback：否则 Session 停在 pending-rollback，
   后面更新 last_synced_at 的 commit 会抛 PendingRollbackError，接口直接 500，
   真实错误被掩盖成一条无信息的服务器异常。

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

from datetime import date
from unittest.mock import patch

import pytest
from sqlmodel import Session, SQLModel, create_engine, select

# create_all 需要 metadata 里齐备所有被外键引用的表（accounts.credential_id → credentials、
# assets.provider_id → providers 等），而 app/models/__init__.py 不导入子模块，
# 所以必须显式导入全部模型模块；否则单跑本文件会抛 NoReferencedTableError
# （跑全量套件时因其他测试先导入了模型而意外通过）
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 NormalizedDomain, parse_iso_date
from app.adapters.cloudflare import CloudflareAdapter
from app.adapters.digitalocean import DigitalOceanAdapter
from app.adapters.vultr import VultrAdapter
from app.models.asset import Account, Asset, DomainDetail
from app.models.provider import Provider
from app.services import sync_service


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


# ----------------------------- 到期日归一化 ----------------------------- #
def test_parse_iso_date_accepts_provider_shapes():
    """各厂商返回形态不一，统一解析为 date"""
    assert parse_iso_date("2027-03-15") == date(2027, 3, 15)
    assert parse_iso_date("2027-03-15T00:00:00Z") == date(2027, 3, 15)
    assert parse_iso_date("2027-03-15T00:00:00+00:00") == date(2027, 3, 15)
    assert parse_iso_date(date(2027, 3, 15)) == date(2027, 3, 15)


def test_parse_iso_date_bad_input_returns_none():
    """脏数据归 None，不能让整批同步炸掉"""
    assert parse_iso_date(None) is None
    assert parse_iso_date("") is None
    assert parse_iso_date("   ") is None
    assert parse_iso_date("not-a-date") is None


def test_cloudflare_domain_expiry_is_date():
    """Cloudflare Registrar 返回 ISO8601 时间戳，必须解析成 date 再交给上层"""
    adapter = CloudflareAdapter({"api_token": "x"})
    accounts = {"result": [{"id": "acc1"}]}
    registrar = {"result": [{"id": "example.com", "expires_at": "2027-03-15T00:00:00Z"}]}
    zones = {
        "result": [{"id": "z1", "name": "example.com", "status": "active"}],
        "result_info": {"total_pages": 1, "page": 1},
    }
    with patch.object(CloudflareAdapter, "_get", side_effect=[accounts, registrar, zones]):
        domains = adapter.list_domains()
    assert len(domains) == 1
    assert isinstance(domains[0].expiry_date, date)
    assert domains[0].expiry_date == date(2027, 3, 15)
    # to_dict 对外仍是 ISO 字符串（JSON 可序列化，形态与旧契约一致）
    assert domains[0].to_dict()["expiry_date"] == "2027-03-15"


def test_vultr_and_do_map_domain_expiry():
    """Vultr / DO 的域名到期日此前完全没映射，导致这两家域名永远不进续费提醒"""
    vultr = VultrAdapter({"api_key": "x"})
    vultr_resp = {
        "domains": [{"domain": "a.com", "expiry_date": "2027-01-02T00:00:00+00:00"}],
        "meta": {"links": {}},
    }
    with patch.object(VultrAdapter, "_get", return_value=vultr_resp):
        assert vultr.list_domains()[0].expiry_date == date(2027, 1, 2)

    do = DigitalOceanAdapter({"api_key": "x"})
    do_resp = {"domains": [{"name": "b.com", "expiry_date": "2027-01-03T00:00:00Z"}], "links": {}}
    with patch.object(DigitalOceanAdapter, "_get", return_value=do_resp):
        assert do.list_domains()[0].expiry_date == date(2027, 1, 3)


# ----------------------------- 同步写库 ----------------------------- #
class _FakeAdapter:
    """最小适配器：只实现同步链路用到的接口"""

    required_config: list = []

    def __init__(self, domains):
        self.config = {}
        self._domains = domains

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

    def list_domains(self):
        return self._domains


def _seed_provider_account(engine) -> tuple[int, int]:
    with Session(engine) as s:
        provider = Provider(
            slug="cloudflare", name="Cloudflare", category="cloudflare", sdk_type="cloudflare-api"
        )
        s.add(provider)
        s.commit()
        s.refresh(provider)
        account = Account(name="me@example.com", platform="cloudflare")
        s.add(account)
        s.commit()
        s.refresh(account)
        return provider.id, account.id


def test_sync_domains_coerces_string_expiry(engine):
    """适配器若仍返回字符串，写库前归一为 date，不应整批失败"""
    provider_id, account_id = _seed_provider_account(engine)
    adapter = _FakeAdapter(
        [
            NormalizedDomain(
                external_id="z1",
                domain_name="a.com",
                registrar="cloudflare",
                expiry_date="2027-03-15",  # 故意给字符串
            )
        ]
    )
    with Session(engine) as s:
        result = sync_service._sync_domains(s, s.get(Provider, provider_id), adapter, account_id)
    assert result == {"created": 1, "updated": 0}

    with Session(engine) as s:
        asset = s.exec(select(Asset)).first()
        assert asset.expiry_date == date(2027, 3, 15)
        detail = s.exec(select(DomainDetail)).first()
        assert detail.domain_name == "a.com"


def test_sync_domains_updates_existing_expiry(engine):
    """重复同步走更新分支：到期日按新值刷新，不新建资产"""
    provider_id, account_id = _seed_provider_account(engine)

    def _run(expiry):
        adapter = _FakeAdapter(
            [NormalizedDomain(external_id="z1", domain_name="a.com", expiry_date=expiry)]
        )
        with Session(engine) as s:
            return sync_service._sync_domains(s, s.get(Provider, provider_id), adapter, account_id)

    assert _run("2027-03-15") == {"created": 1, "updated": 0}
    assert _run("2028-03-15") == {"created": 0, "updated": 1}
    with Session(engine) as s:
        assert s.exec(select(Asset)).first().expiry_date == date(2028, 3, 15)


def test_sync_account_survives_failed_capability(engine, monkeypatch):
    """单个能力写库失败后必须回滚，否则 last_synced_at 的 commit 抛 PendingRollbackError"""
    provider_id, account_id = _seed_provider_account(engine)
    # domain_name=None 违反 Asset.name NOT NULL → flush 失败，Session 进入 pending-rollback
    adapter = _FakeAdapter([NormalizedDomain(external_id="z1", domain_name=None)])
    monkeypatch.setattr(
        sync_service,
        "_build_adapter_for_account",
        lambda session, account: (session.get(Provider, provider_id), adapter),
    )

    with Session(engine) as s:
        result = sync_service.sync_account(s, account_id)  # 修复前此处抛 PendingRollbackError

    # 真实错误被如实带回，而不是被 500 掩盖
    assert result.get("domains") is None
    assert "domains_error" in result
    assert result["last_synced_at"]
    with Session(engine) as s:
        assert s.get(Account, account_id).last_synced_at is not None
        # 失败的那批不残留半条数据
        assert s.exec(select(Asset)).first() is None