source over interval method, tests, and an additional 'endpoint' knob…

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,

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):
 

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):

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",
+                )