Add support for paging through randomly sorted results

mediacloud/api.py

@@ -205,12 +205,14 @@ def stories_by_source_over_interval(self, query: str, start_date: dt.date, end_d
 
     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,
-                   sort_order: Optional[str] = None,
-                   page_size: Optional[int] = None) -> tuple[List[Story], PaginationToken]:
+                   expanded: bool = False, pagination_token: Optional[str] = None,
+                   sort_order: Optional[str] = None, page_size: Optional[int] = None,
+                   randomized: bool = False) -> tuple[List[Story], PaginationToken]:
         params = self._prep_default_params(query, start_date, end_date, collection_ids, source_ids, platform)
         if expanded:
             params['expanded'] = 1
+        if randomized:
+            params['randomize'] = 1
         if pagination_token:
             params['pagination_token'] = pagination_token
         if sort_order:

mediacloud/test/api_base_test.py

@@ -6,6 +6,7 @@
 
 mediacloud.api.BaseApi.BASE_API_URL = os.getenv("MC_API_BASE_URL", "https://search.mediacloud.org/api/")
 
+
 class BaseApiTest(TestCase):
 
     @staticmethod
@@ -32,5 +33,4 @@ def test_user_profile():
         mc_api_key = os.getenv("MC_API_TOKEN")
         client = mediacloud.api.DirectoryApi(mc_api_key)
         _ = client.user_profile()
-        print(_)
         assert True

mediacloud/test/api_search_test.py

@@ -117,6 +117,44 @@ def test_story_list_paging(self):
         assert len(results2) == 1000
         assert next_page_token2 is not None
         assert next_page_token1 != next_page_token2
+        page1_ids = [s['id'] for s in results1]
+        page2_ids = [s['id'] for s in results2]
+        common_ids = set(page1_ids) & set(page2_ids)
+        self.assertEqual(len(common_ids), 0)
+
+    def test_story_list_randomized(self):
+        # make sure ids in results differ from standard call when randomized sent in
+        results_sorted, _ = self._admin_search.story_list(query="weather", start_date=START_DATE,
+                                                          end_date=END_DATE,
+                                                          collection_ids=[COLLECTION_US_NATIONAL])
+        results_random, _ = self._admin_search.story_list(query="weather", start_date=START_DATE,
+                                                          end_date=END_DATE, randomized=True,
+                                                          collection_ids=[COLLECTION_US_NATIONAL])
+        self.assertEqual(len(results_sorted), len(results_random))
+        # make sure the id values in the two lsts are different
+        sorted_ids = [s['id'] for s in results_sorted]
+        random_ids = [s['id'] for s in results_random]
+        self.assertEqual(len(sorted_ids), len(random_ids))
+        self.assertNotEqual(sorted_ids, random_ids)
+
+    def test_story_list_paging_randomizedd(self):
+        results1, next_page_token1 = self._admin_search.story_list(query="weather", start_date=START_DATE,
+                                                                   end_date=END_DATE, randomized=True,
+                                                                   collection_ids=[COLLECTION_US_NATIONAL])
+        time.sleep(31)
+        assert len(results1) == 1000
+        assert next_page_token1 is not None
+        results2, next_page_token2 = self._admin_search.story_list(query="weather", start_date=START_DATE,
+                                                                   end_date=END_DATE, randomized=True,
+                                                                   collection_ids=[COLLECTION_US_NATIONAL],
+                                                                   pagination_token=next_page_token1)
+        assert len(results2) == 1000
+        assert next_page_token2 is not None
+        assert next_page_token1 != next_page_token2
+        page1_ids = [s['id'] for s in results1]
+        page2_ids = [s['id'] for s in results2]
+        common_ids = set(page1_ids) & set(page2_ids)
+        self.assertEqual(len(common_ids), 0)
 
     def test_random_sample(self):
         def _test_random_sample(sample_size: int):
@@ -139,7 +177,20 @@ def _test_random_sample(sample_size: int):
                 assert 'text' not in s.keys()
         _test_random_sample(934)
         _test_random_sample(123)
-        # TO DO: add admin test that passed in `expanded=True` and verifies `text` is in returned item properties
+
+    def test_story_list_random_expanded(self):
+        # note - requires staff API token
+        page, _ = self._admin_search.story_list(query="weather", start_date=START_DATE, end_date=END_DATE,
+                                                collection_ids=[COLLECTION_US_NATIONAL], randomized=True)
+        for story in page:
+            assert 'text' not in story
+        time.sleep(25)
+        page, _ = self._admin_search.story_list(query="weather", start_date=START_DATE, end_date=END_DATE,
+                                                expanded=True, collection_ids=[COLLECTION_US_NATIONAL],
+                                                randomized=True)
+        for story in page:
+            assert 'text' in story
+            assert len(story['text']) > 0
 
     def test_story_list_expanded(self):
         # note - requires staff API token