"""Tests for FeedMonitor module and class methods.""" from collections import OrderedDict import datetime import json from unittest.mock import patch from freezegun import freeze_time import pytest from feed_status.models import feed_monitor from feed_status.models import ingestion_feed from feed_status.models import outgoing_feed from feed_status.models.feed_monitor import FeedMonitor, \ FeedMonitorWithDayOfWeekPattern from feed_status.models.feed_monitor import METRIC_FRESH from feed_status.models.feed_monitor import METRIC_STABLE from feed_status.models.feed_monitor import METRIC_STATUS_CRITICAL from feed_status.models.feed_monitor import METRIC_STATUS_OK from feed_status.models.feed_monitor import METRIC_STATUS_PENDING from feed_status.models.ingestion_feed import INGESTION_TYPE_DAILY from feed_status.models.ingestion_feed import INGESTION_TYPE_WEEKLY from feed_status.models.ingestion_feed import IngestionFeed @pytest.fixture def mock_monitor_fresh(): """Mock FeedMonitor instance.""" return FeedMonitor( feed_id='1', metric_name=METRIC_FRESH, threshold=1, threshold_time='14:00', only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), @pytest.fixture def mock_monitor_stable(): """Mock FeedMonitor instance.""" return FeedMonitor( feed_id='1', metric_name=METRIC_STABLE, threshold=2, threshold_time='00:00', only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), @pytest.fixture def mock_monitor_stable_weekly(): """Mock FeedMonitor instance for weekly tests.""" return FeedMonitor( feed_id='test_weekly', metric_name=METRIC_STABLE, threshold=6, threshold_time='00:00', only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), @pytest.fixture def mock_monitor_dow_pattern(): """Mock FeedMonitor instance for dow pattern tests.""" return FeedMonitorWithDayOfWeekPattern( feed_id='dow_pattern', metric_name=METRIC_STABLE, threshold=7, threshold_time=None, active_days_of_week=[0, 1, 3, 4], success_status=ingestion_feed.INGESTION_STATUS_INGESTED) @pytest.fixture def mock_monitor_stable_outgoing(): """Mock outgoing FeedMonitor instance.""" return FeedMonitor( feed_id='proper_new_releases_tracks', metric_name=METRIC_STABLE, threshold=2, threshold_time='00:00', only_business_days=False, success_status=outgoing_feed.OUTGOING_STATUS_SENT), @pytest.fixture(autouse=True) def mock_feed_statuses(): """Mock feed_statuses for monitor tests.""" today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED yield mock_statuses mock_feed_monitor_config = [ FeedMonitor(feed_id='1', metric_name=METRIC_FRESH, threshold=5, threshold_time=None, only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), FeedMonitor(feed_id='1', metric_name=METRIC_STABLE, threshold=10, threshold_time=None, only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), FeedMonitor(feed_id='2', metric_name=METRIC_FRESH, threshold=5, threshold_time='00:00', only_business_days=True, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), FeedMonitor(feed_id='3', metric_name=METRIC_FRESH, threshold=5, threshold_time=None, only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), FeedMonitor(feed_id='4', metric_name=METRIC_STABLE, threshold=10, threshold_time=None, only_business_days=True, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), FeedMonitor(feed_id='Frosty', metric_name=METRIC_FRESH, threshold=5, threshold_time=None, only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), FeedMonitor(feed_id='Frosty', metric_name=METRIC_STABLE, threshold=10, threshold_time=None, only_business_days=False, success_status=ingestion_feed.INGESTION_STATUS_INGESTED), ] mock_ingestion_feed_config = [ IngestionFeed(feed_id='test_weekly', feed_name='Test Weekly', status_source_db='feed_status', update_freq=1, feed_type=INGESTION_TYPE_WEEKLY) ] mock_ingestion_with_day_of_week_pattern_config = [ IngestionFeed(feed_id='dow_pattern', feed_name='dow_pattern', status_source_db='feed_status', update_freq=1, feed_type=INGESTION_TYPE_DAILY) ] class TestFindMonitors: """Unit tests for find_monitors().""" @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_find_by_feed_id(self): """Test we can find FeedMonitors by feed_id.""" monitors = feed_monitor.find_monitors(feed_id='1') assert len(monitors) == 2 @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_find_by_feed_id_and_metric(self): """Test we can find FeedMonitors by feed_id and metric.""" monitors = feed_monitor.find_monitors(feed_id='Frosty', metric_name=METRIC_FRESH) assert len(monitors) == 1 @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_find_by_threshold(self): """Test we can find FeedMonitors by threshold.""" monitors = feed_monitor.find_monitors(threshold=10) assert len(monitors) == 3 @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_find_by_business_days(self): """Test we can find FeedMonitors by only_business_days.""" monitors = feed_monitor.find_monitors(only_business_days=True) assert len(monitors) == 2 @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_find_nothing_with_unknown_attrib(self): """Test we find no FeedMonitors when looking by unknown attribute.""" monitors = feed_monitor.find_monitors(holidaze=True) assert len(monitors) == 0 class TestGetMetricStatus: """Unit tests for get_metric_status().""" @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_get_metric_status_with_string_feed_id(self, mock_feed_statuses): """Test return metric status when feed has string id.""" feed = IngestionFeed(feed_id='2', feed_name='name', status_source_db='', update_freq=1, feed_type=INGESTION_TYPE_DAILY) assert feed_monitor.get_metric_status(feed, mock_feed_statuses) == [{ 'name': METRIC_FRESH, 'status': 'ok', 'threshold': 5, 'only_business_days': True }] @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_get_metric_status_with_integer_feed_id(self, mock_feed_statuses): """Test return metric status when feed has integer id.""" feed = IngestionFeed(feed_id=2, feed_name='name', status_source_db='', update_freq=1, feed_type=INGESTION_TYPE_DAILY) assert feed_monitor.get_metric_status(feed, mock_feed_statuses) == [{ 'name': METRIC_FRESH, 'status': 'ok', 'threshold': 5, 'only_business_days': True }] @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_get_metric_status_all_attributes(self): """Test return all required metric status attributes.""" feed = IngestionFeed(feed_id='4', feed_name='name', status_source_db='', update_freq=1, feed_type=INGESTION_TYPE_DAILY) assert feed_monitor.get_metric_status(feed, OrderedDict()) == [{ 'name': METRIC_STABLE, 'status': 'critical', 'threshold': 10, 'only_business_days': True }] @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_get_metric_status_when_multiple_monitors( self, mock_feed_statuses): """Return required metric status attribs with multiple moniters.""" feed = IngestionFeed(feed_id='Frosty', feed_name='name', status_source_db='', update_freq=1, feed_type=INGESTION_TYPE_DAILY) assert feed_monitor.get_metric_status(feed, mock_feed_statuses) == [{ 'name': METRIC_FRESH, 'status': 'ok', 'threshold': 5, 'only_business_days': False }, { 'name': METRIC_STABLE, 'status': 'critical', 'threshold': 10, 'only_business_days': False }] @patch('feed_status.models.feed_monitor.FEED_MONITORS', mock_feed_monitor_config) def test_get_metric_status_emits_json_per_monitor( self, mock_feed_statuses, capsys): """Emit one JSON line per monitor to stdout.""" feed = IngestionFeed(feed_id='Frosty', feed_name='Frosty Feed', status_source_db='', update_freq=1, feed_type=INGESTION_TYPE_DAILY) feed_monitor.get_metric_status(feed, mock_feed_statuses) lines = [line for line in capsys.readouterr().out.splitlines() if line.strip()] payloads = [json.loads(line) for line in lines] assert payloads == [{ 'feed_name': 'Frosty Feed', 'feed_id': 'Frosty', 'status': 'ok', 'metrics_category': 'freshness', 'threshold': 5, }, { 'feed_name': 'Frosty Feed', 'feed_id': 'Frosty', 'status': 'critical', 'metrics_category': 'stability', 'threshold': 10, }] class TestIngestionFreshMetricEval: """Unit tests for ingestion metric_eval functions.""" def test_status_fresh_ingestion_eval_ok(self, mock_monitor_fresh): """Return 'ok' for healthy ingestion feed statuses.""" today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED with freeze_time('{today_str} 16:00:00'.format( today_str=today.isoformat())): assert mock_monitor_fresh[0].metric_eval( mock_statuses) == METRIC_STATUS_OK def test_status_fresh_ingestion_eval_critical(self, mock_monitor_fresh): """Return 'critical' for unhealthy ingestion feed statuses.""" monitor = mock_monitor_fresh[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_INGESTED with freeze_time('{today_str} 16:00:00'.format( today_str=today.isoformat())): assert monitor.metric_eval(mock_statuses) == METRIC_STATUS_CRITICAL class TestIngestionFreshMetricEvalCurrentDayHandling: """Unit tests for metric_eval functions when considering current date.""" def test_status_fresh_ingestion_eval_pending_before_threshold_time( self, mock_monitor_fresh): """Return 'pending' when unhealthy before threshold time.""" monitor = mock_monitor_fresh[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_INGESTED with freeze_time('{today_str} 12:00:00'.format( today_str=today.isoformat())): assert monitor.metric_eval(mock_statuses) == METRIC_STATUS_PENDING def test_status_fresh_ingestion_eval_critical_after_threshold_time( self, mock_monitor_fresh): """Return 'critical' when unhealthy after threshold time.""" monitor = mock_monitor_fresh[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_INGESTED with freeze_time('{today_str} 14:01:00'.format( today_str=today.isoformat())): assert monitor.metric_eval(mock_statuses) == METRIC_STATUS_CRITICAL def test_status_fresh_ingest_ok_before_threshold_time_ingested( self, mock_monitor_fresh): """Return 'ok' when unhealthy before threshold time.""" monitor = mock_monitor_fresh[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED with freeze_time('{today_str} 13:59:00'.format( today_str=today.isoformat())): assert monitor.metric_eval(mock_statuses) == METRIC_STATUS_OK class TestOutgoingStableMetricEval: """Unit tests for outgoing metric_eval functions.""" def test_status_stable_outgoing_eval_ok( self, mock_monitor_stable_outgoing): """Return 'ok' for healthy outgoing feed statuses.""" monitor = mock_monitor_stable_outgoing[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = outgoing_feed.OUTGOING_STATUS_SENT assert monitor.metric_eval(mock_statuses) == METRIC_STATUS_OK def test_status_stable_outgoing_eval_critical( self, mock_monitor_stable_outgoing): """Return 'critical' for unhealthy outgoing feed statuses.""" monitor = mock_monitor_stable_outgoing[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) mock_statuses[ date.isoformat() ] = outgoing_feed.OUTGOING_STATUS_PROCESSING assert monitor.metric_eval(mock_statuses) == 'critical' @patch('feed_status.models.ingestion_feed.active_feeds', mock_ingestion_feed_config) class TestWeeklyStabileMetricEval: """Unit tests for get_feed_by_id().""" def test_weekly_status_stable_eval_ok( self, mock_monitor_stable_weekly): """Return 'ok' for healthy weekly feed statuses.""" monitor = mock_monitor_stable_weekly[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) if day in [7, 14, 21, 28]: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED else: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_AVAILABLE assert monitor.metric_eval(mock_statuses) == METRIC_STATUS_OK def test_weekly_status_stable_eval_critical( self, mock_monitor_stable_weekly): """Return 'critical' for unhealthy weekly feed statuses.""" monitor = mock_monitor_stable_weekly[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) if day in [8]: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED else: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_AVAILABLE assert monitor.metric_eval(mock_statuses) == 'critical' def test_weekly_status_stable_eval_critical_on_two_gaps( self, mock_monitor_stable_weekly): """Return 'critical' for unhealthy weekly feed statuses.""" monitor = mock_monitor_stable_weekly[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) if day in [8, 16, 17, 26, 27, 28]: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED else: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_AVAILABLE assert monitor.metric_eval(mock_statuses) == 'critical' def test_weekly_status_stable_eval_critical_on_one_gap( self, mock_monitor_stable_weekly): """Return 'critical' for unhealthy weekly feed statuses.""" monitor = mock_monitor_stable_weekly[0] today = datetime.date.today() mock_statuses = OrderedDict() for day in range(30): date = today - datetime.timedelta(day) if day in [8, 14, 17, 28]: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_INGESTED else: mock_statuses[ date.isoformat() ] = ingestion_feed.INGESTION_STATUS_NOT_AVAILABLE assert monitor.metric_eval(mock_statuses) == 'critical' @patch('feed_status.models.ingestion_feed.active_feeds', mock_ingestion_with_day_of_week_pattern_config) class TestFeedMonitorWithDayOfWeekPatternStabilityEval: """Unit tests for FeedMonitorWithDayOfWeekPattern.""" def test_status_stable_eval_ok( self, mock_monitor_dow_pattern): """Return 'ok' for healthy weekly feed statuses.""" # active_days_of_week = [0, 1, 3, 4], first_day = datetime.datetime.strptime('2025-01-01', '%Y-%m-%d').date() mock_statuses = OrderedDict() ingested = ingestion_feed.INGESTION_STATUS_INGESTED not_ingested = ingestion_feed.INGESTION_STATUS_NOT_INGESTED for week in range(0, 4): # Wednesday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 0)).strftime('%Y-%m-%d')] = not_ingested # Thursday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 1)).strftime('%Y-%m-%d')] = ingested # Friday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 2)).strftime('%Y-%m-%d')] = ingested # Saturday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 3)).strftime('%Y-%m-%d')] = not_ingested # Sunday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 4)).strftime('%Y-%m-%d')] = not_ingested # Monday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 5)).strftime('%Y-%m-%d')] = ingested # Tuesday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 6)).strftime('%Y-%m-%d')] = ingested assert mock_monitor_dow_pattern.metric_eval( mock_statuses) == METRIC_STATUS_OK def test_status_stable_eval_one_day_not_ingested( self, mock_monitor_dow_pattern): """Return 'ok' for healthy weekly feed statuses.""" # active_days_of_week = [0, 1, 3, 4], first_day = datetime.datetime.strptime('2025-01-01', '%Y-%m-%d').date() mock_statuses = OrderedDict() ingested = ingestion_feed.INGESTION_STATUS_INGESTED not_ingested = ingestion_feed.INGESTION_STATUS_NOT_INGESTED for week in range(0, 4): # Wednesday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 0)).strftime('%Y-%m-%d')] = ingested # Thursday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 1)).strftime('%Y-%m-%d')] = ingested # Friday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 2)).strftime('%Y-%m-%d')] = ingested # Saturday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 3)).strftime('%Y-%m-%d')] = ingested # Sunday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 4)).strftime('%Y-%m-%d')] = ingested # Monday if week == 2: monday_status = not_ingested else: monday_status = ingested mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 5)).strftime('%Y-%m-%d')] = monday_status # Tuesday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 6)).strftime('%Y-%m-%d')] = ingested assert mock_monitor_dow_pattern.metric_eval( OrderedDict(mock_statuses)) == METRIC_STATUS_CRITICAL def test_status_stable_eval_critical( self, mock_monitor_dow_pattern): """Return 'critical' for healthy weekly feed statuses.""" # active_days_of_week = [0, 1, 3, 4], first_day = datetime.datetime.strptime('2025-01-01', '%Y-%m-%d').date() mock_statuses = OrderedDict() ingested = ingestion_feed.INGESTION_STATUS_INGESTED not_ingested = ingestion_feed.INGESTION_STATUS_NOT_INGESTED for week in range(0, 4): # Wednesday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 0)).strftime('%Y-%m-%d')] = not_ingested # Thursday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 1)).strftime('%Y-%m-%d')] = not_ingested # Friday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 2)).strftime('%Y-%m-%d')] = ingested # Saturday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 3)).strftime('%Y-%m-%d')] = not_ingested # Sunday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 4)).strftime('%Y-%m-%d')] = not_ingested # Monday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 5)).strftime('%Y-%m-%d')] = ingested # Tuesday mock_statuses[(first_day + datetime.timedelta( days=week * 7 + 6)).strftime('%Y-%m-%d')] = ingested assert mock_monitor_dow_pattern.metric_eval( mock_statuses) == METRIC_STATUS_CRITICAL