test: unit + integration suite for downstream usage, with CI

.github/workflows/tests.yml

@@ -0,0 +1,40 @@
+name: Tests
+
+on:
+  push:
+    branches: [ dtq ]
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        # Matches the DSpace-ISstag-integration consumer CI matrix; the library
+        # itself declares support for >=3.8 (see setup.py).
+        python-version: ["3.10", "3.12"]
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+          cache-dependency-path: |
+            setup.py
+            requirements-test.txt
+
+      - name: Install package + test deps
+        run: |
+          python -m pip install --upgrade pip
+          pip install .
+          pip install -r requirements-test.txt
+
+      - name: Run tests
+        run: python -m pytest tests/ -v

.gitignore

@@ -1,6 +1,10 @@
 __pycache__/
 *.py[cod]
 *$py.class
+*.egg-info/
+build/
+dist/
+.pytest_cache/
 .python-version
 Pipfile.lock
 __pypackages__/

dspace_rest_client/client.py

@@ -764,7 +764,13 @@ def create_bundle(self, parent=None, name='ORIGINAL'):
         if parent is None:
             return None
         url = f'{self.API_ENDPOINT}/core/items/{parent.uuid}/bundles'
-        return Bundle(api_resource=parse_json(self.api_post(url, params=None, json={'name': name, 'metadata': {}})))
+        r = self.api_post(url, params=None, json={'name': name, 'metadata': {}})
+        if r.status_code not in (200, 201):
+            # return None on failure (not a uuid-less Bundle) so callers'
+            # `if not bundle` guards actually fire
+            _logger.error(f'Failed to create bundle: {r.status_code}: {r.text}')
+            return None
+        return Bundle(api_resource=parse_json(r))
 
     # PAGINATION
     def get_bitstreams(self, uuid=None, bundle=None, page=0, size=20, sort=None):
@@ -794,12 +800,18 @@ def get_bitstreams(self, uuid=None, bundle=None, page=0, size=20, sort=None):
         if sort is not None:
             params['sort'] = sort
         r_json = self.fetch_resource(url, params=params)
-        if '_embedded' in r_json:
-            if 'bitstreams' in r_json['_embedded']:
-                bitstreams = list()
-                for bitstream_resource in r_json['_embedded']['bitstreams']:
-                    bitstreams.append(Bitstream(bitstream_resource))
-                return bitstreams
+        if r_json is None and getattr(self._last_err, 'status_code', None) == 404:
+            # the bundle (or item) is gone - no bitstreams, a clean empty result
+            # rather than a crash. Mirrors get_bundles (#16). Any other failure
+            # (a transient 5xx, say) falls through and still surfaces to the
+            # caller so it is retried, not silently recorded as "no bitstreams".
+            _logger.info(f'No bitstreams: resource not found (404) [{url}]')
+            return list()
+        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
 
     def create_bitstream(self, bundle=None, name=None, path=None, mime=None, metadata=None, retry=False):
         """
@@ -1085,7 +1097,12 @@ def create_item(self, parent, item):
         if not isinstance(item, Item):
             _logger.error('Need a valid item')
             return None
-        return Item(api_resource=parse_json(self.create_dso(url, params=params, data=item.as_dict())))
+        r = self.create_dso(url, params=params, data=item.as_dict())
+        if r is None or r.status_code != 201:
+            # return None on failure (not a uuid-less Item) so callers'
+            # `if dso is None` guards actually fire
+            return None
+        return Item(api_resource=parse_json(r))
 
     def update_item(self, item):
         """

requirements-test.txt

@@ -0,0 +1,7 @@
+# Test-only dependencies for the dspace_rest_client suite.
+# 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

@@ -0,0 +1,114 @@
+"""
+Shared helpers for the client test-suite.
+
+The tests mock *only* the HTTP transport (via ``requests_mock``) and let the
+real ``DSpaceClient`` build URLs, send params and parse responses into model
+objects. That way a change to the library that breaks URL construction or
+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
+
+# Make ``dspace_rest_client`` importable when a test module is run directly
+# (``python tests/test_x.py``), not just under pytest (see conftest.py).
+_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+from dspace_rest_client.client import DSpaceClient  # noqa: E402
+
+# Canonical test endpoint. All mocked URLs are built off this so a typo shows
+# up as an unmatched request rather than a false pass.
+API = "http://dspace.test/server/api"
+
+# Real, syntactically-valid UUIDs - several client methods validate their UUID
+# arguments with ``uuid.UUID(...)`` and short-circuit on a bad one, so tests
+# that expect a request to actually go out must use valid values.
+ITEM_UUID = "11111111-1111-1111-1111-111111111111"
+COLLECTION_UUID = "22222222-2222-2222-2222-222222222222"
+BITSTREAM_UUID = "9f54ef33-c454-4d8e-a5fe-79d8291045ba"
+ANON_GROUP_UUID = "6ecfd145-3b7d-429e-ab31-ef6905a05763"
+
+
+def make_client(api_endpoint: str = API) -> DSpaceClient:
+    """A real client with no network touched.
+
+    ``DSpaceClient.__init__`` performs no HTTP (it only creates a
+    ``requests.Session`` and, optionally, a pysolr handle), so a plain
+    construction is safe and gives us the genuine object under test.
+    """
+    return DSpaceClient(api_endpoint, "tester@dspace.test", "secret")
+
+
+def sent_params(request) -> dict:
+    """Case-preserving query params of a captured request.
+
+    ``requests_mock``'s ``request.qs`` lowercases the whole query string, which
+    would mangle case-sensitive values (eg. ``action=READ``). Parsing the
+    original ``request.url`` keeps the real casing.
+    """
+    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)
+    # fail the test loudly rather than returning None and deferring the error
+    assert m is not None, \
+        "create_bitstream multipart body has no JSON 'properties' part"
+    return json.loads(m.group(1))
+
+
+# --- response-body builders (shape mirrors the DSpace 7 REST API) --------- #
+
+def embedded(key: str, resources: list) -> dict:
+    """A HAL ``_embedded`` list envelope, eg. ``{"_embedded": {"bundles": [...]}}``."""
+    return {"_embedded": {key: resources}}
+
+
+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}
+    d.update(extra)
+    return d
+
+
+def bundle_json(uuid: str = "bnd", name: str = "ORIGINAL",
+                bitstreams_href: str = None, **extra) -> dict:
+    d = {"uuid": uuid, "name": name, "type": "bundle", "metadata": {}}
+    if bitstreams_href is not None:
+        d["_links"] = {"bitstreams": {"href": bitstreams_href}}
+    d.update(extra)
+    return d
+
+
+def bitstream_json(uuid: str = "bs1", name: str = "thesis.pdf", size: int = 123,
+                   seq: int = 1, checksum: str = "abc", **extra) -> dict:
+    d = {"uuid": uuid, "name": name, "type": "bitstream", "metadata": {},
+         "sizeBytes": size, "sequenceId": seq,
+         "checkSum": {"checkSumAlgorithm": "MD5", "value": checksum}}
+    d.update(extra)
+    return d
+
+
+def policy_json(pid: int = 1, action: str = "READ", group_name: str = "Anonymous",
+                group_uuid: str = ANON_GROUP_UUID, start_date: str = None) -> dict:
+    """A resource policy in the *live* API shape (group under ``_embedded``)."""
+    d = {"id": pid, "action": action,
+         "_embedded": {"group": {"name": group_name, "uuid": group_uuid}}}
+    if start_date is not None:
+        d["startDate"] = start_date
+    return d

tests/conftest.py

@@ -0,0 +1,15 @@
+"""
+Pytest bootstrap for the dspace_rest_client test-suite.
+
+Ensures the in-tree ``dspace_rest_client`` package is importable when the tests
+are run straight from a checkout (``pytest tests/``) without a prior
+``pip install``. When the package *is* installed, inserting the source root at
+the front of ``sys.path`` means the tests still exercise the working-tree copy,
+which is the one we ship and vendor as a submodule.
+"""
+import os
+import sys
+
+_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)

tests/test_client_auth.py

@@ -0,0 +1,70 @@
+"""
+Construction and authentication contract.
+
+``ingest.dspace_be`` constructs the client from an endpoint/user/password and
+calls ``authenticate()``; a False return is turned into a hard ConnectionError,
+so the True/False semantics here matter.
+"""
+import unittest
+
+import requests_mock
+
+import _helpers  # noqa: F401
+from _helpers import make_client, API
+
+
+class TestConstructor(unittest.TestCase):
+
+    def test_endpoints_derived_from_api_endpoint(self):
+        c = make_client("http://host:8080/server/api")
+        self.assertEqual(c.API_ENDPOINT, "http://host:8080/server/api")
+        self.assertEqual(c.LOGIN_URL, "http://host:8080/server/api/authn/login")
+        self.assertIsNotNone(c.session)
+
+    def test_default_last_err_is_none(self):
+        self.assertIsNone(make_client().last_err)
+
+
+class TestAuthenticate(unittest.TestCase):
+
+    def test_success_returns_true_and_propagates_bearer_token(self):
+        c = make_client()
+        with requests_mock.Mocker() as m:
+            m.post(f"{API}/authn/login", status_code=200,
+                   headers={"Authorization": "Bearer tok123"})
+            m.get(f"{API}/authn/status", status_code=200,
+                  json={"authenticated": True})
+            self.assertTrue(c.authenticate())
+            # the bearer token must land on the session for later calls
+            self.assertEqual(c.session.headers.get("Authorization"), "Bearer tok123")
+
+    def test_invalid_credentials_401_returns_false(self):
+        c = make_client()
+        with requests_mock.Mocker() as m:
+            m.post(f"{API}/authn/login", status_code=401,
+                   json={"message": "invalid"})
+            self.assertFalse(c.authenticate())
+
+    def test_status_not_authenticated_returns_false(self):
+        c = make_client()
+        with requests_mock.Mocker() as m:
+            m.post(f"{API}/authn/login", status_code=200,
+                   headers={"Authorization": "Bearer t"})
+            m.get(f"{API}/authn/status", status_code=200,
+                  json={"authenticated": False})
+            self.assertFalse(c.authenticate())
+
+    def test_csrf_403_retries_once_then_gives_up(self):
+        c = make_client()
+        with requests_mock.Mocker() as m:
+            m.post(f"{API}/authn/login", status_code=403,
+                   json={"message": "CSRF token required"})
+            self.assertFalse(c.authenticate())
+            login_calls = [r for r in m.request_history
+                           if r.path == "/server/api/authn/login"]
+            # initial attempt + exactly one retry with the refreshed token
+            self.assertEqual(len(login_calls), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()