test: add live coverage for pagination cleanups

Author: subzeroid

.github/workflows/live-account-tests.yml

@@ -13,6 +13,8 @@ on:
           - upload
           - direct
           - timeline
+          - location
+          - usertag
           - user
           - all
 
@@ -56,6 +58,14 @@ jobs:
         if: github.event.inputs.test_target == 'timeline' || github.event.inputs.test_target == 'all'
         run: pytest -sv tests/live/test_timeline.py::ClientTimelineLiveTestCase
 
+      - name: Run location test
+        if: github.event.inputs.test_target == 'location' || github.event.inputs.test_target == 'all'
+        run: pytest -sv tests/live/test_location.py::ClientLocationPaginationLiveTestCase
+
+      - name: Run usertag pagination test
+        if: github.event.inputs.test_target == 'usertag' || github.event.inputs.test_target == 'all'
+        run: pytest -sv tests/live/test_user.py::ClientUsertagPaginationLiveTestCase
+
       - name: Run user test
         if: github.event.inputs.test_target == 'user' || github.event.inputs.test_target == 'all'
         run: pytest -sv tests/live/test_user.py::ClientUserTestCase

tests/live/test_location.py

@@ -1,3 +1,5 @@
+import base64
+
 from tests import helpers as _helpers
 from tests.helpers import *
 
@@ -88,3 +90,38 @@ def test_location_medias_recent(self):
         medias = self.cl.location_medias_recent(197780767581661, amount=2)
         self.assertEqual(len(medias), 2)
         self.assertIsInstance(medias[0], Media)
+
+
+class ClientLocationPaginationLiveTestCase(_helpers.ClientPrivateTestCase):
+    def __init__(self, *args, **kwargs):
+        self.cl = None
+        return unittest.TestCase.__init__(self, *args, **kwargs)
+
+    def setup_method(self, *args, **kwargs):
+        return None
+
+    def setUp(self):
+        if not TEST_ACCOUNTS_URL:
+            self.skipTest("TEST_ACCOUNTS_URL is required for location pagination live tests")
+        try:
+            self.cl = self.fresh_account()
+        except RuntimeError as exc:
+            self.skipTest(str(exc))
+
+    def test_location_medias_v1_chunk_live_cursor_shape(self):
+        medias, max_id = self.cl.location_medias_v1_chunk(197780767581661, tab_key="recent")
+        self.assertIsInstance(medias, list)
+        if medias:
+            self.assertIsInstance(medias[0], Media)
+        if not max_id:
+            return
+
+        next_max_id, page, media_ids = json.loads(base64.b64decode(max_id))
+        self.assertTrue(next_max_id)
+        self.assertIsInstance(page, int)
+        self.assertIsInstance(media_ids, list)
+
+        next_medias, _ = self.cl.location_medias_v1_chunk(197780767581661, tab_key="recent", max_id=max_id)
+        self.assertIsInstance(next_medias, list)
+        if next_medias:
+            self.assertIsInstance(next_medias[0], Media)

tests/live/test_user.py

@@ -10,6 +10,58 @@ def test_user_followers(self):
         self.assertIsInstance(list(followers.values())[0], UserShort)
 
 
+class ClientUsertagPaginationLiveTestCase(_helpers.ClientPrivateTestCase):
+    def __init__(self, *args, **kwargs):
+        self.cl = None
+        return unittest.TestCase.__init__(self, *args, **kwargs)
+
+    def setup_method(self, *args, **kwargs):
+        return None
+
+    def setUp(self):
+        if not TEST_ACCOUNTS_URL:
+            self.skipTest("TEST_ACCOUNTS_URL is required for usertag pagination live tests")
+        try:
+            self.cl = self.fresh_account()
+        except RuntimeError as exc:
+            self.skipTest(str(exc))
+
+    def assertTaggedMediaPage(self, medias, amount):
+        self.assertGreater(len(medias), 0)
+        self.assertLessEqual(len(medias), amount)
+        media = medias[0]
+        self.assertIsInstance(media, Media)
+        for field in REQUIRED_MEDIA_FIELDS:
+            self.assertTrue(hasattr(media, field))
+
+    def assertNoDuplicateMediasAcrossPages(self, first_page, next_page):
+        first_ids = {media.pk for media in first_page}
+        next_ids = {media.pk for media in next_page}
+        self.assertTrue(first_ids.isdisjoint(next_ids), f"Duplicate medias across pages: {first_ids & next_ids}")
+
+    def test_usertag_medias_paginated_live(self):
+        user_id = self.user_id_from_username("instagram")
+
+        first_page, end_cursor = self.cl.usertag_medias_paginated(user_id, amount=2)
+        self.assertTaggedMediaPage(first_page, 2)
+        self.assertTrue(end_cursor)
+
+        next_page, _ = self.cl.usertag_medias_paginated(user_id, amount=2, end_cursor=end_cursor)
+        self.assertTaggedMediaPage(next_page, 2)
+        self.assertNoDuplicateMediasAcrossPages(first_page, next_page)
+
+    def test_usertag_medias_paginated_v1_live(self):
+        user_id = self.user_id_from_username("instagram")
+
+        first_page, end_cursor = self.cl.usertag_medias_paginated_v1(user_id, amount=2)
+        self.assertTaggedMediaPage(first_page, 2)
+        self.assertTrue(end_cursor)
+
+        next_page, _ = self.cl.usertag_medias_paginated_v1(user_id, amount=2, end_cursor=end_cursor)
+        self.assertTaggedMediaPage(next_page, 2)
+        self.assertNoDuplicateMediasAcrossPages(first_page, next_page)
+
+
 class ClientUserExtendTestCase(_helpers.ClientPrivateTestCase):
     def test_username_from_user_id(self):
         self.assertEqual(self.cl.username_from_user_id(25025320), "instagram")