mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
junv
2625ef10b4
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).
216 lines
7.7 KiB
Python
216 lines
7.7 KiB
Python
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
|
|
|
|
|
|
@pytest.fixture
|
|
def watcher():
|
|
with patch('pricemon.tasks.schedule_watcher'):
|
|
return PriceWatcher.objects.create(
|
|
name='Test product',
|
|
url='https://example.com/product',
|
|
css_selector='.price',
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def alert(watcher):
|
|
return PriceAlert.objects.create(
|
|
watcher=watcher,
|
|
old_price='100.00',
|
|
new_price='80.00',
|
|
drop_pct='20.00',
|
|
)
|
|
|
|
|
|
@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):
|
|
response = client.get(reverse('pricemon-dashboard'))
|
|
|
|
assert response.status_code == 200
|
|
assert reverse('pricemon-toggle-enabled', args=[watcher.pk]) in response.content.decode()
|
|
assert 'Disable monitoring' in response.content.decode()
|
|
assert response.context['enabled_watchers_count'] == 1
|
|
|
|
def test_toggle_disables_watcher_and_unschedules_job(self, client, watcher):
|
|
with patch('pricemon.tasks.unschedule_watcher') as unschedule_watcher:
|
|
response = client.post(reverse('pricemon-toggle-enabled', args=[watcher.pk]))
|
|
|
|
watcher.refresh_from_db()
|
|
assert response.status_code == 302
|
|
assert response.url == reverse('pricemon-dashboard')
|
|
assert watcher.enabled is False
|
|
unschedule_watcher.assert_called_once()
|
|
|
|
def test_toggle_enables_paused_watcher_and_schedules_job(self, client, watcher):
|
|
with patch('pricemon.tasks.unschedule_watcher'):
|
|
watcher.enabled = False
|
|
watcher.save(update_fields=['enabled'])
|
|
|
|
with patch('pricemon.tasks.schedule_watcher') as schedule_watcher:
|
|
response = client.post(reverse('pricemon-toggle-enabled', args=[watcher.pk]))
|
|
|
|
watcher.refresh_from_db()
|
|
assert response.status_code == 302
|
|
assert watcher.enabled is True
|
|
schedule_watcher.assert_called_once()
|
|
|
|
def test_toggle_rejects_get_requests(self, client, watcher):
|
|
response = client.get(reverse('pricemon-toggle-enabled', args=[watcher.pk]))
|
|
|
|
assert response.status_code == 405
|
|
|
|
|
|
@pytest.mark.django_db
|
|
class TestPriceMonitorAlertsPartial:
|
|
def test_dashboard_renders_with_active_alert(self, client, watcher, alert):
|
|
response = client.get(reverse('pricemon-dashboard'))
|
|
|
|
assert response.status_code == 200
|
|
body = response.content.decode()
|
|
assert 'alert-{}'.format(alert.pk) in body
|
|
assert 'Price Drop Alerts' in body
|
|
|
|
def test_alerts_partial_renders_i18n_tags(self, client, watcher, alert):
|
|
response = client.get(reverse('pricemon-alerts-partial'))
|
|
|
|
assert response.status_code == 200
|
|
body = response.content.decode()
|
|
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')
|