fix(client): follow pagination and verify counts in list getters

dspace_rest_client/client.py

@@ -544,6 +544,52 @@ def fetch_resource(self, url, params=None):
         # ValueError / JSON handling moved to static method
         return parse_json(r)
 
+    def _fetch_all_pages(self, url, params, key, what):
+        """
+        Fetch a HAL list with every page from params['page'] (default 0) on, and
+        verify the collected count against the server's page.totalElements.
+        @param url:     DSpace REST API URL
+        @param params:  query params of the first request, reused for later pages
+        @param key:     name of the list under '_embedded', eg. 'bundles'
+        @param what:    noun for error messages, eg. 'bundles'
+        @return:        list of raw JSON resources, or None when the FIRST request
+                        failed (self._last_err holds it; the caller decides)
+        @raise RuntimeError: a later page failed, or fewer/more elements arrived
+                        than the server reported - never a silently partial list
+        """
+        r_json = self.fetch_resource(url, params=params)
+        if r_json is None:
+            return None
+        resources = list((r_json.get('_embedded') or {}).get(key) or [])
+        page = r_json.get('page')
+        if page is None:
+            return resources
+        first = page.get('number', 0)
+        # bounded by the first response, so a server repeating a page cannot loop
+        for number in range(first + 1, page.get('totalPages', 0)):
+            r_json = self.fetch_resource(url, params={**params, 'page': number})
+            if r_json is None:
+                status = getattr(self._last_err, 'status_code', None)
+                raise RuntimeError(
+                    f'Failed to fetch {what} page {number}: HTTP {status} [{url}]')
+            # a server ignoring `page` repeats an earlier page: the count could
+            # still match totalElements while the real page is missing
+            got = (r_json.get('page') or {}).get('number')
+            if got is not None and got != number:
+                raise RuntimeError(
+                    f'Inconsistent {what}: asked for page {number}, got page {got} '
+                    f'[{url}]')
+            resources.extend((r_json.get('_embedded') or {}).get(key) or [])
+        total = page.get('totalElements')
+        if total is None:
+            return resources
+        expected = max(total - first * page.get('size', 0), 0)
+        if len(resources) != expected:
+            raise RuntimeError(
+                f'Incomplete {what}: collected {len(resources)} of {expected} '
+                f'[{url}]')
+        return resources
+
     def get_resourcepolicy(self, uuid, action='READ'):
         """
         Fetch resource policies for a given resource UUID and action.
@@ -558,13 +604,12 @@ def get_resourcepolicy(self, uuid, action='READ'):
             params = {'uuid': uuid}
             if action is not None:
                 params['action'] = action
-            r_json = self.fetch_resource(url, params=params)
-            if r_json is None:
+            arr = self._fetch_all_pages(url, params, 'resourcepolicies',
+                                        'resource policies')
+            if arr is None:
                 return None
-            if '_embedded' not in r_json:
+            if not arr:
                 _logger.debug(f"No resource policies found for resource UUID: {uuid} [{url}]")
-                return []
-            arr = r_json['_embedded'].get('resourcepolicies') or []
             return [ResourcePolicy(x) for x in arr]
         except ValueError as e:
             _logger.error(f'Invalid resource UUID: {uuid} - {e}')
@@ -761,8 +806,11 @@ def get_bundles(self, parent=None, uuid=None, page=0, size=20, sort=None):
             params['page'] = page
         if sort is not None:
             params['sort'] = sort
-        r_json = self.fetch_resource(url, params=params)
-        if r_json is None:
+        if single_result:
+            result = self.fetch_resource(url, params=params)
+        else:
+            result = self._fetch_all_pages(url, params, 'bundles', 'bundles')
+        if result is None:
             status = getattr(self._last_err, 'status_code', None)
             if status == 404:
                 # a deleted item (or bundle) simply has no bundles, which is a
@@ -775,10 +823,9 @@ def get_bundles(self, parent=None, uuid=None, page=0, size=20, sort=None):
             raise RuntimeError(f'Failed to fetch bundles: HTTP {status} [{url}]')
         try:
             if single_result:
-                bundles.append(Bundle(r_json))
+                bundles.append(Bundle(result))
             if not single_result:
-                resources = r_json['_embedded']['bundles']
-                for resource in resources:
+                for resource in result:
                     bundles.append(Bundle(resource))
         except ValueError as err:
             _logger.error(f'error parsing bundle results: {err}')
@@ -812,7 +859,7 @@ def get_bitstreams(self, uuid=None, bundle=None, page=0, size=20, sort=None):
         Get a specific bitstream UUID, or all bitstreams for a specific bundle
         @param uuid:    UUID of a specific bitstream to retrieve
         @param bundle:  A python Bundle object to parse for bitstream links to retrieve
-        @param page:    Page number, for pagination over large result sets (default: 0)
+        @param page:    First page to fetch; every later page is fetched too (default: 0)
         @param size:    Size of results per page (default: 20)
         @return:        list of python Bitstream objects
         """
@@ -833,8 +880,8 @@ def get_bitstreams(self, uuid=None, bundle=None, page=0, size=20, sort=None):
             params['page'] = page
         if sort is not None:
             params['sort'] = sort
-        r_json = self.fetch_resource(url, params=params)
-        if r_json is None:
+        resources = self._fetch_all_pages(url, params, 'bitstreams', 'bitstreams')
+        if resources is None:
             status = getattr(self._last_err, 'status_code', None)
             if status == 404:
                 # the bundle (or item) is gone - no bitstreams, a clean empty
@@ -844,11 +891,7 @@ def get_bitstreams(self, uuid=None, bundle=None, page=0, size=20, sort=None):
             # a transient 5xx must NOT masquerade as "no bitstreams"; surface it
             # with status + url so the caller can retry, not an opaque TypeError.
             raise RuntimeError(f'Failed to fetch bitstreams: HTTP {status} [{url}]')
-        bitstreams = list()
-        if '_embedded' in r_json and 'bitstreams' in r_json['_embedded']:
-            for bitstream_resource in r_json['_embedded']['bitstreams']:
-                bitstreams.append(Bitstream(bitstream_resource))
-        return bitstreams
+        return [Bitstream(bitstream_resource) for bitstream_resource in resources]
 
     def create_bitstream(self, bundle=None, name=None, path=None, mime=None, metadata=None, retry=False, timeout=None):
         """

tests/_helpers.py

@@ -89,6 +89,15 @@ def embedded(key: str, resources: list) -> dict:
     return {"_embedded": {key: resources}}
 
 
+def paged(key: str, resources: list, number: int, size: int,
+          total_elements: int) -> dict:
+    """One page of a HAL list, with the ``page`` object DSpace sends alongside."""
+    total_pages = -(-total_elements // size)
+    return {"_embedded": {key: resources},
+            "page": {"size": size, "totalElements": total_elements,
+                     "totalPages": total_pages, "number": number}}
+
+
 def item_json(uuid: str = ITEM_UUID, name: str = "Thesis", **extra) -> dict:
     d = {"uuid": uuid, "name": name, "type": "item", "metadata": {},
          "inArchive": True, "discoverable": True, "withdrawn": False}

tests/test_client_read.py

@@ -13,7 +13,7 @@
 from _helpers import (
     make_client, sent_params, embedded, item_json, bundle_json,
     bitstream_json, policy_json, load_fixture, API, ITEM_UUID, BITSTREAM_UUID,
-    ANON_GROUP_UUID, no_warnings_logged)
+    ANON_GROUP_UUID, no_warnings_logged, paged)
 from dspace_rest_client.models import Item, Bundle, Collection, Community
 
 
@@ -147,6 +147,80 @@ def test_no_args_returns_empty_without_request(self):
             self.assertEqual(c.get_bundles(), [])
             self.assertEqual(m.call_count, 0)
 
+    def test_by_parent_follows_every_page(self):
+        # a truncated bundle list makes an item look like it has no ORIGINAL
+        c = make_client()
+        parent = Item(item_json(ITEM_UUID))
+        url = f"{API}/core/items/{ITEM_UUID}/bundles"
+        with requests_mock.Mocker() as m:
+            m.get(f"{url}?page=0", json=paged("bundles", [
+                bundle_json("b1", "ORIGINAL"), bundle_json("b2", "THUMBNAIL")],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", json=paged("bundles", [
+                bundle_json("b3", "LICENSE")], number=1, size=2, total_elements=3))
+            bundles = c.get_bundles(parent=parent, size=2)
+            self.assertEqual([b.uuid for b in bundles], ["b1", "b2", "b3"])
+            self.assertEqual(
+                [sent_params(r) for r in m.request_history],
+                [{"size": ["2"], "page": ["0"]}, {"size": ["2"], "page": ["1"]}])
+
+    def test_single_complete_page_is_one_request(self):
+        c = make_client()
+        parent = Item(item_json(ITEM_UUID))
+        with requests_mock.Mocker() as m:
+            m.get(f"{API}/core/items/{ITEM_UUID}/bundles", json=paged("bundles", [
+                bundle_json("b1"), bundle_json("b2")],
+                number=0, size=20, total_elements=2))
+            self.assertEqual([b.uuid for b in c.get_bundles(parent=parent)],
+                             ["b1", "b2"])
+            self.assertEqual(m.call_count, 1)
+
+    def test_fewer_elements_than_total_raises(self):
+        # the server's own count proves elements are missing
+        c = make_client()
+        parent = Item(item_json(ITEM_UUID))
+        url = f"{API}/core/items/{ITEM_UUID}/bundles"
+        with requests_mock.Mocker() as m:
+            m.get(f"{url}?page=0", json=paged("bundles", [
+                bundle_json("b1"), bundle_json("b2")],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", json=paged("bundles", [],
+                                              number=1, size=2, total_elements=3))
+            with self.assertRaises(RuntimeError) as ctx:
+                c.get_bundles(parent=parent, size=2)
+            msg = str(ctx.exception)
+            self.assertIn("collected 2 of 3", msg)
+            self.assertIn(url, msg)
+
+    def test_server_ignoring_the_page_param_raises(self):
+        # page 0 sent twice: 2 + 2 matches totalElements 4, but the list would
+        # hold duplicates and miss the real second page
+        c = make_client()
+        parent = Item(item_json(ITEM_UUID))
+        url = f"{API}/core/items/{ITEM_UUID}/bundles"
+        with requests_mock.Mocker() as m:
+            m.get(url, json=paged("bundles", [bundle_json("b1"), bundle_json("b2")],
+                                  number=0, size=2, total_elements=4))
+            with self.assertRaises(RuntimeError) as ctx:
+                c.get_bundles(parent=parent, size=2)
+            msg = str(ctx.exception)
+            self.assertIn("asked for page 1, got page 0", msg)
+            self.assertIn(url, msg)
+
+    def test_error_on_later_page_raises_not_partial_list(self):
+        c = make_client()
+        parent = Item(item_json(ITEM_UUID))
+        url = f"{API}/core/items/{ITEM_UUID}/bundles"
+        with requests_mock.Mocker() as m:
+            m.get(f"{url}?page=0", json=paged("bundles", [
+                bundle_json("b1"), bundle_json("b2")],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", status_code=503, text="busy")
+            with self.assertRaises(RuntimeError) as ctx:
+                c.get_bundles(parent=parent, size=2)
+            self.assertIn("503", str(ctx.exception))
+            self.assertIn(url, str(ctx.exception))
+
 
 class TestGetBitstreams(unittest.TestCase):
 
@@ -217,6 +291,55 @@ def test_200_without_bitstreams_returns_empty_list(self):
                   json={"page": {"totalElements": 0}})
             self.assertEqual(c.get_bitstreams(bundle=bundle), [])
 
+    def test_by_bundle_follows_every_page(self):
+        c = make_client()
+        bundle = Bundle(bundle_json("bnd2"))
+        url = f"{API}/core/bundles/bnd2/bitstreams"
+        with requests_mock.Mocker() as m:
+            m.get(f"{url}?page=0", json=paged("bitstreams", [
+                bitstream_json("s1"), bitstream_json("s2")],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", json=paged("bitstreams", [
+                bitstream_json("s3")], number=1, size=2, total_elements=3))
+            bs = c.get_bitstreams(bundle=bundle, size=2)
+            self.assertEqual([b.uuid for b in bs], ["s1", "s2", "s3"])
+            self.assertEqual(
+                [sent_params(r) for r in m.request_history],
+                [{"size": ["2"], "page": ["0"]}, {"size": ["2"], "page": ["1"]}])
+
+    def test_fewer_elements_than_total_raises(self):
+        # eg. a server capping the page size below the requested one
+        c = make_client()
+        bundle = Bundle(bundle_json("bnd2"))
+        url = f"{API}/core/bundles/bnd2/bitstreams"
+        with requests_mock.Mocker() as m:
+            m.get(f"{url}?page=0", json=paged("bitstreams", [
+                bitstream_json("s1"), bitstream_json("s2")],
+                number=0, size=2, total_elements=5))
+            m.get(f"{url}?page=1", json=paged("bitstreams", [
+                bitstream_json("s3"), bitstream_json("s4")],
+                number=1, size=2, total_elements=5))
+            m.get(f"{url}?page=2", json=paged("bitstreams", [],
+                                              number=2, size=2, total_elements=5))
+            with self.assertRaises(RuntimeError) as ctx:
+                c.get_bitstreams(bundle=bundle, size=2)
+            msg = str(ctx.exception)
+            self.assertIn("collected 4 of 5", msg)
+            self.assertIn(url, msg)
+
+    def test_error_on_later_page_raises_not_partial_list(self):
+        c = make_client()
+        bundle = Bundle(bundle_json("bnd2"))
+        url = f"{API}/core/bundles/bnd2/bitstreams"
+        with requests_mock.Mocker() as m:
+            m.get(f"{url}?page=0", json=paged("bitstreams", [
+                bitstream_json("s1"), bitstream_json("s2")],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", status_code=500, text="boom")
+            with self.assertRaises(RuntimeError) as ctx:
+                c.get_bitstreams(bundle=bundle, size=2)
+            self.assertIn("500", str(ctx.exception))
+
 
 class TestGetCollections(unittest.TestCase):
 
@@ -336,6 +459,55 @@ def test_fetch_failure_returns_none(self):
                   status_code=500, text="boom")
             self.assertIsNone(c.get_resourcepolicy(BITSTREAM_UUID))
 
+    def test_follows_every_page(self):
+        # a policy cut off by paging reads downstream as "the file has no such
+        # policy", eg. a missing Anonymous READ
+        c = make_client()
+        url = f"{API}/authz/resourcepolicies/search/resource"
+        with requests_mock.Mocker() as m:
+            # no page param on the first request, as before this paging fix
+            m.get(url, json=paged("resourcepolicies", [
+                policy_json(pid=1), policy_json(pid=2)],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", json=paged("resourcepolicies", [
+                policy_json(pid=3)], number=1, size=2, total_elements=3))
+            rps = c.get_resourcepolicy(BITSTREAM_UUID, action="READ")
+            self.assertEqual([rp.id for rp in rps], [1, 2, 3])
+            p0, p1 = [sent_params(r) for r in m.request_history]
+            self.assertEqual(p0, {"uuid": [BITSTREAM_UUID], "action": ["READ"]})
+            self.assertEqual(p1, {"uuid": [BITSTREAM_UUID], "action": ["READ"],
+                                  "page": ["1"]})
+
+    def test_missing_embedded_with_nonzero_total_raises(self):
+        c = make_client()
+        with requests_mock.Mocker() as m:
+            m.get(f"{API}/authz/resourcepolicies/search/resource",
+                  json={"page": {"size": 20, "totalElements": 1,
+                                 "totalPages": 1, "number": 0}})
+            with self.assertRaises(RuntimeError):
+                c.get_resourcepolicy(BITSTREAM_UUID)
+
+    def test_missing_embedded_with_zero_total_returns_empty_list(self):
+        c = make_client()
+        with requests_mock.Mocker() as m:
+            m.get(f"{API}/authz/resourcepolicies/search/resource",
+                  json={"page": {"size": 20, "totalElements": 0,
+                                 "totalPages": 0, "number": 0}})
+            self.assertEqual(c.get_resourcepolicy(BITSTREAM_UUID), [])
+
+    def test_error_on_later_page_raises_not_partial_list(self):
+        c = make_client()
+        url = f"{API}/authz/resourcepolicies/search/resource"
+        with requests_mock.Mocker() as m:
+            # no page param on the first request, as before this paging fix
+            m.get(url, json=paged("resourcepolicies", [
+                policy_json(pid=1), policy_json(pid=2)],
+                number=0, size=2, total_elements=3))
+            m.get(f"{url}?page=1", status_code=502, text="bad gateway")
+            with self.assertRaises(RuntimeError) as ctx:
+                c.get_resourcepolicy(BITSTREAM_UUID)
+            self.assertIn("502", str(ctx.exception))
+
 
 class TestFetchResource(unittest.TestCase):