test(pricemon): pin telegram notification contract

Add integration tests covering the price-drop → Telegram alert flow:

- Sends Telegram MarkdownV2 message to global SiteSettings chat_id when
  detected price <= watcher.alert_price_threshold.
- No alert when global creds missing.
- No alert when price above threshold.
- No alert when threshold unset on watcher.
- Non-recurring watcher with an undismissed alert is not re-notified.
- Telegram API failure does not break the check (snapshot + alert row
  still persist, last_price updates).
This commit is contained in:
2026-07-16 09:50:37 +10:00
parent b24d59a9ce
commit 2625ef10b4
+130
View File
@@ -1,8 +1,10 @@
from decimal import Decimal
from unittest.mock import patch
import pytest
from django.urls import reverse
from links.models import SiteSettings
from pricemon.models import PriceAlert, PriceWatcher
@@ -26,6 +28,23 @@ def alert(watcher):
)
@pytest.fixture
def telegram_creds():
ss = SiteSettings.get()
ss.telegram_bot_token = '123:abc'
ss.telegram_chat_id = '999'
ss.save()
return ss
@pytest.fixture
def threshold_watcher(watcher):
watcher.alert_price_threshold = Decimal('100.00')
watcher.last_price = Decimal('120.00')
watcher.save()
return watcher
@pytest.mark.django_db
class TestPriceMonitorToggle:
def test_dashboard_shows_disable_action_for_enabled_watcher(self, client, watcher):
@@ -83,3 +102,114 @@ class TestPriceMonitorAlertsPartial:
assert 'alert-{}'.format(alert.pk) in body
assert 'View' in body and 'Dismiss' in body
assert reverse('pricemon-dismiss-alert', args=[alert.pk]) in body
@pytest.mark.django_db
class TestPriceMonitorTelegram:
def _check(self, watcher_pk):
with patch('pricemon.scraper.fetch_price') as fetch, patch(
'pricemon.notifications.requests.post'
) as post:
fetch.return_value = (Decimal('90.00'), '$90', '')
post.return_value.status_code = 200
post.return_value.raise_for_status.return_value = None
from pricemon.tasks import check_watcher
check_watcher(watcher_pk)
return post
def test_sends_telegram_when_price_meets_threshold(self, threshold_watcher, telegram_creds):
post = self._check(threshold_watcher.pk)
post.assert_called_once()
url = post.call_args[0][0]
assert url.endswith(f'/bot{telegram_creds.telegram_bot_token}/sendMessage')
body = post.call_args.kwargs['json']
assert body['chat_id'] == '999'
assert body['parse_mode'] == 'MarkdownV2'
assert 'Price Drop Alert' in body['text']
assert threshold_watcher.name in body['text']
assert PriceAlert.objects.count() == 1
def test_no_telegram_without_global_creds(self, threshold_watcher):
SiteSettings.get().delete()
with patch('pricemon.scraper.fetch_price') as fetch, patch(
'pricemon.notifications.requests.post'
) as post:
fetch.return_value = (Decimal('90.00'), '$90', '')
post.return_value.status_code = 200
post.return_value.raise_for_status.return_value = None
from pricemon.tasks import check_watcher
check_watcher(threshold_watcher.pk)
post.assert_not_called()
assert PriceAlert.objects.count() == 1
def test_no_alert_when_price_above_threshold(self, threshold_watcher, telegram_creds):
with patch('pricemon.scraper.fetch_price') as fetch, patch(
'pricemon.notifications.requests.post'
) as post:
fetch.return_value = (Decimal('110.00'), '$110', '')
post.return_value.status_code = 200
post.return_value.raise_for_status.return_value = None
from pricemon.tasks import check_watcher
check_watcher(threshold_watcher.pk)
post.assert_not_called()
assert PriceAlert.objects.count() == 0
def test_no_alert_when_threshold_unset(self, watcher, telegram_creds):
with patch('pricemon.scraper.fetch_price') as fetch, patch(
'pricemon.notifications.requests.post'
) as post:
fetch.return_value = (Decimal('1.00'), '$1', '')
post.return_value.status_code = 200
post.return_value.raise_for_status.return_value = None
from pricemon.tasks import check_watcher
check_watcher(watcher.pk)
post.assert_not_called()
assert PriceAlert.objects.count() == 0
def test_non_recurring_alert_skipped_if_undismissed(
self, threshold_watcher, telegram_creds
):
threshold_watcher.recurring_notification = False
threshold_watcher.save()
PriceAlert.objects.create(
watcher=threshold_watcher,
old_price='120.00',
new_price='95.00',
drop_pct='20.83',
dismissed=False,
)
with patch('pricemon.scraper.fetch_price') as fetch, patch(
'pricemon.notifications.requests.post'
) as post:
fetch.return_value = (Decimal('90.00'), '$90', '')
post.return_value.status_code = 200
post.return_value.raise_for_status.return_value = None
from pricemon.tasks import check_watcher
check_watcher(threshold_watcher.pk)
post.assert_not_called()
assert PriceAlert.objects.count() == 1
def test_telegram_failure_does_not_break_check(self, threshold_watcher, telegram_creds):
with patch('pricemon.scraper.fetch_price') as fetch, patch(
'pricemon.notifications.requests.post'
) as post:
fetch.return_value = (Decimal('90.00'), '$90', '')
post.side_effect = Exception('Boom from Telegram')
from pricemon.tasks import check_watcher
check_watcher(threshold_watcher.pk)
post.assert_called_once()
assert PriceAlert.objects.count() == 1
threshold_watcher.refresh_from_db()
assert threshold_watcher.last_price == Decimal('90.00')