Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion protocols/timelock/timelock_alerts.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,16 @@
# TELEGRAM_BOT_TOKEN_YEARN_TIMELOCK_INTERNAL / TELEGRAM_CHAT_ID_YEARN_TIMELOCK_INTERNAL.
YEARN_TIMELOCK_INTERNAL_PROTOCOL = "YEARN_TIMELOCK_INTERNAL"

# TimelockConfig.protocol is the Telegram routing key (TELEGRAM_TOPIC_ID_<KEY>),
# but the alert store must use the protocol key the website queries
# (`GET /v1/alerts?protocol=<key>`, exact match) or the alert never shows on the
# protocol's page. Keys default to the lowercased routing key; list exceptions here.
ALERT_HISTORY_PROTOCOLS: dict[str, str] = {
"YEARN_TIMELOCK": "yearn",
"RTOKEN": "ethplus",
"LRT": "pegs",
}


@dataclass(frozen=True)
class TimelockConfig:
Expand Down Expand Up @@ -86,6 +96,11 @@ class TimelockConfig:
_logger = get_logger("timelock_alerts")


def alert_history_protocol(routing_protocol: str) -> str:
"""Return the protocol key the alert store and website use for a routing key."""
return ALERT_HISTORY_PROTOCOLS.get(routing_protocol, routing_protocol.lower())


def http_json(url: str, method: str = "GET", body: dict | None = None, headers: dict | None = None) -> dict | None:
"""Make an HTTP request and return JSON response."""
_logger.info("http_json %s %s", method, url)
Expand Down Expand Up @@ -571,9 +586,12 @@ def process_events(events: list[dict], use_cache: bool) -> None:
if current_parts:
chunks.append(separator.join(current_parts))

# The internal mirror keeps its own key so the website doesn't list every
# Yearn timelock alert twice.
origin_protocol = None if protocol == YEARN_TIMELOCK_INTERNAL_PROTOCOL else alert_history_protocol(protocol)
for chunk in chunks:
try:
send_telegram_message(chunk, protocol)
send_telegram_message(chunk, protocol, origin_protocol=origin_protocol)
except Exception:
_logger.exception("Failed to send Telegram alert for protocol %s", protocol)
all_sent = False
Expand Down
31 changes: 30 additions & 1 deletion tests/test_timelock_alerts.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,18 @@
import unittest.mock
from unittest.mock import patch

from protocols.timelock.timelock_alerts import TIMELOCKS, TimelockConfig, build_alert_message
from protocols.timelock.timelock_alerts import (
TIMELOCKS,
YEARN_TIMELOCK_INTERNAL_PROTOCOL,
TimelockConfig,
alert_history_protocol,
build_alert_message,
process_events,
)
from utils.telegram import MAX_MESSAGE_LENGTH

YEARN_TIMELOCK_ADDRESS = "0x88ba032be87d5ef1fbe87336b7090767f367bf73"


def test_3jane_seven_day_timelock_is_monitored() -> None:
timelock = TIMELOCKS[("0x3d3c41419ab401cd25055e8f9421d7d96d887885", 1)]
Expand All @@ -15,6 +24,26 @@ def test_3jane_seven_day_timelock_is_monitored() -> None:
assert timelock.label == "3Jane 7d TimelockController"


def test_alert_history_protocol_matches_website_keys() -> None:
assert alert_history_protocol("YEARN_TIMELOCK") == "yearn"
assert alert_history_protocol("RTOKEN") == "ethplus"
assert alert_history_protocol("COMP") == "comp"
assert alert_history_protocol("3JANE") == "3jane"
assert alert_history_protocol("CAP") == "cap"


@patch("protocols.timelock.timelock_alerts.send_telegram_message")
@patch("protocols.timelock.timelock_alerts.build_alert_message", return_value="msg")
def test_yearn_timelock_alert_recorded_under_yearn(_mock_build: object, mock_send: unittest.mock.MagicMock) -> None:
"""Routing stays on YEARN_TIMELOCK, but the stored protocol must be `yearn` for the website."""
event = _make_event(id="1", timelockAddress=YEARN_TIMELOCK_ADDRESS)

process_events([event], use_cache=False)

calls = {c.args[1]: c.kwargs["origin_protocol"] for c in mock_send.call_args_list}
assert calls == {"YEARN_TIMELOCK": "yearn", YEARN_TIMELOCK_INTERNAL_PROTOCOL: None}


def _make_event(
timelock_type: str = "TimelockController",
chain_id: int = 1,
Expand Down