Merge pull request #40 from pselamy/fix/21
feat: add Discord and Telegram webhook dispatcher
This commit is contained in:
Patrick Selamy
@@ -1,9 +1,23 @@
|
||||
"""Alerting layer - Real-time notification delivery."""
|
||||
|
||||
from polymarket_insider_tracker.alerter.channels.discord import DiscordChannel
|
||||
from polymarket_insider_tracker.alerter.channels.telegram import TelegramChannel
|
||||
from polymarket_insider_tracker.alerter.dispatcher import (
|
||||
AlertChannel,
|
||||
AlertDispatcher,
|
||||
CircuitBreakerState,
|
||||
DispatchResult,
|
||||
)
|
||||
from polymarket_insider_tracker.alerter.formatter import AlertFormatter
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
__all__ = [
|
||||
"AlertChannel",
|
||||
"AlertDispatcher",
|
||||
"AlertFormatter",
|
||||
"CircuitBreakerState",
|
||||
"DiscordChannel",
|
||||
"DispatchResult",
|
||||
"FormattedAlert",
|
||||
"TelegramChannel",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Alert channel implementations for various platforms."""
|
||||
|
||||
from polymarket_insider_tracker.alerter.channels.discord import DiscordChannel
|
||||
from polymarket_insider_tracker.alerter.channels.telegram import TelegramChannel
|
||||
|
||||
__all__ = [
|
||||
"DiscordChannel",
|
||||
"TelegramChannel",
|
||||
]
|
||||
@@ -0,0 +1,118 @@
|
||||
"""Discord webhook channel implementation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import httpx
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DiscordChannel:
|
||||
"""Discord webhook channel for sending alerts.
|
||||
|
||||
Sends formatted alerts to Discord via webhook URL with rate limiting
|
||||
and retry support.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
webhook_url: str,
|
||||
*,
|
||||
rate_limit_per_minute: int = 30,
|
||||
max_retries: int = 3,
|
||||
retry_delay: float = 1.0,
|
||||
timeout: float = 10.0,
|
||||
) -> None:
|
||||
"""Initialize Discord channel.
|
||||
|
||||
Args:
|
||||
webhook_url: Discord webhook URL.
|
||||
rate_limit_per_minute: Maximum messages per minute (Discord limit is 30).
|
||||
max_retries: Maximum retry attempts on failure.
|
||||
retry_delay: Base delay between retries (exponential backoff).
|
||||
timeout: HTTP request timeout in seconds.
|
||||
"""
|
||||
self.webhook_url = webhook_url
|
||||
self.rate_limit_per_minute = rate_limit_per_minute
|
||||
self.max_retries = max_retries
|
||||
self.retry_delay = retry_delay
|
||||
self.timeout = timeout
|
||||
self.name = "discord"
|
||||
|
||||
# Rate limiting state
|
||||
self._request_times: list[float] = []
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def _wait_for_rate_limit(self) -> None:
|
||||
"""Wait if rate limit is exceeded."""
|
||||
async with self._lock:
|
||||
now = asyncio.get_event_loop().time()
|
||||
# Remove requests older than 1 minute
|
||||
self._request_times = [t for t in self._request_times if now - t < 60]
|
||||
|
||||
if len(self._request_times) >= self.rate_limit_per_minute:
|
||||
# Wait until the oldest request expires
|
||||
wait_time = 60 - (now - self._request_times[0])
|
||||
if wait_time > 0:
|
||||
logger.debug(f"Discord rate limit hit, waiting {wait_time:.2f}s")
|
||||
await asyncio.sleep(wait_time)
|
||||
|
||||
self._request_times.append(now)
|
||||
|
||||
async def send(self, alert: FormattedAlert) -> bool:
|
||||
"""Send alert to Discord webhook.
|
||||
|
||||
Args:
|
||||
alert: Formatted alert with discord_embed.
|
||||
|
||||
Returns:
|
||||
True if delivery succeeded, False otherwise.
|
||||
"""
|
||||
await self._wait_for_rate_limit()
|
||||
|
||||
payload = {
|
||||
"embeds": [alert.discord_embed],
|
||||
}
|
||||
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.post(
|
||||
self.webhook_url,
|
||||
json=payload,
|
||||
)
|
||||
|
||||
if response.status_code == 204:
|
||||
logger.info("Discord alert delivered successfully")
|
||||
return True
|
||||
|
||||
if response.status_code == 429:
|
||||
# Rate limited by Discord
|
||||
retry_after = response.json().get("retry_after", 1.0)
|
||||
logger.warning(f"Discord rate limited, retry after {retry_after}s")
|
||||
await asyncio.sleep(retry_after)
|
||||
continue
|
||||
|
||||
logger.error(
|
||||
f"Discord webhook failed: {response.status_code} {response.text}"
|
||||
)
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.warning(f"Discord webhook timeout (attempt {attempt + 1})")
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Discord webhook error: {e}")
|
||||
|
||||
# Exponential backoff
|
||||
if attempt < self.max_retries - 1:
|
||||
delay = self.retry_delay * (2**attempt)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
logger.error("Discord delivery failed after all retries")
|
||||
return False
|
||||
@@ -0,0 +1,131 @@
|
||||
"""Telegram Bot API channel implementation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import httpx
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TELEGRAM_API_BASE = "https://api.telegram.org/bot{token}/sendMessage"
|
||||
|
||||
|
||||
class TelegramChannel:
|
||||
"""Telegram Bot API channel for sending alerts.
|
||||
|
||||
Sends formatted alerts to Telegram via Bot API with rate limiting
|
||||
and retry support.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
bot_token: str,
|
||||
chat_id: str,
|
||||
*,
|
||||
rate_limit_per_minute: int = 20,
|
||||
max_retries: int = 3,
|
||||
retry_delay: float = 1.0,
|
||||
timeout: float = 10.0,
|
||||
) -> None:
|
||||
"""Initialize Telegram channel.
|
||||
|
||||
Args:
|
||||
bot_token: Telegram bot token.
|
||||
chat_id: Target chat/channel ID.
|
||||
rate_limit_per_minute: Maximum messages per minute.
|
||||
max_retries: Maximum retry attempts on failure.
|
||||
retry_delay: Base delay between retries (exponential backoff).
|
||||
timeout: HTTP request timeout in seconds.
|
||||
"""
|
||||
self.bot_token = bot_token
|
||||
self.chat_id = chat_id
|
||||
self.rate_limit_per_minute = rate_limit_per_minute
|
||||
self.max_retries = max_retries
|
||||
self.retry_delay = retry_delay
|
||||
self.timeout = timeout
|
||||
self.name = "telegram"
|
||||
|
||||
self._api_url = TELEGRAM_API_BASE.format(token=bot_token)
|
||||
|
||||
# Rate limiting state
|
||||
self._request_times: list[float] = []
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def _wait_for_rate_limit(self) -> None:
|
||||
"""Wait if rate limit is exceeded."""
|
||||
async with self._lock:
|
||||
now = asyncio.get_event_loop().time()
|
||||
# Remove requests older than 1 minute
|
||||
self._request_times = [t for t in self._request_times if now - t < 60]
|
||||
|
||||
if len(self._request_times) >= self.rate_limit_per_minute:
|
||||
# Wait until the oldest request expires
|
||||
wait_time = 60 - (now - self._request_times[0])
|
||||
if wait_time > 0:
|
||||
logger.debug(f"Telegram rate limit hit, waiting {wait_time:.2f}s")
|
||||
await asyncio.sleep(wait_time)
|
||||
|
||||
self._request_times.append(now)
|
||||
|
||||
async def send(self, alert: FormattedAlert) -> bool:
|
||||
"""Send alert to Telegram channel.
|
||||
|
||||
Args:
|
||||
alert: Formatted alert with telegram_markdown.
|
||||
|
||||
Returns:
|
||||
True if delivery succeeded, False otherwise.
|
||||
"""
|
||||
await self._wait_for_rate_limit()
|
||||
|
||||
payload = {
|
||||
"chat_id": self.chat_id,
|
||||
"text": alert.telegram_markdown,
|
||||
"parse_mode": "MarkdownV2",
|
||||
"disable_web_page_preview": False,
|
||||
}
|
||||
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.post(
|
||||
self._api_url,
|
||||
json=payload,
|
||||
)
|
||||
|
||||
result = response.json()
|
||||
|
||||
if result.get("ok"):
|
||||
logger.info("Telegram alert delivered successfully")
|
||||
return True
|
||||
|
||||
error_code = result.get("error_code", 0)
|
||||
description = result.get("description", "Unknown error")
|
||||
|
||||
if error_code == 429:
|
||||
# Rate limited
|
||||
retry_after = result.get("parameters", {}).get("retry_after", 1)
|
||||
logger.warning(f"Telegram rate limited, retry after {retry_after}s")
|
||||
await asyncio.sleep(retry_after)
|
||||
continue
|
||||
|
||||
logger.error(f"Telegram API error: {error_code} - {description}")
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.warning(f"Telegram API timeout (attempt {attempt + 1})")
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Telegram API error: {e}")
|
||||
|
||||
# Exponential backoff
|
||||
if attempt < self.max_retries - 1:
|
||||
delay = self.retry_delay * (2**attempt)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
logger.error("Telegram delivery failed after all retries")
|
||||
return False
|
||||
@@ -0,0 +1,239 @@
|
||||
"""Alert dispatcher for multi-channel delivery."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Protocol
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AlertChannel(Protocol):
|
||||
"""Protocol for alert delivery channels."""
|
||||
|
||||
name: str
|
||||
|
||||
async def send(self, alert: FormattedAlert) -> bool:
|
||||
"""Send alert to channel. Returns True on success."""
|
||||
...
|
||||
|
||||
|
||||
@dataclass
|
||||
class CircuitBreakerState:
|
||||
"""State for circuit breaker pattern.
|
||||
|
||||
Tracks failures and manages open/closed state for a channel.
|
||||
"""
|
||||
|
||||
failure_count: int = 0
|
||||
last_failure_time: datetime | None = None
|
||||
is_open: bool = False
|
||||
half_open_attempts: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class DispatchResult:
|
||||
"""Result of dispatching an alert to all channels."""
|
||||
|
||||
success_count: int
|
||||
failure_count: int
|
||||
channel_results: dict[str, bool] = field(default_factory=dict)
|
||||
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
|
||||
@property
|
||||
def all_succeeded(self) -> bool:
|
||||
"""Return True if all channels succeeded."""
|
||||
return self.failure_count == 0 and self.success_count > 0
|
||||
|
||||
|
||||
class AlertDispatcher:
|
||||
"""Dispatcher for sending alerts to multiple channels.
|
||||
|
||||
Manages concurrent delivery to all configured channels with
|
||||
circuit breaker protection for failing channels.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
channels: list[AlertChannel],
|
||||
*,
|
||||
failure_threshold: int = 5,
|
||||
recovery_timeout_seconds: int = 60,
|
||||
half_open_max_attempts: int = 3,
|
||||
) -> None:
|
||||
"""Initialize the dispatcher.
|
||||
|
||||
Args:
|
||||
channels: List of alert channels to dispatch to.
|
||||
failure_threshold: Number of consecutive failures before opening circuit.
|
||||
recovery_timeout_seconds: Time to wait before half-opening circuit.
|
||||
half_open_max_attempts: Number of test attempts in half-open state.
|
||||
"""
|
||||
self.channels = channels
|
||||
self.failure_threshold = failure_threshold
|
||||
self.recovery_timeout_seconds = recovery_timeout_seconds
|
||||
self.half_open_max_attempts = half_open_max_attempts
|
||||
|
||||
# Circuit breaker state per channel
|
||||
self._circuit_state: dict[str, CircuitBreakerState] = {
|
||||
ch.name: CircuitBreakerState() for ch in channels
|
||||
}
|
||||
|
||||
def _should_attempt(self, channel_name: str) -> bool:
|
||||
"""Check if we should attempt delivery to this channel."""
|
||||
state = self._circuit_state[channel_name]
|
||||
|
||||
if not state.is_open:
|
||||
return True
|
||||
|
||||
# Check if we should try half-open
|
||||
if state.last_failure_time:
|
||||
elapsed = (datetime.now(UTC) - state.last_failure_time).total_seconds()
|
||||
if (
|
||||
elapsed >= self.recovery_timeout_seconds
|
||||
and state.half_open_attempts < self.half_open_max_attempts
|
||||
):
|
||||
# Allow half-open attempt
|
||||
logger.info(
|
||||
f"Circuit half-open for {channel_name}, "
|
||||
f"attempt {state.half_open_attempts + 1}"
|
||||
)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _record_success(self, channel_name: str) -> None:
|
||||
"""Record a successful delivery."""
|
||||
state = self._circuit_state[channel_name]
|
||||
state.failure_count = 0
|
||||
state.is_open = False
|
||||
state.half_open_attempts = 0
|
||||
state.last_failure_time = None
|
||||
logger.debug(f"Circuit closed for {channel_name}")
|
||||
|
||||
def _record_failure(self, channel_name: str) -> None:
|
||||
"""Record a failed delivery."""
|
||||
state = self._circuit_state[channel_name]
|
||||
state.failure_count += 1
|
||||
state.last_failure_time = datetime.now(UTC)
|
||||
|
||||
if state.is_open:
|
||||
# Failed during half-open, increment attempts
|
||||
state.half_open_attempts += 1
|
||||
elif state.failure_count >= self.failure_threshold:
|
||||
# Open the circuit
|
||||
state.is_open = True
|
||||
logger.warning(
|
||||
f"Circuit opened for {channel_name} after "
|
||||
f"{state.failure_count} failures"
|
||||
)
|
||||
|
||||
async def _send_to_channel(
|
||||
self, channel: AlertChannel, alert: FormattedAlert
|
||||
) -> tuple[str, bool]:
|
||||
"""Send alert to a single channel with circuit breaker."""
|
||||
channel_name = channel.name
|
||||
|
||||
if not self._should_attempt(channel_name):
|
||||
logger.debug(f"Skipping {channel_name} - circuit open")
|
||||
return (channel_name, False)
|
||||
|
||||
try:
|
||||
success = await channel.send(alert)
|
||||
if success:
|
||||
self._record_success(channel_name)
|
||||
else:
|
||||
self._record_failure(channel_name)
|
||||
return (channel_name, success)
|
||||
except Exception as e:
|
||||
logger.error(f"Error sending to {channel_name}: {e}")
|
||||
self._record_failure(channel_name)
|
||||
return (channel_name, False)
|
||||
|
||||
async def dispatch(self, alert: FormattedAlert) -> DispatchResult:
|
||||
"""Dispatch alert to all channels concurrently.
|
||||
|
||||
Args:
|
||||
alert: Formatted alert to send.
|
||||
|
||||
Returns:
|
||||
DispatchResult with per-channel status.
|
||||
"""
|
||||
if not self.channels:
|
||||
logger.warning("No channels configured for dispatch")
|
||||
return DispatchResult(success_count=0, failure_count=0)
|
||||
|
||||
# Send to all channels concurrently
|
||||
tasks = [self._send_to_channel(ch, alert) for ch in self.channels]
|
||||
results = await asyncio.gather(*tasks)
|
||||
|
||||
# Aggregate results
|
||||
channel_results = dict(results)
|
||||
success_count = sum(1 for success in channel_results.values() if success)
|
||||
failure_count = len(channel_results) - success_count
|
||||
|
||||
result = DispatchResult(
|
||||
success_count=success_count,
|
||||
failure_count=failure_count,
|
||||
channel_results=channel_results,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Dispatch complete: {success_count}/{len(channel_results)} succeeded"
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
async def dispatch_batch(
|
||||
self, alerts: list[FormattedAlert]
|
||||
) -> list[DispatchResult]:
|
||||
"""Dispatch multiple alerts sequentially.
|
||||
|
||||
Args:
|
||||
alerts: List of formatted alerts to send.
|
||||
|
||||
Returns:
|
||||
List of DispatchResult for each alert.
|
||||
"""
|
||||
results = []
|
||||
for alert in alerts:
|
||||
result = await self.dispatch(alert)
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
def get_circuit_status(self) -> dict[str, dict[str, object]]:
|
||||
"""Get current circuit breaker status for all channels."""
|
||||
return {
|
||||
name: {
|
||||
"is_open": state.is_open,
|
||||
"failure_count": state.failure_count,
|
||||
"half_open_attempts": state.half_open_attempts,
|
||||
"last_failure": (
|
||||
state.last_failure_time.isoformat()
|
||||
if state.last_failure_time
|
||||
else None
|
||||
),
|
||||
}
|
||||
for name, state in self._circuit_state.items()
|
||||
}
|
||||
|
||||
def reset_circuit(self, channel_name: str) -> bool:
|
||||
"""Manually reset circuit breaker for a channel.
|
||||
|
||||
Args:
|
||||
channel_name: Name of channel to reset.
|
||||
|
||||
Returns:
|
||||
True if channel was found and reset.
|
||||
"""
|
||||
if channel_name in self._circuit_state:
|
||||
self._circuit_state[channel_name] = CircuitBreakerState()
|
||||
logger.info(f"Circuit reset for {channel_name}")
|
||||
return True
|
||||
return False
|
||||
@@ -0,0 +1,506 @@
|
||||
"""Tests for alert dispatcher and channels."""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from polymarket_insider_tracker.alerter.channels.discord import DiscordChannel
|
||||
from polymarket_insider_tracker.alerter.channels.telegram import TelegramChannel
|
||||
from polymarket_insider_tracker.alerter.dispatcher import (
|
||||
AlertDispatcher,
|
||||
CircuitBreakerState,
|
||||
DispatchResult,
|
||||
)
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
# ============================================================================
|
||||
# Fixtures
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_alert() -> FormattedAlert:
|
||||
"""Create a sample formatted alert."""
|
||||
return FormattedAlert(
|
||||
title="Test Alert",
|
||||
body="Test body",
|
||||
discord_embed={
|
||||
"title": "Test",
|
||||
"color": 15158332,
|
||||
"fields": [],
|
||||
},
|
||||
telegram_markdown="*Test Alert*\nTest body",
|
||||
plain_text="TEST ALERT\nTest body",
|
||||
links={"market": "https://polymarket.com/test"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_discord_channel() -> MagicMock:
|
||||
"""Create a mock Discord channel."""
|
||||
channel = MagicMock()
|
||||
channel.name = "discord"
|
||||
channel.send = AsyncMock(return_value=True)
|
||||
return channel
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_telegram_channel() -> MagicMock:
|
||||
"""Create a mock Telegram channel."""
|
||||
channel = MagicMock()
|
||||
channel.name = "telegram"
|
||||
channel.send = AsyncMock(return_value=True)
|
||||
return channel
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# DiscordChannel Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestDiscordChannel:
|
||||
"""Tests for Discord channel."""
|
||||
|
||||
def test_init(self) -> None:
|
||||
"""Test channel initialization."""
|
||||
channel = DiscordChannel(
|
||||
webhook_url="https://discord.com/api/webhooks/123/abc",
|
||||
rate_limit_per_minute=30,
|
||||
)
|
||||
assert channel.webhook_url == "https://discord.com/api/webhooks/123/abc"
|
||||
assert channel.rate_limit_per_minute == 30
|
||||
assert channel.name == "discord"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_success(self, sample_alert: FormattedAlert) -> None:
|
||||
"""Test successful Discord message send."""
|
||||
channel = DiscordChannel(
|
||||
webhook_url="https://discord.com/api/webhooks/123/abc"
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 204
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await channel.send(sample_alert)
|
||||
|
||||
assert result is True
|
||||
mock_client.post.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_rate_limited(self, sample_alert: FormattedAlert) -> None:
|
||||
"""Test Discord rate limit handling."""
|
||||
channel = DiscordChannel(
|
||||
webhook_url="https://discord.com/api/webhooks/123/abc",
|
||||
max_retries=2,
|
||||
retry_delay=0.01,
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response_429 = MagicMock()
|
||||
mock_response_429.status_code = 429
|
||||
mock_response_429.json.return_value = {"retry_after": 0.01}
|
||||
|
||||
mock_response_success = MagicMock()
|
||||
mock_response_success.status_code = 204
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.side_effect = [mock_response_429, mock_response_success]
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await channel.send(sample_alert)
|
||||
|
||||
assert result is True
|
||||
assert mock_client.post.call_count == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_failure(self, sample_alert: FormattedAlert) -> None:
|
||||
"""Test Discord send failure after retries."""
|
||||
channel = DiscordChannel(
|
||||
webhook_url="https://discord.com/api/webhooks/123/abc",
|
||||
max_retries=2,
|
||||
retry_delay=0.01,
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 500
|
||||
mock_response.text = "Internal Server Error"
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await channel.send(sample_alert)
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TelegramChannel Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestTelegramChannel:
|
||||
"""Tests for Telegram channel."""
|
||||
|
||||
def test_init(self) -> None:
|
||||
"""Test channel initialization."""
|
||||
channel = TelegramChannel(
|
||||
bot_token="123456:ABC-DEF",
|
||||
chat_id="-1001234567890",
|
||||
rate_limit_per_minute=20,
|
||||
)
|
||||
assert channel.bot_token == "123456:ABC-DEF"
|
||||
assert channel.chat_id == "-1001234567890"
|
||||
assert channel.name == "telegram"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_success(self, sample_alert: FormattedAlert) -> None:
|
||||
"""Test successful Telegram message send."""
|
||||
channel = TelegramChannel(
|
||||
bot_token="123456:ABC-DEF",
|
||||
chat_id="-1001234567890",
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"ok": True}
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await channel.send(sample_alert)
|
||||
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_rate_limited(self, sample_alert: FormattedAlert) -> None:
|
||||
"""Test Telegram rate limit handling."""
|
||||
channel = TelegramChannel(
|
||||
bot_token="123456:ABC-DEF",
|
||||
chat_id="-1001234567890",
|
||||
max_retries=2,
|
||||
retry_delay=0.01,
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response_429 = MagicMock()
|
||||
mock_response_429.json.return_value = {
|
||||
"ok": False,
|
||||
"error_code": 429,
|
||||
"parameters": {"retry_after": 0.01},
|
||||
}
|
||||
|
||||
mock_response_success = MagicMock()
|
||||
mock_response_success.json.return_value = {"ok": True}
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.side_effect = [mock_response_429, mock_response_success]
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await channel.send(sample_alert)
|
||||
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_failure(self, sample_alert: FormattedAlert) -> None:
|
||||
"""Test Telegram send failure."""
|
||||
channel = TelegramChannel(
|
||||
bot_token="123456:ABC-DEF",
|
||||
chat_id="-1001234567890",
|
||||
max_retries=2,
|
||||
retry_delay=0.01,
|
||||
)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client_class:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {
|
||||
"ok": False,
|
||||
"error_code": 400,
|
||||
"description": "Bad Request",
|
||||
}
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.return_value = mock_response
|
||||
mock_client.__aenter__.return_value = mock_client
|
||||
mock_client.__aexit__.return_value = None
|
||||
mock_client_class.return_value = mock_client
|
||||
|
||||
result = await channel.send(sample_alert)
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# CircuitBreakerState Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestCircuitBreakerState:
|
||||
"""Tests for circuit breaker state."""
|
||||
|
||||
def test_default_state(self) -> None:
|
||||
"""Test default circuit breaker state."""
|
||||
state = CircuitBreakerState()
|
||||
assert state.failure_count == 0
|
||||
assert state.is_open is False
|
||||
assert state.half_open_attempts == 0
|
||||
assert state.last_failure_time is None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# DispatchResult Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestDispatchResult:
|
||||
"""Tests for dispatch result."""
|
||||
|
||||
def test_all_succeeded(self) -> None:
|
||||
"""Test all_succeeded property."""
|
||||
result = DispatchResult(
|
||||
success_count=2,
|
||||
failure_count=0,
|
||||
channel_results={"discord": True, "telegram": True},
|
||||
)
|
||||
assert result.all_succeeded is True
|
||||
|
||||
def test_partial_success(self) -> None:
|
||||
"""Test partial success."""
|
||||
result = DispatchResult(
|
||||
success_count=1,
|
||||
failure_count=1,
|
||||
channel_results={"discord": True, "telegram": False},
|
||||
)
|
||||
assert result.all_succeeded is False
|
||||
|
||||
def test_empty_channels(self) -> None:
|
||||
"""Test with no channels."""
|
||||
result = DispatchResult(success_count=0, failure_count=0)
|
||||
assert result.all_succeeded is False
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# AlertDispatcher Tests
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestAlertDispatcher:
|
||||
"""Tests for alert dispatcher."""
|
||||
|
||||
def test_init(
|
||||
self,
|
||||
mock_discord_channel: MagicMock,
|
||||
mock_telegram_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test dispatcher initialization."""
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel, mock_telegram_channel]
|
||||
)
|
||||
assert len(dispatcher.channels) == 2
|
||||
assert "discord" in dispatcher._circuit_state
|
||||
assert "telegram" in dispatcher._circuit_state
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_all_success(
|
||||
self,
|
||||
sample_alert: FormattedAlert,
|
||||
mock_discord_channel: MagicMock,
|
||||
mock_telegram_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test successful dispatch to all channels."""
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel, mock_telegram_channel]
|
||||
)
|
||||
|
||||
result = await dispatcher.dispatch(sample_alert)
|
||||
|
||||
assert result.success_count == 2
|
||||
assert result.failure_count == 0
|
||||
assert result.all_succeeded is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_partial_failure(
|
||||
self,
|
||||
sample_alert: FormattedAlert,
|
||||
mock_discord_channel: MagicMock,
|
||||
mock_telegram_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test dispatch with one channel failing."""
|
||||
mock_telegram_channel.send.return_value = False
|
||||
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel, mock_telegram_channel]
|
||||
)
|
||||
|
||||
result = await dispatcher.dispatch(sample_alert)
|
||||
|
||||
assert result.success_count == 1
|
||||
assert result.failure_count == 1
|
||||
assert result.channel_results["discord"] is True
|
||||
assert result.channel_results["telegram"] is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_no_channels(
|
||||
self, sample_alert: FormattedAlert
|
||||
) -> None:
|
||||
"""Test dispatch with no channels configured."""
|
||||
dispatcher = AlertDispatcher(channels=[])
|
||||
|
||||
result = await dispatcher.dispatch(sample_alert)
|
||||
|
||||
assert result.success_count == 0
|
||||
assert result.failure_count == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_circuit_opens_after_failures(
|
||||
self,
|
||||
sample_alert: FormattedAlert,
|
||||
mock_discord_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test circuit breaker opens after threshold failures."""
|
||||
mock_discord_channel.send.return_value = False
|
||||
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel],
|
||||
failure_threshold=3,
|
||||
)
|
||||
|
||||
# First 3 failures
|
||||
for _ in range(3):
|
||||
await dispatcher.dispatch(sample_alert)
|
||||
|
||||
# Circuit should be open now
|
||||
assert dispatcher._circuit_state["discord"].is_open is True
|
||||
assert dispatcher._circuit_state["discord"].failure_count == 3
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_circuit_skips_when_open(
|
||||
self,
|
||||
sample_alert: FormattedAlert,
|
||||
mock_discord_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test that open circuit skips delivery."""
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel],
|
||||
failure_threshold=3,
|
||||
recovery_timeout_seconds=3600, # Long timeout
|
||||
)
|
||||
|
||||
# Manually open the circuit
|
||||
dispatcher._circuit_state["discord"].is_open = True
|
||||
dispatcher._circuit_state["discord"].last_failure_time = datetime.now(UTC)
|
||||
|
||||
result = await dispatcher.dispatch(sample_alert)
|
||||
|
||||
assert result.channel_results["discord"] is False
|
||||
# send() should not be called
|
||||
mock_discord_channel.send.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_circuit_closes_on_success(
|
||||
self,
|
||||
sample_alert: FormattedAlert,
|
||||
mock_discord_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test circuit closes on successful delivery."""
|
||||
mock_discord_channel.send.return_value = False
|
||||
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel],
|
||||
failure_threshold=2,
|
||||
)
|
||||
|
||||
# Cause failures to open circuit
|
||||
await dispatcher.dispatch(sample_alert)
|
||||
await dispatcher.dispatch(sample_alert)
|
||||
assert dispatcher._circuit_state["discord"].is_open is True
|
||||
|
||||
# Now succeed
|
||||
mock_discord_channel.send.return_value = True
|
||||
# Force half-open by resetting last_failure to past
|
||||
dispatcher._circuit_state["discord"].last_failure_time = datetime(
|
||||
2020, 1, 1, tzinfo=UTC
|
||||
)
|
||||
|
||||
result = await dispatcher.dispatch(sample_alert)
|
||||
|
||||
assert result.channel_results["discord"] is True
|
||||
assert dispatcher._circuit_state["discord"].is_open is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatch_batch(
|
||||
self,
|
||||
sample_alert: FormattedAlert,
|
||||
mock_discord_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test batch dispatch."""
|
||||
dispatcher = AlertDispatcher(channels=[mock_discord_channel])
|
||||
|
||||
alerts = [sample_alert, sample_alert, sample_alert]
|
||||
results = await dispatcher.dispatch_batch(alerts)
|
||||
|
||||
assert len(results) == 3
|
||||
assert all(r.success_count == 1 for r in results)
|
||||
|
||||
def test_get_circuit_status(
|
||||
self,
|
||||
mock_discord_channel: MagicMock,
|
||||
mock_telegram_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test getting circuit status."""
|
||||
dispatcher = AlertDispatcher(
|
||||
channels=[mock_discord_channel, mock_telegram_channel]
|
||||
)
|
||||
|
||||
status = dispatcher.get_circuit_status()
|
||||
|
||||
assert "discord" in status
|
||||
assert "telegram" in status
|
||||
assert status["discord"]["is_open"] is False
|
||||
assert status["discord"]["failure_count"] == 0
|
||||
|
||||
def test_reset_circuit(
|
||||
self,
|
||||
mock_discord_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test manual circuit reset."""
|
||||
dispatcher = AlertDispatcher(channels=[mock_discord_channel])
|
||||
|
||||
# Set up failure state
|
||||
dispatcher._circuit_state["discord"].failure_count = 5
|
||||
dispatcher._circuit_state["discord"].is_open = True
|
||||
|
||||
# Reset
|
||||
result = dispatcher.reset_circuit("discord")
|
||||
|
||||
assert result is True
|
||||
assert dispatcher._circuit_state["discord"].failure_count == 0
|
||||
assert dispatcher._circuit_state["discord"].is_open is False
|
||||
|
||||
def test_reset_circuit_unknown_channel(
|
||||
self,
|
||||
mock_discord_channel: MagicMock,
|
||||
) -> None:
|
||||
"""Test reset with unknown channel name."""
|
||||
dispatcher = AlertDispatcher(channels=[mock_discord_channel])
|
||||
|
||||
result = dispatcher.reset_circuit("unknown")
|
||||
|
||||
assert result is False
|
||||
Reference in New Issue
Block a user