test: harden suite after adversarial review (faithfulness + gaps)

Acting on a two-reviewer audit (workflow + Opus-5 advisor):

Faithfulness fixes (tests that could pass against a broken library):
- get_bitstreams embedded-link test used an href byte-identical to the
  fallback URL, so it could not distinguish the branches; give it a distinct
  href that the fallback could not produce.
- MCP chain fed a link-bearing Bundle to get_bitstreams; the real consumer
  round-trips through as_dict() (drops _links) and hits the fallback URL -
  rebuild the Bundle from as_dict() and assert the fallback is used.
- export chain fetched policies with action='READ' (raw-client default) but
  the real exporter goes through a wrapper defaulting to action=None (no
  filter); call with action=None and assert no action param is sent.
- get_resourcepolicy empty test used a no-_embedded body (defensive branch);
  the live API returns an _embedded envelope even when empty - use that.

Coverage / stronger assertions:
- get_resourcepolicy action=None omits the filter and returns all actions.
- create_item now asserts the POST body (name/metadata/type/flags), not just
  the uuid; create_bitstream asserts the multipart 'properties' payload
  (name/bundleName/metadata).
- search_objects result now asserts .as_dict() and links['self']['href'],
  the two accessors every consumer reads.
- model tests assert parsed .metadata and checkSum.checkSumAlgorithm (dropped
  the tautological hard-set .type assertions' reliance).
- get_bitstreams non-200, and create_item/create_bundle server-error:
  characterization tests pinning the current (non-fail-safe) behavior the
  consumers depend on, flagged in-comment for a future library hardening.

60 tests, still no network.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: jm

requirements-test.txt

@@ -1,5 +1,7 @@
 # Test-only dependencies for the dspace_rest_client suite.
-# The library itself only needs `requests` (see setup.py); these add the test
-# runner and an HTTP transport mock so tests never touch a real DSpace server.
+# The test runner and an HTTP transport mock so tests never touch a real DSpace
+# server. `requests` is listed explicitly so `pip install -r requirements-test.txt`
+# alone is enough to import the package (CI also `pip install .`s it via setup.py).
 pytest>=7.0
 requests-mock>=1.11
+requests

tests/_helpers.py

@@ -7,7 +7,9 @@
 response parsing - the two things downstream code (this repo) depends on -
 fails a test instead of silently shipping.
 """
+import json
 import os
+import re
 import sys
 from urllib.parse import urlparse, parse_qs
 
@@ -52,6 +54,21 @@ def sent_params(request) -> dict:
     return parse_qs(urlparse(request.url).query)
 
 
+def multipart_properties(request) -> dict:
+    """Parse the JSON ``properties`` part of a create_bitstream multipart body.
+
+    ``create_bitstream`` sends ``properties = json.dumps({name, metadata,
+    bundleName}) + ';application/json'`` as a form field. This is what actually
+    carries the bitstream's metadata to DSpace, so tests assert on it.
+    """
+    body = request.body
+    if isinstance(body, bytes):
+        body = body.decode("utf-8", "replace")
+    m = re.search(r'name="properties"\r?\n\r?\n(.*?);application/json',
+                  body, re.DOTALL)
+    return json.loads(m.group(1)) if m else None
+
+
 # --- response-body builders (shape mirrors the DSpace 7 REST API) --------- #
 
 def embedded(key: str, resources: list) -> dict:

tests/test_client_read.py

@@ -20,10 +20,14 @@ class TestSearchObjects(unittest.TestCase):
 
     def test_builds_url_params_and_parses_objects(self):
         c = make_client()
+        # every caller's next step is dso.as_dict() + dso.links['self']['href']
+        # (repo._search.dso2dict, mcp._dso_to_dict), so the search result must
+        # carry both - include a self link to prove it survives parsing.
+        obj1 = item_json("u1", "A", _links={"self": {"href": f"{API}/items/u1"}})
         body = {"_embedded": {"searchResult": {
             "page": {"totalElements": 2, "size": 100},
             "_embedded": {"objects": [
-                {"_embedded": {"indexableObject": item_json("u1", "A")}},
+                {"_embedded": {"indexableObject": obj1}},
                 {"_embedded": {"indexableObject": item_json("u2", "B")}},
             ]}}}}
         with requests_mock.Mocker() as m:
@@ -32,6 +36,9 @@ def test_builds_url_params_and_parses_objects(self):
             res = c.search_objects(query="dc.identifier:123", size=100,
                                    page=0, details=details)
             self.assertEqual([d.uuid for d in res], ["u1", "u2"])
+            # the two accessors every consumer reads off a search hit
+            self.assertEqual(res[0].links["self"]["href"], f"{API}/items/u1")
+            self.assertEqual(res[0].as_dict()["uuid"], "u1")
             p = sent_params(m.last_request)
             self.assertEqual(p["query"], ["dc.identifier:123"])
             self.assertEqual(p["size"], ["100"])
@@ -132,30 +139,53 @@ class TestGetBitstreams(unittest.TestCase):
 
     def test_by_bundle_uses_embedded_link(self):
         c = make_client()
-        href = f"{API}/core/bundles/bnd/bitstreams"
+        # href deliberately NOT equal to the fallback URL (.../bundles/bnd/
+        # bitstreams): if the embedded-link branch were removed the client would
+        # build the fallback, which is unmocked, and this test would fail.
+        href = f"{API}/core/bundles/HREF-ONLY-PATH/bitstreams"
         bundle = Bundle(bundle_json("bnd", bitstreams_href=href))
         body = embedded("bitstreams", [bitstream_json("s1", "a.pdf", size=10)])
         with requests_mock.Mocker() as m:
             m.get(href, json=body)
             bs = c.get_bitstreams(bundle=bundle, size=500)
             self.assertEqual([b.uuid for b in bs], ["s1"])
             self.assertEqual(bs[0].sizeBytes, 10)
+            self.assertEqual(m.last_request.url.split("?")[0], href)
             self.assertEqual(sent_params(m.last_request)["size"], ["500"])
 
     def test_by_bundle_without_link_constructs_url(self):
         c = make_client()
         bundle = Bundle(bundle_json("bnd2"))  # no _links -> manual URL
         with requests_mock.Mocker() as m:
             m.get(f"{API}/core/bundles/bnd2/bitstreams",
-                  json=embedded("bitstreams", []))
-            self.assertEqual(c.get_bitstreams(bundle=bundle), [])
+                  json=embedded("bitstreams",
+                                [bitstream_json("s9", "x.pdf", size=7)]))
+            bs = c.get_bitstreams(bundle=bundle)
+            # proves both the constructed URL AND parsing on the fallback path
+            self.assertEqual([b.uuid for b in bs], ["s9"])
+            self.assertEqual(bs[0].sizeBytes, 7)
 
     def test_no_args_returns_empty_list(self):
         c = make_client()
         with requests_mock.Mocker() as m:
             self.assertEqual(c.get_bitstreams(), [])
             self.assertEqual(m.call_count, 0)
 
+    def test_non_200_currently_raises_no_failsafe(self):
+        # CHARACTERIZATION of a known sharp edge: unlike get_bundles (which since
+        # PR #16 returns [] on a 404), get_bitstreams has no fail-safe - a non-200
+        # makes fetch_resource return None which this method then subscripts, so
+        # it raises. The consumers (export/_dspace, reposync/_files) iterate the
+        # result unguarded, so this is a real crash risk. Pinned deliberately: if
+        # the library is hardened to return [], update this test to assert that.
+        c = make_client()
+        bundle = Bundle(bundle_json("bnd2"))
+        with requests_mock.Mocker() as m:
+            m.get(f"{API}/core/bundles/bnd2/bitstreams",
+                  status_code=500, text="boom")
+            with self.assertRaises(Exception):
+                c.get_bitstreams(bundle=bundle)
+
 
 class TestGetCollections(unittest.TestCase):
 
@@ -220,11 +250,28 @@ def test_parses_live_policies_and_sends_uuid_action(self):
             self.assertEqual(p["uuid"], [BITSTREAM_UUID])
             self.assertEqual(p["action"], ["READ"])
 
-    def test_no_embedded_returns_empty_list(self):
+    def test_action_none_omits_the_action_filter(self):
+        # The bitstream export path calls this via the ingest wrapper whose
+        # default is action=None (ingest/_dspace.py get_resourcepolicy), which
+        # must fetch policies of ALL actions - so no `action` param is sent.
+        c = make_client()
+        body = embedded("resourcepolicies", [
+            policy_json(pid=1, action="READ"),
+            policy_json(pid=2, action="WRITE")])
+        with requests_mock.Mocker() as m:
+            m.get(f"{API}/authz/resourcepolicies/search/resource", json=body)
+            rps = c.get_resourcepolicy(BITSTREAM_UUID, action=None)
+            self.assertEqual([rp.action for rp in rps], ["READ", "WRITE"])
+            p = sent_params(m.last_request)
+            self.assertEqual(p["uuid"], [BITSTREAM_UUID])
+            self.assertNotIn("action", p)
+
+    def test_empty_result_set_returns_empty_list(self):
+        # The live endpoint returns an _embedded envelope even when empty.
         c = make_client()
         with requests_mock.Mocker() as m:
             m.get(f"{API}/authz/resourcepolicies/search/resource",
-                  json={"page": {"totalElements": 0}})
+                  json=embedded("resourcepolicies", []))
             self.assertEqual(c.get_resourcepolicy(BITSTREAM_UUID), [])
 
     def test_invalid_uuid_returns_none_without_request(self):

tests/test_client_write.py

@@ -14,8 +14,9 @@
 
 import _helpers  # noqa: F401
 from _helpers import (
-    make_client, sent_params, bundle_json, bitstream_json, item_json,
-    API, ITEM_UUID, COLLECTION_UUID, BITSTREAM_UUID, ANON_GROUP_UUID)
+    make_client, sent_params, multipart_properties, bundle_json,
+    bitstream_json, item_json, API, ITEM_UUID, COLLECTION_UUID,
+    BITSTREAM_UUID, ANON_GROUP_UUID)
 from dspace_rest_client.models import Item, Bundle, Bitstream
 
 
@@ -100,6 +101,21 @@ def test_posts_to_item_bundles_and_returns_bundle(self):
     def test_none_parent_returns_none(self):
         self.assertIsNone(make_client().create_bundle(parent=None))
 
+    def test_server_error_returns_truthy_uuidless_bundle(self):
+        # CHARACTERIZATION: like create_item, create_bundle wraps the response
+        # unconditionally -> a truthy Bundle with uuid=None on failure, not None.
+        # The importer's `if not bundle` guard (reposync/_importer.py:118-120)
+        # never fires because a Bundle instance is always truthy. Pinned.
+        c = make_client()
+        parent = Item(item_json(ITEM_UUID))
+        with requests_mock.Mocker() as m:
+            m.post(f"{API}/core/items/{ITEM_UUID}/bundles",
+                   status_code=500, text="boom")
+            out = c.create_bundle(parent=parent)
+            self.assertIsInstance(out, Bundle)
+            self.assertIsNone(out.uuid)
+            self.assertTrue(out)
+
 
 class TestCreateItem(unittest.TestCase):
 
@@ -114,6 +130,27 @@ def test_posts_with_owning_collection_param_and_returns_item(self):
             self.assertEqual(out.uuid, "newu")
             self.assertEqual(sent_params(m.last_request)["owningCollection"],
                              [COLLECTION_UUID])
+            # the POST body is item.as_dict() - this is how the importer's built
+            # metadata actually reaches DSpace, so pin it, not just the uuid.
+            body = m.last_request.json()
+            self.assertEqual(body["name"], "New thesis")
+            self.assertEqual(body["type"], "item")
+            self.assertEqual(body["metadata"], {})
+            self.assertIs(body["inArchive"], True)
+
+    def test_server_error_returns_truthy_uuidless_item(self):
+        # CHARACTERIZATION: create_item wraps the response unconditionally, so a
+        # failed create yields a truthy Item with uuid=None, NOT None. The
+        # importer guards with `if dso is None` (reposync/_importer.py:127-129),
+        # which therefore never fires on failure. Pinned; see the fail-safe note.
+        c = make_client()
+        item = Item({"name": "x", "metadata": {}})
+        with requests_mock.Mocker() as m:
+            m.post(f"{API}/core/items", status_code=500, text="boom")
+            out = c.create_item(parent=COLLECTION_UUID, item=item)
+            self.assertIsInstance(out, Item)
+            self.assertIsNone(out.uuid)
+            self.assertTrue(out)  # truthy despite the failure
 
     def test_non_item_returns_none(self):
         self.assertIsNone(make_client().create_item(
@@ -145,9 +182,15 @@ def test_success_multipart_upload_returns_bitstream(self):
             self.assertIsInstance(bs, Bitstream)
             self.assertEqual(bs.uuid, "bsnew")
             self.assertEqual(bs.sizeBytes, 20)
-            # the request really was a multipart file upload
+            # the request really was a multipart file upload...
             self.assertIn("multipart/form-data",
                           m.last_request.headers["Content-Type"])
+            # ...carrying the name/bundleName/metadata that actually attach the
+            # bitstream's metadata in DSpace (reposync/_utils.create_new_bitstream)
+            props = multipart_properties(m.last_request)
+            self.assertEqual(props["name"], "a.pdf")
+            self.assertEqual(props["bundleName"], "ORIGINAL")  # == bundle.name
+            self.assertEqual(props["metadata"], {"dc.title": [{"value": "a.pdf"}]})
 
     def test_server_error_returns_none(self):
         c = make_client()

tests/test_models.py

@@ -73,9 +73,12 @@ def test_bundle_fields_and_bitstreams_link(self):
         # get_bitstreams(bundle=...) prefers this embedded link over a manually
         # constructed URL, so it is part of the contract.
         b = Bundle({"uuid": "b1", "name": "ORIGINAL", "type": "bundle",
-                    "metadata": {},
+                    "metadata": {"dc.title": [{"value": "ORIGINAL"}]},
                     "_links": {"bitstreams": {"href": "http://x/bundles/b1/bitstreams"}}})
         self.assertEqual((b.uuid, b.name, b.type), ("b1", "ORIGINAL", "bundle"))
+        # .metadata is parsed from the response (unlike .type, a class constant)
+        # and is serialised by export/_dspace.py:323, so pin it.
+        self.assertEqual(b.metadata, {"dc.title": [{"value": "ORIGINAL"}]})
         self.assertEqual(b.links["bitstreams"]["href"],
                          "http://x/bundles/b1/bitstreams")
 
@@ -90,6 +93,10 @@ def test_bitstream_file_fields(self):
         self.assertEqual(b.sizeBytes, 2048)
         self.assertEqual(b.sequenceId, 3)
         self.assertEqual(b.checkSum["value"], "deadbeef")
+        # the checksum verifier compares checkSumAlgorithm == "MD5"
+        # (reposync/_files.py:187-189); .metadata is serialised by the exporter.
+        self.assertEqual(b.checkSum["checkSumAlgorithm"], "MD5")
+        self.assertEqual(b.metadata, {"dc.title": [{"value": "f.pdf"}]})
         d = b.as_dict()
         self.assertEqual(d["sizeBytes"], 2048)
         self.assertEqual(d["checkSum"]["value"], "deadbeef")

tests/test_repo_usage_contract.py

@@ -15,9 +15,9 @@
 
 import _helpers  # noqa: F401
 from _helpers import (
-    make_client, embedded, item_json, bundle_json, bitstream_json,
+    make_client, sent_params, embedded, item_json, bundle_json, bitstream_json,
     policy_json, API, ITEM_UUID, BITSTREAM_UUID, ANON_GROUP_UUID)
-from dspace_rest_client.models import Item
+from dspace_rest_client.models import Item, Bundle
 
 
 class TestBitstreamExportChain(unittest.TestCase):
@@ -33,15 +33,19 @@ def test_full_chain_yields_serialisable_attributes(self):
         c = make_client()
         item = Item(item_json(ITEM_UUID))
         bits_href = f"{API}/core/bundles/{self.BUNDLE_UUID}/bitstreams"
+        bundle_md = {"dc.title": [{"value": "ORIGINAL"}]}
+        bs_md = {"dc.description": [{"value": "VŠKP"}]}
         with requests_mock.Mocker() as m:
             m.get(f"{API}/core/items/{ITEM_UUID}/bundles",
                   json=embedded("bundles",
                                 [bundle_json(self.BUNDLE_UUID, "ORIGINAL",
-                                             bitstreams_href=bits_href)]))
+                                             bitstreams_href=bits_href,
+                                             metadata=bundle_md)]))
             m.get(bits_href,
                   json=embedded("bitstreams",
                                 [bitstream_json(BITSTREAM_UUID, "thesis.pdf",
-                                                size=123, seq=1, checksum="abc")]))
+                                                size=123, seq=1, checksum="abc",
+                                                metadata=bs_md)]))
             m.get(f"{API}/authz/resourcepolicies/search/resource",
                   json=embedded("resourcepolicies", [policy_json(pid=1)]))
 
@@ -50,20 +54,26 @@ def test_full_chain_yields_serialisable_attributes(self):
             bundle = bundles[0]
             self.assertEqual((bundle.name, bundle.uuid, bundle.type),
                              ("ORIGINAL", self.BUNDLE_UUID, "bundle"))
+            self.assertEqual(bundle.metadata, bundle_md)
 
-            bundle_rp = c.get_resourcepolicy(bundle.uuid)
+            # the exporter calls get_resourcepolicy via a wrapper defaulting to
+            # action=None (ingest/_dspace.py), so NO action filter is sent.
+            bundle_rp = c.get_resourcepolicy(bundle.uuid, action=None)
             self.assertEqual([rp.as_dict()["groupName"] for rp in bundle_rp],
                              ["Anonymous"])
+            self.assertNotIn("action", sent_params(m.last_request))
 
             bitstreams = c.get_bitstreams(bundle=bundle, size=1000)
             self.assertEqual(len(bitstreams), 1)
             b = bitstreams[0]
-            # the exporter reads exactly these off each bitstream
+            # a representative set of the attributes the exporter serialises
+            # (src/export/_dspace.py:333-340) - name/uuid/size/seq/checksum/meta
             self.assertEqual((b.name, b.uuid, b.sizeBytes, b.sequenceId),
                              ("thesis.pdf", BITSTREAM_UUID, 123, 1))
             self.assertEqual(b.checkSum["value"], "abc")
+            self.assertEqual(b.metadata, bs_md)
 
-            bs_rp = c.get_resourcepolicy(b.uuid)
+            bs_rp = c.get_resourcepolicy(b.uuid, action=None)
             self.assertEqual(bs_rp[0].as_dict()["groupUUID"], ANON_GROUP_UUID)
 
 
@@ -104,7 +114,11 @@ class TestMcpBundleWalk(unittest.TestCase):
 
     def test_lookup_then_item_bundles_bitstreams(self):
         c = make_client()
-        bits_href = f"{API}/core/bundles/bnd/bitstreams"
+        # mcp/core.py drops _links (it round-trips bundles through as_dict), so
+        # get_bitstreams must use the manually-constructed fallback URL, not the
+        # embedded link. The link below is a decoy that must NOT be requested.
+        decoy_href = f"{API}/core/bundles/DECOY-LINK/bitstreams"
+        fallback = f"{API}/core/bundles/bnd/bitstreams"
         with requests_mock.Mocker() as m:
             m.get(f"{API}/discover/search/objects", json={"_embedded": {
                 "searchResult": {"page": {"totalElements": 1},
@@ -115,8 +129,8 @@ def test_lookup_then_item_bundles_bitstreams(self):
             m.get(f"{API}/core/items/{ITEM_UUID}/bundles",
                   json=embedded("bundles",
                                 [bundle_json("bnd", "ORIGINAL",
-                                             bitstreams_href=bits_href)]))
-            m.get(bits_href,
+                                             bitstreams_href=decoy_href)]))
+            m.get(fallback,
                   json=embedded("bitstreams", [bitstream_json("bs1", "a.pdf")]))
 
             matches = c.search_objects(query="dc.identifier:42")
@@ -130,8 +144,13 @@ def test_lookup_then_item_bundles_bitstreams(self):
             bundles = c.get_bundles(parent=parent, size=200)
             self.assertEqual(bundles[0].name, "ORIGINAL")
 
-            bitstreams = c.get_bitstreams(bundle=bundles[0], size=500)
+            # mirror mcp: rebuild the Bundle from as_dict() (which strips _links)
+            # so get_bitstreams takes the fallback-URL branch the consumer hits.
+            bstub = Bundle(bundles[0].as_dict())
+            self.assertNotIn("bitstreams", bstub.links)
+            bitstreams = c.get_bitstreams(bundle=bstub, size=500)
             self.assertEqual(bitstreams[0].uuid, "bs1")
+            self.assertEqual(m.last_request.url.split("?")[0], fallback)
 
 
 class TestGroupSearchForUuidResolution(unittest.TestCase):