From 90695c82fd4f6966df8b4d3db155303e2ed466ec Mon Sep 17 00:00:00 2001
From: gouki <gouki.xiao@gmail.com>
Date: Thu, 15 Oct 2026 21:18:54 +0000
Subject: [PATCH] =?UTF-8?q?test:=20=E8=A1=A5=E5=85=85=20renewal=5Fservice?=
 =?UTF-8?q?=20/=20ssl=5Fservice=20/=20=E6=A0=B8=E5=BF=83=20router=20HTTP?=
 =?UTF-8?q?=20smoke=20test?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- test_renewal_service.py: 14 个测试覆盖到期查询、文案生成、消息构建、续费检查流程
- test_ssl_service.py: 24 个测试覆盖子域名 CRUD、证书辅助函数、探测结果写入、全量刷新
- test_router_smoke.py: 16 个 HTTP smoke test 覆盖 assets/accounts/providers/stats 路由

全部 394 个测试通过
---
 tests/test_renewal_service.py | 174 +++++++++++++++++++
 tests/test_router_smoke.py    | 225 +++++++++++++++++++++++++
 tests/test_ssl_service.py     | 303 ++++++++++++++++++++++++++++++++++
 3 files changed, 702 insertions(+)
 create mode 100644 tests/test_renewal_service.py
 create mode 100644 tests/test_router_smoke.py
 create mode 100644 tests/test_ssl_service.py

diff --git a/tests/test_renewal_service.py b/tests/test_renewal_service.py
new file mode 100644
index 0000000..e23c1df
--- /dev/null
+++ b/tests/test_renewal_service.py
@@ -0,0 +1,174 @@
+"""renewal_service 测试：续费检查与提醒消息生成
+
+覆盖：
+- get_expiring_assets: 到期资产查询（含已过期、排除已注销、按天数升序）
+- day_text: 到期文案生成（已过期/今天/未来）
+- build_renewal_message: 消息拼装（有过期+未过期混合、全部未过期、空列表）
+- check_renewals: 完整流程（send=True 触发通知 / send=False 仅预览 / 默认阈值取配置）
+
+全程 tmp_path + create_engine 临时库，mock 通知服务避免真实发送。
+
+运行：.venv/bin/pytest tests/test_renewal_service.py -v
+"""
+
+from datetime import date, timedelta
+from unittest.mock import patch
+
+import pytest
+from sqlmodel import Session, SQLModel, create_engine
+
+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.models.asset import Asset, AssetStatus, AssetType
+from app.services import renewal_service
+
+
+@pytest.fixture()
+def engine(tmp_path):
+    eng = create_engine(
+        f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False}
+    )
+    SQLModel.metadata.create_all(eng)
+    yield eng
+    eng.dispose()
+
+
+def _seed(engine, name, *, expiry, status=AssetStatus.ACTIVE, provider="vultr"):
+    with Session(engine) as s:
+        s.add(Asset(
+            name=name, asset_type=AssetType.VPS, provider=provider,
+            expiry_date=expiry, status=status,
+        ))
+        s.commit()
+
+
+# ----------------------------- day_text ----------------------------- #
+def test_day_text_overdue():
+    assert renewal_service.day_text(-5) == "已过期 5 天"
+    assert renewal_service.day_text(-1) == "已过期 1 天"
+
+
+def test_day_text_today():
+    assert renewal_service.day_text(0) == "今天到期"
+
+
+def test_day_text_future():
+    assert renewal_service.day_text(1) == "1 天后"
+    assert renewal_service.day_text(30) == "30 天后"
+
+
+# ----------------------------- get_expiring_assets ----------------------------- #
+def test_get_expiring_assets_empty(engine):
+    with Session(engine) as s:
+        result = renewal_service.get_expiring_assets(s, threshold_days=30)
+    assert result == []
+
+
+def test_get_expiring_assets_includes_overdue_and_sorted(engine):
+    """已过期资产也纳入提醒（不能因过了到期日就消失），按剩余天数升序"""
+    today = date.today()
+    _seed(engine, "overdue", expiry=today - timedelta(days=3))
+    _seed(engine, "soon", expiry=today + timedelta(days=5))
+    _seed(engine, "far", expiry=today + timedelta(days=90))
+    with Session(engine) as s:
+        result = renewal_service.get_expiring_assets(s, threshold_days=30)
+    names = [a.name for a, _ in result]
+    assert names == ["overdue", "soon"]
+    assert result[0][1] < result[1][1]  # -3 < 5
+
+
+def test_get_expiring_assets_excludes_cancelled(engine):
+    """已注销资产不参与续费提醒"""
+    today = date.today()
+    _seed(engine, "active", expiry=today + timedelta(days=5))
+    _seed(engine, "cancelled", expiry=today + timedelta(days=5), status=AssetStatus.CANCELLED)
+    with Session(engine) as s:
+        result = renewal_service.get_expiring_assets(s, threshold_days=30)
+    names = [a.name for a, _ in result]
+    assert names == ["active"]
+
+
+def test_get_expiring_assets_no_expiry_date_skipped(engine):
+    """未填到期日的资产不出现在到期列表"""
+    _seed(engine, "no-expiry", expiry=None)
+    # _seed 不支持 None expiry，直接插入
+    with Session(engine) as s:
+        s.add(Asset(name="no-expiry", asset_type=AssetType.VPS, provider="vultr"))
+        s.commit()
+        result = renewal_service.get_expiring_assets(s, threshold_days=30)
+    assert result == []
+
+
+# ----------------------------- build_renewal_message ----------------------------- #
+def test_build_message_empty():
+    assert renewal_service.build_renewal_message([], 30) == ""
+
+
+def test_build_message_with_overdue():
+    """混合已过期和即将到期时，消息头部区分两类"""
+    today = date.today()
+    a1 = Asset(name="vps-1", provider="vultr", expiry_date=today - timedelta(days=2))
+    a2 = Asset(name="vps-2", provider="vultr", expiry_date=today + timedelta(days=10))
+    expiring = [(a1, -2), (a2, 10)]
+    msg = renewal_service.build_renewal_message(expiring, threshold_days=30)
+    assert "已过期 1 项" in msg
+    assert "30 天内到期 1 项" in msg
+    assert "vps-1" in msg
+    assert "vps-2" in msg
+    assert "已过期 2 天" in msg
+    assert "10 天后" in msg
+
+
+def test_build_message_no_overdue():
+    """全部未过期时头部只显示常规文案"""
+    today = date.today()
+    a1 = Asset(name="vps-1", provider="vultr", expiry_date=today + timedelta(days=5))
+    expiring = [(a1, 5)]
+    msg = renewal_service.build_renewal_message(expiring, threshold_days=30)
+    assert "已过期" not in msg
+    assert "共 1 项" in msg
+
+
+# ----------------------------- check_renewals ----------------------------- #
+def test_check_renewals_no_send(engine):
+    """send=False 时不触发通知，仅返回预览"""
+    today = date.today()
+    _seed(engine, "soon", expiry=today + timedelta(days=5))
+    with Session(engine) as s:
+        result = renewal_service.check_renewals(s, threshold_days=30, send=False)
+    assert result["expiring_count"] == 1
+    assert result["threshold_days"] == 30
+    assert result["notifications"] == []
+    assert result["expiring"][0]["name"] == "soon"
+    assert result["expiring"][0]["days"] == 5
+
+
+def test_check_renewals_send_with_no_expiring(engine):
+    """无到期资产时不发送通知"""
+    with Session(engine) as s:
+        result = renewal_service.check_renewals(s, threshold_days=30, send=True)
+    assert result["expiring_count"] == 0
+    assert result["message"] == ""
+    assert result["notifications"] == []
+
+
+def test_check_renewals_send_dispatches_notification(engine):
+    """send=True 且有到期资产时调用通知服务"""
+    today = date.today()
+    _seed(engine, "soon", expiry=today + timedelta(days=3))
+    with Session(engine) as s:
+        with patch.object(renewal_service.notification_service, "notify", return_value=[{"ok": True}]) as mock_notify:
+            result = renewal_service.check_renewals(s, threshold_days=30, send=True)
+            mock_notify.assert_called_once()
+    assert result["expiring_count"] == 1
+    assert result["notifications"] == [{"ok": True}]
+
+
+def test_check_renewals_default_threshold_from_config(engine):
+    """threshold_days=None 时从 settings 读取默认值"""
+    with Session(engine) as s:
+        result = renewal_service.check_renewals(s, send=False)
+    assert result["threshold_days"] == renewal_service.settings.RENEWAL_THRESHOLD_DAYS
diff --git a/tests/test_router_smoke.py b/tests/test_router_smoke.py
new file mode 100644
index 0000000..73640f4
--- /dev/null
+++ b/tests/test_router_smoke.py
@@ -0,0 +1,225 @@
+"""核心 router HTTP smoke test：验证 assets / accounts / providers / stats 路由装配与基本 CRUD
+
+使用 TestClient 挂载真实 app，依赖覆盖到 tmp_path 临时库，覆盖：
+- GET 列表接口返回 200 + 正确结构（空库 / 有数据）
+- POST 创建 → GET 详情 → PUT 更新 → DELETE 删除的完整链路
+- /api/stats/overview 和 /api/stats/expiring 返回正确结构
+- /api/providers/adapters 返回适配器元信息
+
+安全前提：TestClient 使用本机回环来源（可信内网），读操作无需 API Key；
+写操作同样来自可信内网，不走 fail-closed 分支。
+
+全程免凭证、免网络：不触发 lifespan，get_session / get_metrics_session 覆盖到临时库。
+
+运行：.venv/bin/pytest tests/test_router_smoke.py -v
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlmodel import Session, SQLModel, create_engine
+
+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.core.config import settings
+from app.database import get_metrics_session, get_session
+from app.main import app
+
+
+@pytest.fixture()
+def client(tmp_path, monkeypatch):
+    """挂载真实 app 的 TestClient：依赖覆盖到临时库"""
+    assets_eng = create_engine(
+        f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False}
+    )
+    metrics_eng = create_engine(
+        f"sqlite:///{tmp_path / 'metrics.db'}", connect_args={"check_same_thread": False}
+    )
+    SQLModel.metadata.create_all(assets_eng)
+    SQLModel.metadata.create_all(metrics_eng)
+
+    def _assets_session():
+        with Session(assets_eng, expire_on_commit=False) as s:
+            yield s
+
+    def _metrics_session():
+        with Session(metrics_eng, expire_on_commit=False) as s:
+            yield s
+
+    app.dependency_overrides[get_session] = _assets_session
+    app.dependency_overrides[get_metrics_session] = _metrics_session
+    monkeypatch.setattr(settings, "API_KEY", "")
+    monkeypatch.setattr(settings, "AGENT_KEY", "")
+
+    yield TestClient(app, client=("127.0.0.1", 0))
+
+    app.dependency_overrides.clear()
+    assets_eng.dispose()
+    metrics_eng.dispose()
+
+
+# ============================= /api/assets ============================= #
+class TestAssetsRouter:
+    def test_list_empty(self, client):
+        resp = client.get("/api/assets")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+    def test_create_and_get(self, client):
+        body = {
+            "name": "smoke-vps",
+            "asset_type": "vps",
+            "provider": "vultr",
+            "vps_detail": {"ip_address": "10.0.0.1"},
+        }
+        resp = client.post("/api/assets", json=body)
+        assert resp.status_code == 201
+        created = resp.json()
+        assert created["name"] == "smoke-vps"
+        assert created["vps_detail"]["ip_address"] == "10.0.0.1"
+
+        # GET 详情
+        resp = client.get(f"/api/assets/{created['id']}")
+        assert resp.status_code == 200
+        assert resp.json()["name"] == "smoke-vps"
+
+    def test_create_list_then_delete(self, client):
+        body = {
+            "name": "del-vps",
+            "asset_type": "vps",
+            "provider": "vultr",
+            "vps_detail": {"ip_address": "10.0.0.2"},
+        }
+        created = client.post("/api/assets", json=body).json()
+        assert len(client.get("/api/assets").json()) == 1
+
+        resp = client.delete(f"/api/assets/{created['id']}")
+        assert resp.status_code == 204
+        assert client.get("/api/assets").json() == []
+
+    def test_get_nonexistent_returns_404(self, client):
+        resp = client.get("/api/assets/999999")
+        assert resp.status_code == 404
+
+
+# ============================= /api/accounts ============================= #
+class TestAccountsRouter:
+    def test_list_empty(self, client):
+        resp = client.get("/api/accounts")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+    def test_create_and_list(self, client):
+        body = {"name": "test@example.com", "platform": "vultr"}
+        resp = client.post("/api/accounts", json=body)
+        assert resp.status_code == 201
+        created = resp.json()
+        assert created["name"] == "test@example.com"
+        assert created["platform"] == "vultr"
+
+        listing = client.get("/api/accounts").json()
+        assert len(listing) == 1
+        assert listing[0]["name"] == "test@example.com"
+
+    def test_create_duplicate_returns_400(self, client):
+        body = {"name": "dup@example.com", "platform": "vultr"}
+        client.post("/api/accounts", json=body)
+        resp = client.post("/api/accounts", json=body)
+        assert resp.status_code == 400
+
+    def test_delete_account(self, client):
+        body = {"name": "del@example.com", "platform": "vultr"}
+        created = client.post("/api/accounts", json=body).json()
+        resp = client.delete(f"/api/accounts/{created['id']}")
+        assert resp.status_code == 200
+        assert "affected_assets" in resp.json()
+        assert client.get("/api/accounts").json() == []
+
+
+# ============================= /api/providers ============================= #
+class TestProvidersRouter:
+    def test_list_empty(self, client):
+        resp = client.get("/api/providers")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+    def test_create_and_get(self, client):
+        body = {
+            "slug": "test-provider",
+            "name": "Test Provider",
+            "category": "vps",
+        }
+        resp = client.post("/api/providers", json=body)
+        assert resp.status_code == 201
+        created = resp.json()
+        assert created["slug"] == "test-provider"
+
+        resp = client.get(f"/api/providers/{created['id']}")
+        assert resp.status_code == 200
+        assert resp.json()["name"] == "Test Provider"
+
+    def test_create_duplicate_slug_returns_400(self, client):
+        body = {"slug": "dup-slug", "name": "First", "category": "vps"}
+        client.post("/api/providers", json=body)
+        body["name"] = "Second"
+        resp = client.post("/api/providers", json=body)
+        assert resp.status_code == 400
+
+    def test_adapters_endpoint(self, client):
+        resp = client.get("/api/providers/adapters")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "adapters" in data
+        assert "supported_types" in data
+        assert "capabilities" in data
+
+    def test_delete_provider(self, client):
+        body = {"slug": "del-provider", "name": "Del", "category": "vps"}
+        created = client.post("/api/providers", json=body).json()
+        resp = client.delete(f"/api/providers/{created['id']}")
+        assert resp.status_code == 204
+        assert client.get("/api/providers").json() == []
+
+
+# ============================= /api/stats ============================= #
+class TestStatsRouter:
+    def test_overview_empty(self, client):
+        resp = client.get("/api/stats/overview")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total"] == 0
+        assert "year_cost" in data
+        assert "expiring_30" in data
+        assert "overdue_count" in data
+
+    def test_expiring_empty(self, client):
+        resp = client.get("/api/stats/expiring")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+    def test_overview_with_data(self, client):
+        from datetime import date
+        # 先创建一个带到期日的资产（year_cost 只统计本年到期资产）
+        body = {
+            "name": "stats-vps",
+            "asset_type": "vps",
+            "provider": "vultr",
+            "vps_detail": {"ip_address": "10.0.0.3"},
+            "cost": 12.5,
+            "expiry_date": str(date.today()),
+        }
+        client.post("/api/assets", json=body)
+        data = client.get("/api/stats/overview").json()
+        assert data["total"] == 1
+        assert data["year_cost"] == 12.5
+
+    def test_expiring_param_validation(self, client):
+        """days 参数有范围限制（1-365）"""
+        resp = client.get("/api/stats/expiring?days=0")
+        assert resp.status_code == 422
+        resp = client.get("/api/stats/expiring?days=366")
+        assert resp.status_code == 422
+        resp = client.get("/api/stats/expiring?days=30")
+        assert resp.status_code == 200
diff --git a/tests/test_ssl_service.py b/tests/test_ssl_service.py
new file mode 100644
index 0000000..0069db2
--- /dev/null
+++ b/tests/test_ssl_service.py
@@ -0,0 +1,303 @@
+"""ssl_service 测试：子域名 CRUD + 站点证书辅助函数 + 证书监控流程
+
+覆盖：
+- days_until / _judge_status: 纯函数辅助
+- _to_dict: 模型转 API 返回结构
+- 子域名 CRUD: create / list / update / delete（含异常分支）
+- _apply_probe: 探测结果写入模型
+- build_cert_message: 证书提醒文案
+- list_site_certs / delete_site_cert: 站点证书列表与删除
+- check_all_site_certs: 全量刷新（mock probe 避免网络）
+
+全程 tmp_path + create_engine 临时库，mock probe_site_cert 避免真实网络探测。
+
+运行：.venv/bin/pytest tests/test_ssl_service.py -v
+"""
+
+from datetime import date, timedelta
+from unittest.mock import patch
+
+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.models.asset import Asset, AssetType
+from app.models.ssl import SiteCert, Subdomain
+from app.services import ssl_service
+
+
+@pytest.fixture()
+def engine(tmp_path):
+    eng = create_engine(
+        f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False}
+    )
+    SQLModel.metadata.create_all(eng)
+    yield eng
+    eng.dispose()
+
+
+def _seed_asset(engine, name="example.com", asset_type=AssetType.DOMAIN, provider="namesilo"):
+    with Session(engine) as s:
+        a = Asset(name=name, asset_type=asset_type, provider=provider)
+        s.add(a)
+        s.commit()
+        s.refresh(a)
+        return a.id
+
+
+# ----------------------------- 纯函数 ----------------------------- #
+def test_days_until_none():
+    assert ssl_service.days_until(None) is None
+
+
+def test_days_until_future():
+    d = date.today() + timedelta(days=10)
+    assert ssl_service.days_until(d) == 10
+
+
+def test_days_until_past():
+    d = date.today() - timedelta(days=5)
+    assert ssl_service.days_until(d) == -5
+
+
+def test_judge_status():
+    assert ssl_service._judge_status(None) == "unknown"
+    assert ssl_service._judge_status(-1) == "expired"
+    assert ssl_service._judge_status(0) == "expiring"
+    assert ssl_service._judge_status(15) == "expiring"
+    assert ssl_service._judge_status(31) == "valid"
+    # 自定义阈值
+    assert ssl_service._judge_status(5, threshold=3) == "valid"
+    assert ssl_service._judge_status(3, threshold=3) == "expiring"
+
+
+# ----------------------------- 子域名 CRUD ----------------------------- #
+def test_create_subdomain(engine):
+    asset_id = _seed_asset(engine)
+    with Session(engine) as s:
+        sub = ssl_service.create_subdomain(s, {
+            "asset_id": asset_id, "host": "WWW.Example.com",
+            "record_type": "A", "record_value": "1.2.3.4",
+        })
+        assert sub.host == "www.example.com"  # strip + lower
+        assert sub.is_active is True
+        assert sub.record_type == "A"
+
+
+def test_create_subdomain_invalid_asset_raises(engine):
+    with Session(engine) as s:
+        with pytest.raises(ValueError, match="关联资产不存在"):
+            ssl_service.create_subdomain(s, {"asset_id": 99999, "host": "www"})
+
+
+def test_list_subdomains(engine):
+    asset_id = _seed_asset(engine)
+    with Session(engine) as s:
+        s.add(Subdomain(asset_id=asset_id, host="www"))
+        s.add(Subdomain(asset_id=asset_id, host="api"))
+        s.commit()
+        result = ssl_service.list_subdomains(s, asset_id=asset_id)
+    assert len(result) == 2
+    hosts = [r["host"] for r in result]
+    assert "api" in hosts
+    assert "www" in hosts
+    # 应带上 asset_name
+    assert result[0]["asset_name"] == "example.com"
+
+
+def test_list_subdomains_all(engine):
+    """不传 asset_id 时返回全部子域名"""
+    a1 = _seed_asset(engine, "site1.com")
+    a2 = _seed_asset(engine, "site2.com")
+    with Session(engine) as s:
+        s.add(Subdomain(asset_id=a1, host="www"))
+        s.add(Subdomain(asset_id=a2, host="blog"))
+        s.commit()
+        result = ssl_service.list_subdomains(s)
+    assert len(result) == 2
+
+
+def test_update_subdomain(engine):
+    asset_id = _seed_asset(engine)
+    with Session(engine) as s:
+        sub = ssl_service.create_subdomain(s, {"asset_id": asset_id, "host": "www"})
+        updated = ssl_service.update_subdomain(s, sub.id, {"host": "API", "is_active": False})
+        assert updated.host == "api"
+        assert updated.is_active is False
+
+
+def test_update_subdomain_not_found_raises(engine):
+    with Session(engine) as s:
+        with pytest.raises(ValueError, match="记录不存在"):
+            ssl_service.update_subdomain(s, 99999, {"host": "x"})
+
+
+def test_delete_subdomain(engine):
+    asset_id = _seed_asset(engine)
+    with Session(engine) as s:
+        sub = ssl_service.create_subdomain(s, {"asset_id": asset_id, "host": "www"})
+        ssl_service.delete_subdomain(s, sub.id)
+        assert s.get(Subdomain, sub.id) is None
+
+
+def test_delete_subdomain_not_found_raises(engine):
+    with Session(engine) as s:
+        with pytest.raises(ValueError, match="记录不存在"):
+            ssl_service.delete_subdomain(s, 99999)
+
+
+# ----------------------------- 站点证书辅助 ----------------------------- #
+def test_to_dict_with_asset_name(engine):
+    with Session(engine) as s:
+        cert = SiteCert(hostname="example.com", port=443, valid_to=date.today() + timedelta(days=10))
+        s.add(cert)
+        s.commit()
+        s.refresh(cert)
+        d = ssl_service._to_dict(cert, asset_name="my-asset")
+    assert d["hostname"] == "example.com"
+    assert d["asset_name"] == "my-asset"
+    assert d["days_to_expiry"] == 10
+    assert d["status"] == "unknown"  # 模型默认值
+
+
+def test_apply_probe_sets_status():
+    """探测结果写入模型后，状态按剩余天数自动判定"""
+    cert = SiteCert(hostname="example.com")
+    info = {
+        "subject_cn": "example.com",
+        "issuer": "Let's Encrypt",
+        "valid_from": date(2026, 1, 1),
+        "valid_to": date.today() + timedelta(days=60),
+        "fingerprint": "abc123",
+        "san_list": ["example.com"],
+    }
+    result = ssl_service._apply_probe(cert, info)
+    assert result.status == "valid"
+    assert result.error is None
+    assert result.fingerprint == "abc123"
+    assert result.last_checked_at is not None
+
+
+def test_apply_probe_expiring():
+    cert = SiteCert(hostname="example.com")
+    info = {
+        "subject_cn": "example.com", "issuer": "LE",
+        "valid_from": date(2026, 1, 1),
+        "valid_to": date.today() + timedelta(days=5),
+        "fingerprint": "fp", "san_list": [],
+    }
+    ssl_service._apply_probe(cert, info)
+    assert cert.status == "expiring"
+
+
+def test_build_cert_message():
+    cert = SiteCert(hostname="example.com", valid_to=date.today() + timedelta(days=10))
+    msg = ssl_service.build_cert_message(cert)
+    assert "10 天后到期" in msg
+    assert "example.com" in msg
+
+
+def test_build_cert_message_expired():
+    cert = SiteCert(hostname="example.com", valid_to=date.today() - timedelta(days=3))
+    msg = ssl_service.build_cert_message(cert)
+    assert "已过期 3 天" in msg
+
+
+def test_build_cert_message_unknown():
+    cert = SiteCert(hostname="example.com")
+    msg = ssl_service.build_cert_message(cert)
+    assert "未知" in msg
+
+
+# ----------------------------- 站点证书 CRUD ----------------------------- #
+def test_list_site_certs_empty(engine):
+    with Session(engine) as s:
+        assert ssl_service.list_site_certs(s) == []
+
+
+def test_list_site_certs_with_filter(engine):
+    with Session(engine) as s:
+        s.add(SiteCert(hostname="a.com", status="valid"))
+        s.add(SiteCert(hostname="b.com", status="error"))
+        s.commit()
+        all_certs = ssl_service.list_site_certs(s)
+        valid_only = ssl_service.list_site_certs(s, status="valid")
+    assert len(all_certs) == 2
+    assert len(valid_only) == 1
+    assert valid_only[0]["hostname"] == "a.com"
+
+
+def test_delete_site_cert(engine):
+    with Session(engine) as s:
+        cert = SiteCert(hostname="del.com")
+        s.add(cert)
+        s.commit()
+        s.refresh(cert)
+        ssl_service.delete_site_cert(s, cert.id)
+        assert s.get(SiteCert, cert.id) is None
+
+
+def test_delete_site_cert_not_found_raises(engine):
+    with Session(engine) as s:
+        with pytest.raises(ValueError, match="记录不存在"):
+            ssl_service.delete_site_cert(s, 99999)
+
+
+def test_create_site_cert_reuses_existing(engine):
+    """hostname+port 已存在时复用并刷新，不新建"""
+    with Session(engine) as s:
+        s.add(SiteCert(hostname="dup.com", port=443))
+        s.commit()
+
+    probe_result = {
+        "subject_cn": "dup.com", "issuer": "LE",
+        "valid_from": date(2026, 1, 1),
+        "valid_to": date.today() + timedelta(days=60),
+        "fingerprint": "fp", "san_list": [],
+    }
+    with Session(engine) as s:
+        with patch.object(ssl_service, "probe_site_cert", return_value=probe_result):
+            result = ssl_service.create_site_cert(s, "dup.com")
+        assert result.hostname == "dup.com"
+        assert result.status == "valid"
+        # 应该只有一条记录
+        count = s.exec(select(SiteCert)).all()
+        assert len(count) == 1
+
+
+def test_check_all_site_certs_with_mock(engine):
+    """全量刷新：mock probe 避免网络，验证统计与问题清单"""
+    with Session(engine) as s:
+        s.add(SiteCert(hostname="ok.com"))
+        s.add(SiteCert(hostname="err.com"))
+        s.commit()
+
+    def fake_probe(hostname, port=443):
+        if hostname == "ok.com":
+            return {
+                "subject_cn": hostname, "issuer": "LE",
+                "valid_from": date(2026, 1, 1),
+                "valid_to": date.today() + timedelta(days=90),
+                "fingerprint": "fp1", "san_list": [],
+            }
+        raise ConnectionError("connection refused")
+
+    with Session(engine) as s:
+        with patch.object(ssl_service, "probe_site_cert", side_effect=fake_probe):
+            result = ssl_service.check_all_site_certs(s)
+    assert result["stats"]["total"] == 2
+    assert result["stats"]["ok"] == 1
+    assert result["stats"]["error"] == 1
+    assert len(result["problems"]) == 1
+    assert result["problems"][0]["hostname"] == "err.com"
+
+
+def test_check_all_site_certs_empty(engine):
+    with Session(engine) as s:
+        result = ssl_service.check_all_site_certs(s)
+    assert result["stats"]["total"] == 0
+    assert result["problems"] == []