gouki
abd2fa8206
容错策略：核对失败/无可用通道时标记「未核对」并继续按本地数据提醒， 避免因核对链路故障漏掉真正到期的资产；已停止资产仅在 auto_renew=true 时提醒（停止实例不等于停止扣费）。新增 WHMCS 面板只读适配器， 用于 TotHost 这类没有开放 API 的平台。
229 lines
9.7 KiB
Python
229 lines
9.7 KiB
Python
"""云资产同步链路测试：路由注册 + 定时脚本 + systemd 单元接线
|
||
|
||
这条链有三个环节，任一环断了都不会报错，只会「夜间同步悄悄不跑了」：
|
||
1. /api/sync/assets 没注册 -> 手动触发按钮 404
|
||
2. scripts/sync_all_assets.py 退出码恒为 0 -> systemd 永远记成功，失败无告警
|
||
3. deploy/*.timer 没被 setup.sh enable -> 全新部署后定时器根本不存在
|
||
|
||
运行：.venv/bin/pytest tests/test_sync_chain.py -v
|
||
"""
|
||
|
||
import importlib.util
|
||
import re
|
||
from datetime import time
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlmodel import 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
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
SCRIPTS_DIR = REPO_ROOT / "scripts"
|
||
DEPLOY_DIR = REPO_ROOT / "deploy"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 1. 路由注册
|
||
# --------------------------------------------------------------------------- #
|
||
def test_sync_assets_endpoint_is_registered():
|
||
"""/api/sync/assets 必须出现在 OpenAPI 里
|
||
|
||
这个 FastAPI 版本把 include_router 进来的路由包在 _IncludedRouter 里延迟展开，
|
||
遍历 app.routes 看不到，只能查 app.openapi()['paths']。
|
||
"""
|
||
from app.main import app
|
||
|
||
paths = app.openapi()["paths"]
|
||
assert "/api/sync/assets" in paths, f"未注册；现有 sync 路由：{[p for p in paths if 'sync' in p]}"
|
||
assert "post" in paths["/api/sync/assets"]
|
||
|
||
|
||
def test_sync_assets_endpoint_is_key_protected():
|
||
"""批量同步会解密所有账号凭证并打云端接口，不能匿名可调"""
|
||
from app.core.security import require_api_key
|
||
from app.routers.sync import router
|
||
|
||
route = next(r for r in router.routes if r.path == "/api/sync/assets")
|
||
calls = [d.call for d in route.dependant.dependencies]
|
||
assert require_api_key in calls, "缺少 require_api_key 依赖，端点对外裸奔"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 2. 定时脚本
|
||
# --------------------------------------------------------------------------- #
|
||
def _load_script():
|
||
"""按文件路径加载 scripts/sync_all_assets.py（scripts/ 不是包）"""
|
||
path = SCRIPTS_DIR / "sync_all_assets.py"
|
||
spec = importlib.util.spec_from_file_location("_sync_all_assets_under_test", path)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
@pytest.fixture()
|
||
def script(tmp_path, monkeypatch):
|
||
eng = create_engine(
|
||
f"sqlite:///{tmp_path / 'chain.db'}", connect_args={"check_same_thread": False}
|
||
)
|
||
SQLModel.metadata.create_all(eng)
|
||
# sync_service 内部是「函数里 from app.database import assets_engine」，所以
|
||
# 补丁打在 app.database 上才生效；而脚本自己在模块级 import 了同名对象，
|
||
# 那份绑定必须单独换掉，否则 init_db() 会去建真的 data/*.db。
|
||
monkeypatch.setattr("app.database.assets_engine", eng)
|
||
mod = _load_script()
|
||
monkeypatch.setattr(mod, "init_db", lambda: None)
|
||
monkeypatch.setattr(mod, "assets_engine", eng)
|
||
yield mod
|
||
eng.dispose()
|
||
|
||
|
||
def test_script_summary_prints_every_key_the_service_returns(script, capsys):
|
||
"""端到端跑一次真实 sync_all_assets（空库），确保摘要行不因缺键 KeyError
|
||
|
||
脚本直接下标访问 result['skipped_no_config'] 等键；服务层返回结构一改，
|
||
这里就会在夜间定时任务里抛 KeyError，而 timer 的失败没人看。
|
||
"""
|
||
rc = script.main()
|
||
out = capsys.readouterr().out
|
||
assert rc == 0
|
||
assert "[云资产同步]" in out
|
||
for fragment in ("账号 0", "成功 0", "失败 0", "新增 0", "更新 0",
|
||
"跳过(无凭证) 0", "跳过(无适配器) 0", "跳过(面板核对) 0"):
|
||
assert fragment in out, f"摘要缺少 {fragment}：{out!r}"
|
||
|
||
|
||
def test_script_exits_zero_when_nothing_failed(script, monkeypatch):
|
||
fake = {
|
||
"total": 2, "success": 2, "failed": 0, "created": 3, "updated": 1,
|
||
"skipped_no_config": 0, "skipped_no_adapter": 0, "errors": [],
|
||
"synced_at": "2026-01-01T00:00:00",
|
||
}
|
||
monkeypatch.setattr(script.sync_service, "sync_all_assets", lambda s, **kw: fake)
|
||
assert script.main() == 0
|
||
|
||
|
||
def test_script_exits_nonzero_when_any_account_failed(script, capsys, monkeypatch):
|
||
"""退出码是唯一的告警通道：返回 0 的话 systemd 记成功，凭证过期也没人知道"""
|
||
fake = {
|
||
"total": 2, "success": 0, "failed": 2, "created": 0, "updated": 0,
|
||
"skipped_no_config": 0, "skipped_no_adapter": 0,
|
||
"errors": ["vultr/main: HTTP 401：凭证无效或已过期",
|
||
"aliyun/backup: HTTP 403：凭证权限不足"],
|
||
"synced_at": "2026-01-01T00:00:00",
|
||
}
|
||
monkeypatch.setattr(script.sync_service, "sync_all_assets", lambda s, **kw: fake)
|
||
assert script.main() == 1
|
||
out = capsys.readouterr().out
|
||
assert "失败 2" in out
|
||
for err in fake["errors"]:
|
||
assert err in out, f"错误明细没打出来，journalctl 里查不到原因：{err}"
|
||
|
||
|
||
def test_script_uses_session_bound_to_assets_engine(script, monkeypatch):
|
||
"""main() 传给服务层的 Session 必须开在 assets_engine 上
|
||
|
||
写错成 metrics_engine 的话，账号表查不到 -> total=0 -> 退出码 0，
|
||
表现是「同步一切正常，但一条数据都没进来」。
|
||
"""
|
||
seen = {}
|
||
|
||
def _fake(session, **kwargs):
|
||
seen["bind"] = session.get_bind()
|
||
return {
|
||
"total": 0, "success": 0, "failed": 0, "created": 0, "updated": 0,
|
||
"skipped_no_config": 0, "skipped_no_adapter": 0, "errors": [],
|
||
"synced_at": "2026-01-01T00:00:00",
|
||
}
|
||
|
||
monkeypatch.setattr(script.sync_service, "sync_all_assets", _fake)
|
||
script.main()
|
||
assert seen["bind"] is script.assets_engine
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 3. systemd 单元接线
|
||
# --------------------------------------------------------------------------- #
|
||
def _read(name):
|
||
return (DEPLOY_DIR / name).read_text(encoding="utf-8")
|
||
|
||
|
||
def test_sync_assets_service_points_at_an_existing_script():
|
||
"""ExecStart 指向的脚本必须真实存在
|
||
|
||
unit 里是部署机绝对路径 /opt/vps-manager/...，本地按仓库相对路径校验。
|
||
脚本改名后 timer 每晚会失败，但因为是 oneshot + 无人看日志，等同于静默停摆。
|
||
"""
|
||
m = re.search(r"^ExecStart=(?P<interp>\S+)\s+(?P<script>\S+)\s*$",
|
||
_read("vps-sync-assets.service"), re.M)
|
||
assert m, "vps-sync-assets.service 缺少 ExecStart"
|
||
assert m.group("interp").endswith("/.venv/bin/python"), "必须用虚拟环境解释器，系统 python 没有项目依赖"
|
||
rel = m.group("script").replace("/opt/vps-manager/", "")
|
||
assert (REPO_ROOT / rel).is_file(), f"ExecStart 指向的脚本不存在：{rel}"
|
||
|
||
|
||
def test_every_timer_points_at_an_existing_service():
|
||
for timer in sorted(DEPLOY_DIR.glob("*.timer")):
|
||
m = re.search(r"^Unit=(\S+)\s*$", timer.read_text(encoding="utf-8"), re.M)
|
||
assert m, f"{timer.name} 缺少 Unit="
|
||
assert (DEPLOY_DIR / m.group(1)).is_file(), f"{timer.name} 指向不存在的 {m.group(1)}"
|
||
|
||
|
||
def test_every_timer_is_enabled_by_setup_sh():
|
||
"""deploy/ 下每个 timer 都必须在 setup.sh 里 enable --now
|
||
|
||
只 cp 不 enable：单元文件躺在 /etc/systemd/system 里但从不触发，
|
||
全新部署后所有夜间任务都是死的。
|
||
"""
|
||
enable_lines = [ln for ln in _read("setup.sh").splitlines() if "enable --now" in ln]
|
||
assert enable_lines, "setup.sh 里找不到任何 enable --now"
|
||
for timer in sorted(p.name for p in DEPLOY_DIR.glob("*.timer")):
|
||
assert any(timer in ln for ln in enable_lines), f"{timer} 未在 setup.sh 中 enable --now"
|
||
|
||
|
||
def test_update_sh_enables_rather_than_restarts_new_timers():
|
||
"""自动更新同步单元时，timer 分支必须是 enable --now
|
||
|
||
新 timer 只 restart 不会建开机自启软链：当下能跑，宿主一重启就永久消失。
|
||
"""
|
||
text = _read("update.sh")
|
||
m = re.search(r"\*\.timer\)(?P<body>.*?);;", text, re.S)
|
||
assert m, "update.sh 没有 *.timer) 分支"
|
||
assert "enable --now" in m.group("body"), "timer 分支未使用 enable --now"
|
||
|
||
|
||
def _oncalendar(name):
|
||
m = re.search(r"^OnCalendar=\*-\*-\*\s+(\d{2}):(\d{2}):(\d{2})\s*$", _read(name), re.M)
|
||
assert m, f"{name} 的 OnCalendar 不是固定的每日 HH:MM:SS 格式"
|
||
return time(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||
|
||
|
||
def test_asset_sync_runs_before_renewal_check():
|
||
"""资产同步必须早于续费检查
|
||
|
||
续费提醒读的是本地库的 expiry_date。顺序反了，提醒就用上一天的旧到期日，
|
||
当天新到期的资产不会进通知——而这正是这套台账存在的理由。
|
||
"""
|
||
assert _oncalendar("vps-sync-assets.timer") < _oncalendar("vps-renewal-check.timer")
|
||
|
||
|
||
def test_sync_timers_are_persistent():
|
||
"""Persistent=true：机器在预定时刻关机时，开机后补跑一次
|
||
|
||
没有它，一次夜间重启就会整天跳过同步。
|
||
"""
|
||
for name in ("vps-sync-assets.timer", "vps-renewal-check.timer"):
|
||
assert re.search(r"^Persistent=true\s*$", _read(name), re.M), f"{name} 缺少 Persistent=true"
|
||
|
||
|
||
def test_sync_assets_service_is_oneshot_and_loads_env():
|
||
"""oneshot + EnvironmentFile：前者让 systemd 正确判定成败，
|
||
后者提供 MASTER_KEY（缺了就解密不了任何账号凭证，同步全部失败）"""
|
||
text = _read("vps-sync-assets.service")
|
||
assert re.search(r"^Type=oneshot\s*$", text, re.M)
|
||
assert re.search(r"^EnvironmentFile=-?\S*\.env\s*$", text, re.M), "未加载 .env，拿不到 MASTER_KEY"
|