115 lines
5.1 KiB
Python
115 lines
5.1 KiB
Python
"""提醒状态机测试：低于阈值提醒一次、恢复后重新武装、发送失败不切换状态。"""
|
|
|
|
import pytest
|
|
|
|
from app import db
|
|
from app.alert import build_alert_message, evaluate, notify_disabled
|
|
|
|
ACCOUNT = {
|
|
"id": 1, "name": "主账号", "threshold": 20.0, "alert_armed": 1,
|
|
"last_check_at": "2026-08-04 10:00:00",
|
|
}
|
|
PLATFORM = {"name": "OpenAI", "currency": "USD"}
|
|
|
|
|
|
def _insert_account(test_db, armed=1, enabled=1):
|
|
with db.get_conn() as conn:
|
|
conn.execute("INSERT INTO platforms (id, name, url, balance_path) VALUES (1, 'OpenAI', 'http://x', 'b')")
|
|
conn.execute(
|
|
"""INSERT INTO accounts (id, platform_id, name, api_key, threshold, enabled, alert_armed)
|
|
VALUES (1, 1, '主账号', 'a2tva2V5', 20.0, ?, ?)""",
|
|
(enabled, armed),
|
|
)
|
|
|
|
|
|
def _armed(test_db):
|
|
with db.get_conn() as conn:
|
|
return conn.execute("SELECT alert_armed FROM accounts WHERE id=1").fetchone()[0]
|
|
|
|
|
|
def _log(test_db):
|
|
with db.get_conn() as conn:
|
|
return conn.execute("SELECT type FROM alert_log WHERE account_id=1 ORDER BY id").fetchall()
|
|
|
|
|
|
class TestEvaluate:
|
|
def test_below_threshold_armed_sends_and_disarms(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=1)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(a) or True)
|
|
evaluate(ACCOUNT, PLATFORM, 12.5, type("Cfg", (), {"telegram_bot_token": "t", "telegram_chat_id": "c"})())
|
|
assert len(sent) == 1
|
|
assert "12.5" in sent[0][2] and "OpenAI" in sent[0][2]
|
|
assert _armed(test_db) == 0
|
|
assert [r[0] for r in _log(test_db)] == ["below"]
|
|
|
|
def test_still_below_no_second_alert(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=0)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(1) or True)
|
|
evaluate(dict(ACCOUNT, alert_armed=0), PLATFORM, 5.0, None)
|
|
assert sent == []
|
|
|
|
def test_recover_after_disarmed(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=0)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(a[2]) or True)
|
|
evaluate(dict(ACCOUNT, alert_armed=0), PLATFORM, 30.0, type("Cfg", (), {"telegram_bot_token": "t", "telegram_chat_id": "c"})())
|
|
assert len(sent) == 1 and "恢复" in sent[0]
|
|
assert _armed(test_db) == 1
|
|
|
|
def test_above_threshold_armed_no_alert(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=1)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(1) or True)
|
|
evaluate(ACCOUNT, PLATFORM, 99.0, None)
|
|
assert sent == []
|
|
assert _armed(test_db) == 1
|
|
|
|
def test_exactly_threshold_is_not_below(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=1)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(1) or True)
|
|
evaluate(ACCOUNT, PLATFORM, 20.0, None)
|
|
assert sent == []
|
|
|
|
def test_send_failure_keeps_armed(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=1)
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: False)
|
|
evaluate(ACCOUNT, PLATFORM, 5.0, type("Cfg", (), {"telegram_bot_token": "t", "telegram_chat_id": "c"})())
|
|
assert _armed(test_db) == 1 # 发送失败 → 保持武装，下次再试
|
|
assert _log(test_db) == []
|
|
|
|
def test_full_cycle(self, test_db, monkeypatch):
|
|
_insert_account(test_db, armed=1)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(a[2]) or True)
|
|
cfg = type("Cfg", (), {"telegram_bot_token": "t", "telegram_chat_id": "c"})()
|
|
evaluate(ACCOUNT, PLATFORM, 8.0, cfg)
|
|
evaluate(dict(ACCOUNT, alert_armed=0), PLATFORM, 8.0, cfg) # 仍低于 → 不再发
|
|
evaluate(dict(ACCOUNT, alert_armed=0), PLATFORM, 50.0, cfg) # 恢复 → 发恢复
|
|
evaluate(dict(ACCOUNT, alert_armed=1), PLATFORM, 60.0, cfg) # 正常 → 不发
|
|
evaluate(dict(ACCOUNT, alert_armed=1), PLATFORM, 3.0, cfg) # 又低于 → 再提醒
|
|
assert len(sent) == 3
|
|
assert [r[0] for r in _log(test_db)] == ["below", "recovered", "below"]
|
|
|
|
|
|
class TestNotifyDisabled:
|
|
def test_disabled_notification(self, test_db, monkeypatch):
|
|
_insert_account(test_db)
|
|
sent = []
|
|
monkeypatch.setattr("app.alert.send_message", lambda *a: sent.append(a[2]) or True)
|
|
notify_disabled(dict(ACCOUNT, last_error="HTTP 401（认证失败）"), PLATFORM,
|
|
type("Cfg", (), {"telegram_bot_token": "t", "telegram_chat_id": "c"})())
|
|
assert len(sent) == 1 and "禁用" in sent[0] and "401" in sent[0]
|
|
assert [r[0] for r in _log(test_db)] == ["disabled"]
|
|
|
|
|
|
class TestBuildMessage:
|
|
def test_below_message_content(self):
|
|
msg = build_alert_message("below", ACCOUNT, PLATFORM, 12.5)
|
|
assert "12.5 USD" in msg and "20 USD" in msg and "主账号" in msg
|
|
|
|
def test_unknown_kind_empty(self):
|
|
assert build_alert_message("nope", ACCOUNT, PLATFORM, 1) == ""
|