94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""扫描 worker 的同步核心 _scan_and_alert:水位读写 / 冷启动 / 发送成功推进、失败不推进。
|
|
|
|
monkeypatch 掉真正的飞书发送;用真实 SQLite(SessionLocal)插入几条记录。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from app.core import compare_alert_worker as w
|
|
from app.db.session import SessionLocal
|
|
from app.models.app_config import AppConfig
|
|
from app.models.comparison import ComparisonRecord
|
|
|
|
WM_KEY = w.WATERMARK_KEY
|
|
|
|
|
|
@pytest.fixture()
|
|
def clean_db():
|
|
db = SessionLocal()
|
|
db.query(ComparisonRecord).delete()
|
|
db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
|
|
db.commit()
|
|
yield db
|
|
db.query(ComparisonRecord).delete()
|
|
db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
def _add(db, trace, status, **kw):
|
|
rec = ComparisonRecord(trace_id=trace, status=status, **kw)
|
|
db.add(rec)
|
|
db.commit()
|
|
db.refresh(rec)
|
|
return rec
|
|
|
|
|
|
def _card_title(card: dict) -> str:
|
|
return card.get("header", {}).get("title", {}).get("content", "")
|
|
|
|
|
|
def _card_md_content(card: dict) -> str:
|
|
elements = card.get("body", {}).get("elements", [])
|
|
return elements[0].get("content", "") if elements else ""
|
|
|
|
|
|
def test_cold_start_sets_watermark_no_alert(clean_db, monkeypatch):
|
|
# 冷启动:表非空 → 水位=当前 max(updated_at),不回溯已有历史失败、不报
|
|
_add(clean_db, "old-fail", "failed", fail_reason=None, information="比价过程出错")
|
|
sent = []
|
|
monkeypatch.setattr(w, "_send_card", lambda card: sent.append(card))
|
|
w._scan_and_alert()
|
|
assert sent == [] # 冷启动不报历史
|
|
row = clean_db.get(AppConfig, WM_KEY)
|
|
assert row is not None # 水位已初始化
|
|
|
|
|
|
def test_alerts_on_new_failed_and_advances(clean_db, monkeypatch):
|
|
# seed 一条 + 冷启动建水位;sleep 1.1s 拉开时间(SQLite 秒级精度,否则新记录同秒、追不上水位)
|
|
_add(clean_db, "seed", "running")
|
|
monkeypatch.setattr(w, "_send_card", lambda card: None)
|
|
w._scan_and_alert() # 冷启动,水位=seed.updated_at
|
|
time.sleep(1.1)
|
|
sent = []
|
|
monkeypatch.setattr(w, "_send_card", lambda card: sent.append(card))
|
|
_add(clean_db, "new-fail", "failed", fail_reason=None, information="比价过程出错")
|
|
w._scan_and_alert()
|
|
assert len(sent) == 1
|
|
card = sent[0]
|
|
# title 含关键词
|
|
assert "比价失败报警" in _card_title(card)
|
|
# markdown 摘要含"系统技术失败"
|
|
assert "系统技术失败 1" in _card_md_content(card)
|
|
|
|
|
|
def test_send_failure_does_not_advance_watermark(clean_db, monkeypatch):
|
|
_add(clean_db, "seed2", "running")
|
|
monkeypatch.setattr(w, "_send_card", lambda card: None)
|
|
w._scan_and_alert() # 冷启动建水位
|
|
wm_before = clean_db.get(AppConfig, WM_KEY).value
|
|
time.sleep(1.1)
|
|
_add(clean_db, "fail-send", "failed", fail_reason=None, information="比价过程出错")
|
|
|
|
def boom(card):
|
|
raise w.feishu_notifier.FeishuNotifyError("down")
|
|
|
|
monkeypatch.setattr(w, "_send_card", boom)
|
|
w._scan_and_alert() # 发送失败
|
|
clean_db.expire_all()
|
|
wm_after = clean_db.get(AppConfig, WM_KEY).value
|
|
assert wm_after == wm_before # 未推进,下轮补发
|