ilia
c8a82e264c
Tests cover providers, dedup, Telegram, scoring, main runner, and Airbnb stubs. Ticketmaster and SeatGeek use configurable lat/lon/radius (Thornhill default). Pipeline filters noise listings, merges same-day sports duplicates, optional MIN_ALERT_SCORE, and Telegram severity summary. Made-with: Cursor
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""Tests for the Telegram notification module."""
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from src.models import NormalizedEvent
|
|
from src.notifications.telegram import (
|
|
send_alert,
|
|
_format_message,
|
|
_escape_md,
|
|
_score_indicator,
|
|
_severity_summary,
|
|
)
|
|
|
|
|
|
def _make_event(name: str, event_date: date, venue: str, score: float = 0.5) -> NormalizedEvent:
|
|
return NormalizedEvent(
|
|
name=name, event_date=event_date, venue=venue,
|
|
source="test", url="", score=score,
|
|
)
|
|
|
|
|
|
class TestEscapeMd:
|
|
def test_escapes_special_characters(self):
|
|
assert _escape_md("hello_world") == "hello\\_world"
|
|
assert _escape_md("*bold*") == "\\*bold\\*"
|
|
assert _escape_md("a.b") == "a\\.b"
|
|
|
|
def test_plain_text_unchanged(self):
|
|
assert _escape_md("hello") == "hello"
|
|
|
|
def test_multiple_special_chars(self):
|
|
result = _escape_md("[link](url)")
|
|
assert "\\[" in result
|
|
assert "\\]" in result
|
|
assert "\\(" in result
|
|
assert "\\)" in result
|
|
|
|
|
|
class TestScoreIndicator:
|
|
def test_high_score(self):
|
|
assert _score_indicator(0.8) == "\U0001f534"
|
|
assert _score_indicator(1.0) == "\U0001f534"
|
|
|
|
def test_medium_score(self):
|
|
assert _score_indicator(0.5) == "\U0001f7e1"
|
|
assert _score_indicator(0.7) == "\U0001f7e1"
|
|
|
|
def test_low_score(self):
|
|
assert _score_indicator(0.3) == "\U0001f7e2"
|
|
assert _score_indicator(0.0) == "\U0001f7e2"
|
|
|
|
|
|
class TestSeveritySummary:
|
|
def test_counts_buckets(self):
|
|
events = [
|
|
_make_event("A", date(2026, 5, 10), "V", 0.9),
|
|
_make_event("B", date(2026, 5, 10), "V", 0.6),
|
|
_make_event("C", date(2026, 5, 11), "V", 0.2),
|
|
]
|
|
s = _severity_summary(events)
|
|
assert "3 events" in s
|
|
assert "1 high" in s
|
|
assert "1 medium" in s
|
|
assert "1 lower" in s
|
|
|
|
|
|
class TestFormatMessage:
|
|
def test_groups_events_by_date(self):
|
|
events = [
|
|
_make_event("Show A", date(2026, 5, 10), "Venue A"),
|
|
_make_event("Show B", date(2026, 5, 10), "Venue B"),
|
|
_make_event("Show C", date(2026, 5, 12), "Venue C"),
|
|
]
|
|
message = _format_message(events)
|
|
assert "Show A" in message
|
|
assert "Show B" in message
|
|
assert "Show C" in message
|
|
|
|
def test_sorts_by_date(self):
|
|
events = [
|
|
_make_event("Late", date(2026, 5, 15), "V1"),
|
|
_make_event("Early", date(2026, 5, 10), "V2"),
|
|
]
|
|
message = _format_message(events)
|
|
early_pos = message.index("Early")
|
|
late_pos = message.index("Late")
|
|
assert early_pos < late_pos
|
|
|
|
def test_contains_header(self):
|
|
events = [_make_event("Show", date(2026, 5, 10), "Venue")]
|
|
message = _format_message(events)
|
|
assert "EventRate Alert" in message
|
|
assert "1 event ·" in message
|
|
|
|
def test_includes_venue(self):
|
|
events = [_make_event("Show", date(2026, 5, 10), "Scotiabank Arena")]
|
|
message = _format_message(events)
|
|
assert "Scotiabank Arena" in message
|
|
|
|
|
|
class TestSendAlert:
|
|
def test_skips_when_no_credentials(self):
|
|
assert send_alert([], "", "chat123") is False
|
|
assert send_alert([], "token", "") is False
|
|
|
|
def test_returns_true_when_no_events(self):
|
|
assert send_alert([], "token", "chat123") is True
|
|
|
|
def test_sends_message_on_success(self, httpx_mock):
|
|
httpx_mock.add_response(json={"ok": True})
|
|
events = [_make_event("Raptors", date(2026, 5, 10), "Scotiabank Arena")]
|
|
result = send_alert(events, "fake-token", "123456")
|
|
assert result is True
|
|
|
|
request = httpx_mock.get_request()
|
|
assert request is not None
|
|
assert "fake-token" in str(request.url)
|
|
|
|
def test_returns_false_on_api_error(self, httpx_mock):
|
|
httpx_mock.add_response(status_code=400)
|
|
events = [_make_event("Raptors", date(2026, 5, 10), "Scotiabank Arena")]
|
|
result = send_alert(events, "fake-token", "123456")
|
|
assert result is False
|
|
|
|
def test_returns_false_on_network_error(self, httpx_mock):
|
|
httpx_mock.add_exception(ConnectionError("network down"))
|
|
events = [_make_event("Raptors", date(2026, 5, 10), "Scotiabank Arena")]
|
|
result = send_alert(events, "fake-token", "123456")
|
|
assert result is False
|