gouki
abd2fa8206
容错策略：核对失败/无可用通道时标记「未核对」并继续按本地数据提醒， 避免因核对链路故障漏掉真正到期的资产；已停止资产仅在 auto_renew=true 时提醒（停止实例不等于停止扣费）。新增 WHMCS 面板只读适配器， 用于 TotHost 这类没有开放 API 的平台。
262 lines
10 KiB
Python
262 lines
10 KiB
Python
"""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
|
|
from app.services.verification_service import VerificationResult
|
|
|
|
|
|
@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", auto_renew=False):
|
|
with Session(engine) as s:
|
|
s.add(Asset(
|
|
name=name, asset_type=AssetType.VPS, provider=provider,
|
|
expiry_date=expiry, status=status, auto_renew=auto_renew,
|
|
))
|
|
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_stopped_without_auto_renew_excluded(engine):
|
|
"""已停止且未开自动续费的资产不再提醒"""
|
|
today = date.today()
|
|
_seed(engine, "stopped", expiry=today + timedelta(days=5), status=AssetStatus.STOPPED)
|
|
with Session(engine) as s:
|
|
result = renewal_service.get_expiring_assets(s, threshold_days=30)
|
|
assert result == []
|
|
|
|
|
|
def test_get_expiring_assets_stopped_with_auto_renew_included(engine):
|
|
"""已停止但自动续费仍开着，可能继续扣费，必须提醒"""
|
|
today = date.today()
|
|
_seed(
|
|
engine, "stopped-autorenew", expiry=today + timedelta(days=5),
|
|
status=AssetStatus.STOPPED, auto_renew=True,
|
|
)
|
|
with Session(engine) as s:
|
|
result = renewal_service.get_expiring_assets(s, threshold_days=30)
|
|
assert [a.name for a, _ in result] == ["stopped-autorenew"]
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_build_message_labels_verification_source():
|
|
"""消息中区分「已核对」与「未核对」，并给出核对汇总"""
|
|
today = date.today()
|
|
a1 = Asset(id=1, name="vps-1", provider="tothost", expiry_date=today - timedelta(days=1))
|
|
a2 = Asset(id=2, name="vps-2", provider="agnes", expiry_date=today + timedelta(days=5))
|
|
verification = {
|
|
1: VerificationResult(
|
|
asset_id=1, verified=True, source="TotHost 面板",
|
|
detail="服务 TOT K11 vpn.host：可用，下次到期 2026-11-25",
|
|
),
|
|
2: VerificationResult(asset_id=2, verified=False, detail="平台「Agnes」未接入自动核对"),
|
|
}
|
|
msg = renewal_service.build_renewal_message(
|
|
[(a1, -1), (a2, 5)], threshold_days=30, verification=verification
|
|
)
|
|
assert "已核对：服务 TOT K11 vpn.host" in msg
|
|
assert "未核对：平台「Agnes」未接入自动核对" in msg
|
|
assert "核对：已核对 1 项，未核对 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
|
|
|
|
|
|
def test_check_renewals_verification_suppresses_false_alarm(engine):
|
|
"""核对发现已续费到 90 天后时，本轮不再提醒"""
|
|
today = date.today()
|
|
_seed(engine, "renewed", expiry=today - timedelta(days=1))
|
|
|
|
def fake_verify(session, assets):
|
|
results = {}
|
|
for asset in assets:
|
|
asset.expiry_date = today + timedelta(days=90)
|
|
session.add(asset)
|
|
results[asset.id] = VerificationResult(
|
|
asset_id=asset.id, verified=True, source="TotHost 面板",
|
|
expiry_date=asset.expiry_date, detail="已续费",
|
|
)
|
|
session.commit()
|
|
return results
|
|
|
|
with Session(engine) as s:
|
|
with patch.object(
|
|
renewal_service.verification_service, "verify_assets", side_effect=fake_verify
|
|
) as mock_verify:
|
|
result = renewal_service.check_renewals(s, threshold_days=30, send=False)
|
|
mock_verify.assert_called_once()
|
|
assert result["expiring_count"] == 0
|
|
assert result["message"] == ""
|
|
assert result["notifications"] == []
|
|
|
|
|
|
def test_check_renewals_can_disable_verification(engine):
|
|
"""verify=False 时不调用核对服务"""
|
|
today = date.today()
|
|
_seed(engine, "soon", expiry=today + timedelta(days=5))
|
|
with Session(engine) as s:
|
|
with patch.object(
|
|
renewal_service.verification_service, "verify_assets",
|
|
side_effect=AssertionError("不应调用核对"),
|
|
) as mock_verify:
|
|
result = renewal_service.check_renewals(
|
|
s, threshold_days=30, send=False, verify=False
|
|
)
|
|
mock_verify.assert_not_called()
|
|
assert result["expiring_count"] == 1
|
|
assert result["verification"] is None
|