dekun
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
import os
|
|
|
|
from packages.config.settings import get_settings
|
|
from packages.notify import wecom
|
|
|
|
|
|
def _with_env(**kwargs):
|
|
old = {}
|
|
for k, v in kwargs.items():
|
|
old[k] = os.environ.get(k)
|
|
os.environ[k] = v
|
|
get_settings.cache_clear()
|
|
return old
|
|
|
|
|
|
def _restore(old: dict):
|
|
for k, v in old.items():
|
|
if v is None:
|
|
os.environ.pop(k, None)
|
|
else:
|
|
os.environ[k] = v
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_build_markdown_contains_tag():
|
|
md = wecom.build_markdown(tag="FAULT", title="测试", lines=["a", "b"])
|
|
assert "`FAULT`" in md
|
|
assert "比特骆驼行情采集分析" in md
|
|
assert "测试" in md
|
|
|
|
|
|
def test_fault_below_threshold():
|
|
wecom.reset_alert_state()
|
|
old = _with_env(WECOM_ENABLED="0", ALERT_FAIL_THRESHOLD="5")
|
|
try:
|
|
ok, msg = wecom.notify_collector_fault(error="x", consecutive_failures=2)
|
|
assert ok is False
|
|
assert "below threshold" in msg
|
|
finally:
|
|
_restore(old)
|
|
wecom.reset_alert_state()
|
|
|
|
|
|
def test_fault_when_disabled():
|
|
wecom.reset_alert_state()
|
|
old = _with_env(WECOM_ENABLED="0", ALERT_FAIL_THRESHOLD="3")
|
|
try:
|
|
ok, msg = wecom.notify_collector_fault(error="boom", consecutive_failures=3)
|
|
assert ok is False
|
|
assert "未开启" in msg
|
|
ok2, msg2 = wecom.notify_collector_fault(error="boom", consecutive_failures=3)
|
|
assert ok2 is False
|
|
assert msg2 == "dedup"
|
|
finally:
|
|
_restore(old)
|
|
wecom.reset_alert_state()
|