From cb36837f31a7ac05bd47c11ddf440320981af6eb Mon Sep 17 00:00:00 2001 From: Paige Gulley Date: Wed, 15 Apr 2026 13:37:30 -0400 Subject: [PATCH] source over interval method, tests, and an additional 'endpoint' knob on tests for instance targeting --- mediacloud/api.py | 9 ++++ mediacloud/test/api_base_test.py | 1 + mediacloud/test/api_directory_test.py | 1 + mediacloud/test/api_search_test.py | 71 +++++++++++++++++++++++++++ 4 files changed, 82 insertions(+) diff --git a/mediacloud/api.py b/mediacloud/api.py index abb42de..cebd650 100644 --- a/mediacloud/api.py +++ b/mediacloud/api.py @@ -177,6 +177,15 @@ def stories_by_source_week(self, query: str, start_date: dt.date, end_date: dt.d results = self._query('search/count-by-source-week', params) return results['source-week-attention'] + def stories_by_source_over_interval(self, query: str, start_date: dt.date, end_date: dt.date, + collection_ids: Optional[List[int]] = [], source_ids: Optional[List[int]] = [], + platform: Optional[str] = None, interval: Optional[str] = None) -> List[Dict]: + params = self._prep_default_params(query, start_date, end_date, collection_ids, source_ids, platform) + if interval: + params['interval'] = interval + results = self._query('search/count-by-source-over-interval', params) + return results['source-interval-attention'] + def story_list(self, query: str, start_date: dt.date, end_date: dt.date, collection_ids: Optional[List[int]] = [], source_ids: Optional[List[int]] = [], platform: Optional[str] = None, expanded: Optional[bool] = None, pagination_token: Optional[str] = None, diff --git a/mediacloud/test/api_base_test.py b/mediacloud/test/api_base_test.py index e4fcde1..a0faf8f 100644 --- a/mediacloud/test/api_base_test.py +++ b/mediacloud/test/api_base_test.py @@ -4,6 +4,7 @@ import mediacloud.api from mediacloud.error import MCException +mediacloud.api.BaseApi.BASE_API_URL = os.getenv("MC_API_BASE_URL", "https://search.mediacloud.org/api/") class BaseApiTest(TestCase): diff --git a/mediacloud/test/api_directory_test.py b/mediacloud/test/api_directory_test.py index e3f3c42..bc8a5e3 100644 --- a/mediacloud/test/api_directory_test.py +++ b/mediacloud/test/api_directory_test.py @@ -9,6 +9,7 @@ TEST_COLLECTION_ID = 34412234 # US -National sources TEST_SOURCE_ID = 1095 # cnn.com TEST_FEED_ID = 1 +mediacloud.api.BaseApi.BASE_API_URL = os.getenv("MC_API_BASE_URL", "https://search.mediacloud.org/api/") class DirectoryTest(TestCase): diff --git a/mediacloud/test/api_search_test.py b/mediacloud/test/api_search_test.py index 1f4f60b..b86c695 100644 --- a/mediacloud/test/api_search_test.py +++ b/mediacloud/test/api_search_test.py @@ -2,6 +2,7 @@ import os import time from unittest import TestCase +from unittest.mock import patch import pytest @@ -13,6 +14,9 @@ START_DATE = dt.date(2023, 11, 1) END_DATE = dt.date(2023, 12, 1) +#Optionally override the target instance when testing, for staging/dev cases +mediacloud.api.BaseApi.BASE_API_URL = os.getenv("MC_API_BASE_URL", "https://search.mediacloud.org/api/") + class BaseSearchTest(TestCase): @@ -362,3 +366,70 @@ def test_datetime(self): collection_ids=[COLLECTION_US_NATIONAL])['relevant'] assert result_via_date == result_via_datetime + + def test_stories_by_source_over_interval_day(self): + expected = [{ + "media_name": "example.com", + "interval": "day", + "bucket": "2025-01-01", + "matching_stories": 10, + "total_stories": 100, + "ratio": 0.1, + }] + with patch.object(self._search, "_query", return_value={"source-interval-attention": expected}) as mock_query: + result = self._search.stories_by_source_over_interval( + query="tariff AND Trump", + start_date=self.START_DATE, + end_date=self.END_DATE, + collection_ids=[COLLECTION_US_NATIONAL], + interval="day", + ) + assert result == expected + endpoint, params = mock_query.call_args.args + assert endpoint == "search/count-by-source-over-interval" + assert params["interval"] == "day" + + def test_stories_by_source_over_interval_week(self): + expected = [{ + "media_name": "example.com", + "interval": "week", + "bucket": "2025-W01", + "matching_stories": 5, + "total_stories": 50, + "ratio": 0.1, + }] + with patch.object(self._search, "_query", return_value={"source-interval-attention": expected}) as mock_query: + result = self._search.stories_by_source_over_interval( + query="tariff AND Trump", + start_date=self.START_DATE, + end_date=self.END_DATE, + interval="week", + ) + assert result == expected + endpoint, params = mock_query.call_args.args + assert endpoint == "search/count-by-source-over-interval" + assert params["interval"] == "week" + + def test_stories_by_source_over_interval_default_omits_interval(self): + with patch.object(self._search, "_query", return_value={"source-interval-attention": []}) as mock_query: + self._search.stories_by_source_over_interval( + query="tariff AND Trump", + start_date=self.START_DATE, + end_date=self.END_DATE, + ) + _, params = mock_query.call_args.args + assert "interval" not in params + + def test_stories_by_source_over_interval_propagates_api_error(self): + with patch.object(self._search, "_query", side_effect=mediacloud.error.APIResponseError( + response=type("Resp", (), {"status_code": 400})(), + params={"interval": "bad"}, + data={"note": "invalid interval"}, + )): + with pytest.raises(mediacloud.error.APIResponseError): + self._search.stories_by_source_over_interval( + query="tariff AND Trump", + start_date=self.START_DATE, + end_date=self.END_DATE, + interval="bad", + )