diff --git a/tests/test_pricemon.py b/tests/test_pricemon.py index 74c8e10..6db5924 100644 --- a/tests/test_pricemon.py +++ b/tests/test_pricemon.py @@ -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')