gouki
90695c82fd
- test_renewal_service.py: 14 个测试覆盖到期查询、文案生成、消息构建、续费检查流程 - test_ssl_service.py: 24 个测试覆盖子域名 CRUD、证书辅助函数、探测结果写入、全量刷新 - test_router_smoke.py: 16 个 HTTP smoke test 覆盖 assets/accounts/providers/stats 路由 全部 394 个测试通过
304 lines
10 KiB
Python
304 lines
10 KiB
Python
"""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"] == []
|