From c58b005bf658a67cddad51645a7a30c2215711f4 Mon Sep 17 00:00:00 2001 From: Tilo Mathes Date: Mon, 21 Sep 2026 14:25:46 +0200 Subject: [PATCH 1/3] Client layer: rebase advertised links, refuse redirects, add the calls the filesystems need URLs the server advertises in `_links` are rebased onto the configured host so paging and download links work behind a proxy or port-forward. The session never follows a redirect, because `requests` would carry the `apiKey` header to the new host. `doDelete` tolerates a leading slash. `upload_file` and `upload_attachment_by_global_id` take `filename=`, `list_folder_tree` takes `page_size=`, and the Inventory client gains `get_workbench_by_id` and `attach_gallery_file_by_global_id`. Inventory CSV import goes through the shared session. The existing filesystem tests now expect the redirect flag on every request. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 11 +++ rspace_client/client_base.py | 45 ++++++++++-- rspace_client/eln/eln.py | 12 +++- rspace_client/inv/inv.py | 61 +++++++++++++--- rspace_client/tests/eln_fs_test.py | 8 +-- rspace_client/tests/http_layer_test.py | 72 +++++++++++++++++++ rspace_client/tests/inv_attachment_fs_test.py | 6 +- 7 files changed, 189 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5eea76e..e87f580 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ All notable changes to this project will be documented in this file - replace naming of classes and methods using 'Workbench' to 'Bench' - replace create_sample, create_container long argument lists with new XXXPost objects +## 2.8.0 (unreleased) + +- Security: the client never follows an HTTP redirect, since `requests` would carry the + `apiKey` header to the new host. +- Client-level changes: `_links` URLs are rebased onto the configured host so paging and + download links work behind a proxy; `doDelete` tolerates a leading slash (fixes + `delete_document`); `upload_file` and `upload_attachment_by_global_id` take an optional + `filename=`; `list_folder_tree` takes `page_size=`; new `get_workbench_by_id`; Inventory + CSV import uses the shared session. +- Removed a stray debug print from `InventoryClient.upload_attachment_by_global_id`. + ## Unreleased - Gallery upload routing and section-mismatch handling (PR #56): clearer diff --git a/rspace_client/client_base.py b/rspace_client/client_base.py index 695f60c..0333ad9 100644 --- a/rspace_client/client_base.py +++ b/rspace_client/client_base.py @@ -1,6 +1,7 @@ import logging import re import requests +from urllib.parse import urlparse, urlunparse from typing import Optional @@ -21,6 +22,7 @@ DEFAULT_MAX_RETRIES = 3 DEFAULT_BACKOFF_FACTOR = 0.5 RETRY_STATUSES = (429, 500, 502, 503, 504) +REDIRECT_STATUSES = (301, 302, 303, 307, 308) class _RSpaceRetry(Retry): @@ -119,6 +121,22 @@ def __exit__(self, exc_type, exc_value, traceback): def _is_absolute_url(url): return url.startswith("http://") or url.startswith("https://") + def _rebase_url(self, url): + """ + Point an absolute URL at the server this client was configured with. + + RSpace builds the ``_links`` in its responses from its own configured base URL, which + is not necessarily the address the client used: behind a reverse proxy, in a container, + or through a port-forward the advertised host is unreachable. Following such a link + verbatim fails with a connection error, so keep the path and query and substitute our + own scheme and host. + """ + link = urlparse(url) + ours = urlparse(self.rspace_url) + if (link.scheme, link.netloc) == (ours.scheme, ours.netloc): + return url + return urlunparse((ours.scheme, ours.netloc, link.path, link.params, link.query, link.fragment)) + def _get_headers(self, content_type="application/json"): headers = {"apiKey": self.api_key} if content_type is not None: @@ -176,6 +194,17 @@ def _check_status(response): raise AuthenticationError( "Error code: 401, {}".format(ClientBase._get_error_detail(response)) ) + if response.status_code in REDIRECT_STATUSES: + # Redirects are never followed: requests keeps custom headers such as apiKey + # across a redirect to another host, which would hand the key to that host. + raise ApiError( + "Error code: {}, the server redirected to {!r}; rspace-client does not follow " + "redirects, so point the client at the final RSpace URL".format( + response.status_code, response.headers.get("Location") + ), + response_status_code=response.status_code, + response=response, + ) try: response.raise_for_status() @@ -205,7 +234,7 @@ def doDelete(self, path, resource_id): """ numeric_id = self._get_numeric_record_id(resource_id) return self.retrieve_api_results( - "/{}/{}".format(path, numeric_id), + "/{}/{}".format(path.strip("/"), numeric_id), # tolerate callers passing "/documents" content_type=None, request_type="DELETE", ) @@ -222,15 +251,13 @@ def retrieve_api_results( :param content_type: content type :return: parsed JSON response as a dictionary """ - url = endpoint - if not self._is_absolute_url(endpoint): - url = self._get_api_url() + endpoint + url = self._rebase_url(endpoint) if self._is_absolute_url(endpoint) else self._get_api_url() + endpoint headers = self._get_headers(content_type) try: if request_type == "GET": response = self._session.get( - url, params=params, headers=headers, timeout=self.timeout + url, params=params, headers=headers, timeout=self.timeout, allow_redirects=False ) elif ( request_type == "PUT" @@ -238,7 +265,8 @@ def retrieve_api_results( or request_type == "DELETE" ): response = self._session.request( - request_type, url, json=params, headers=headers, timeout=self.timeout + request_type, url, json=params, headers=headers, timeout=self.timeout, + allow_redirects=False ) else: raise ValueError( @@ -271,6 +299,7 @@ def _post_multipart(self, endpoint, files=None, data=None): data=data, headers=self._get_headers(), timeout=self.timeout, + allow_redirects=False, ) return self._handle_response(response) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: @@ -288,6 +317,7 @@ def _get_raw_response(self, url, params=None, accept="application/octet-stream") params=params, headers={"apiKey": self.api_key, "Accept": accept}, timeout=self.timeout, + allow_redirects=False, ) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: raise RSpaceConnectionError(e) @@ -340,10 +370,11 @@ def download_link_to_file(self, url, filename, chunk_size=8192): :param filename: file path to save the file to or an already opened file object :param chunk_size: size of the chunks to download at a time, default is 8192 """ + url = self._rebase_url(url) if self._is_absolute_url(url) else url headers = {"apiKey": self.api_key, "Accept": "application/octet-stream"} try: with self._session.get( - url, headers=headers, stream=True, timeout=self.timeout + url, headers=headers, stream=True, timeout=self.timeout, allow_redirects=False ) as response: self._check_status(response) if isinstance(filename, str): diff --git a/rspace_client/eln/eln.py b/rspace_client/eln/eln.py index d969158..b725678 100644 --- a/rspace_client/eln/eln.py +++ b/rspace_client/eln/eln.py @@ -347,13 +347,14 @@ def download_file(self, file_id, filename, chunk_size=128): f"{url_base}/files/{numeric_file_id}/file", filename, chunk_size ) - def upload_file(self, file, folder_id=None, caption=None): + def upload_file(self, file, folder_id=None, caption=None, filename=None): """ Upload a file to the gallery. More information on https://community.researchspace.com/public/apiDocs (or your own instance's /public/apiDocs). :param file: open file object :param folder_id: folder id of the destination folder :param caption: optional caption + :param filename: name to store the file under; defaults to the file object's own name :return: parsed response as a dictionary """ data = {} @@ -365,7 +366,8 @@ def upload_file(self, file, folder_id=None, caption=None): if caption is not None: data["caption"] = caption - return self._post_multipart("/files", files={"file": file}, data=data) + part = file if filename is None else (filename, file) + return self._post_multipart("/files", files={"file": part}, data=data) def update_file(self, file, fileId): """ @@ -793,12 +795,14 @@ def get_folder(self, folder_id): numeric_folder_id = self._get_numeric_record_id(folder_id) return self.retrieve_api_results("/folders/{}".format(numeric_folder_id)) - def list_folder_tree(self, folder_id=None, typesToInclude=[]): + def list_folder_tree(self, folder_id=None, typesToInclude=[], page_size=None): """ Lists contents of a folder by its ID. :param folder_id. Optional folderId. If none, will return listing of Home Folder :param typesToInclude: An optional list of any of 'folder', 'notebook' or 'document'. Results will be restricted to these types + :param page_size: optional number of records per page (the server default applies otherwise); + follow the response's 'next' link for further pages :return a paginated folder listing """ url = "" @@ -817,6 +821,8 @@ def list_folder_tree(self, folder_id=None, typesToInclude=[]): 'typesToInclude must be contain "document", "notebook" and/or "folder"' ) params["typesToInclude"] = ",".join(typesToInclude) + if page_size is not None: + params["pageSize"] = int(page_size) return self.retrieve_api_results(url, params) # Groups methods diff --git a/rspace_client/inv/inv.py b/rspace_client/inv/inv.py index 3106a31..f4b9352 100644 --- a/rspace_client/inv/inv.py +++ b/rspace_client/inv/inv.py @@ -1817,13 +1817,44 @@ def download_attachment_by_id(self, attachment_id: Union[str, int], file_path: U f"{url_base}/files/{attachment_id}/file", file_path, chunk_size ) - def upload_attachment_by_global_id(self, record_global_id: str, file: BinaryIO) -> dict: + def upload_attachment_by_global_id(self, record_global_id: str, file: BinaryIO, + filename: Optional[str] = None) -> dict: + """Attach a file to a record or an attachment field. ``filename`` overrides the name + taken from the file object.""" return self._post_multipart( "/files", - files={"file": file}, + files={"file": file if filename is None else (filename, file)}, data={"fileSettings": json.dumps({"parentGlobalId": record_global_id})}, ) + def attach_gallery_file_by_global_id(self, parent_global_id: str, media_file_global_id: str) -> dict: + """ + Attach a file that is already in the Gallery to an Inventory record or attachment field. + + This is the "link from Gallery" the web interface offers alongside "upload new". No + bytes move: the resulting Inventory file references the Gallery file, and its + ``mediaFileGlobalId`` names it. Deleting the attachment leaves the Gallery file alone. + + Parameters + ---------- + parent_global_id : str + A sample (SA...), subsample (SS...), container (IC...), instrument (IN...) or an + attachment-type field (SF...). Benches and sample templates are refused by the + server. + media_file_global_id : str + The Gallery file to link (GL...). + + Returns + ------- + Dict of the created InventoryFile, whose ``mediaFileGlobalId`` is the linked file. + """ + return self.retrieve_api_results( + "/attachments", + request_type="POST", + params={"parentGlobalId": Id(parent_global_id).as_global_id(), + "mediaFileGlobalId": Id(media_file_global_id).as_global_id()}, + ) + def split_subsample( self, subsample: Union[int, str, dict], @@ -1981,6 +2012,22 @@ def get_workbenches(self) -> Sequence[dict]: result = self.retrieve_api_results("/workbenches") return [wb for wb in result["containers"]] + def get_workbench_by_id(self, workbench_id: Union[str, int], include_content: bool = False) -> dict: + """ + Gets a workbench (bench) by id. Benches are containers but must be read through + /workbenches: /containers/{id} answers 422 for a bench. + + Parameters + ---------- + workbench_id : Union[str, int] + Numeric id or global id (BE...) of the bench. + include_content : bool + Include the bench's locations and their content (subsamples, containers). + """ + numeric_id = self._get_numeric_record_id(workbench_id) + return self.retrieve_api_results( + f"/workbenches/{numeric_id}?includeContent={'true' if include_content else 'false'}") + ## --------------------------------------------------------------------- ## CSV import ## --------------------------------------------------------------------- @@ -1994,13 +2041,9 @@ def _multipart_post(self, endpoint: str, files: dict, data: dict) -> dict: multipart Content-Type (with boundary) when ``files`` is supplied, and adds each entry of ``data`` as an additional form field. """ - response = requests.post( - self._get_api_url() + endpoint, - files=files, - data=data, - headers=self._get_headers(), - ) - return self._handle_response(response) + # through the shared session, so timeout, retries, redirect refusal and error + # handling are the same as for every other request + return self._post_multipart(endpoint, files=files, data=data) def parse_csv_import_file( self, file: BinaryIO, record_type: Union[str, ImportRecordType] diff --git a/rspace_client/tests/eln_fs_test.py b/rspace_client/tests/eln_fs_test.py index 1e290fc..c1d2184 100644 --- a/rspace_client/tests/eln_fs_test.py +++ b/rspace_client/tests/eln_fs_test.py @@ -193,7 +193,7 @@ def test_makedir(self, mock_get, mock_post): 'https://example.com/api/v1/folders', json={'name': 'newFolder', 'parentFolderId': 123, 'notebook': False}, headers=ANY, - timeout=ANY + timeout=ANY, allow_redirects=False ) @patch('requests.Session.request', side_effect=mock_requests_post) @@ -205,7 +205,7 @@ def test_removedir(self, mock_get, mock_post): 'https://example.com/api/v1/folders/456', json=ANY, headers=ANY, - timeout=ANY + timeout=ANY, allow_redirects=False ) @patch('requests.Session.get') @@ -223,7 +223,7 @@ def test_download(self, mock_get): 'https://example.com/api/v1/files/123/file', headers=ANY, stream=True, - timeout=ANY + timeout=ANY, allow_redirects=False ) @patch('requests.Session.post') @@ -242,7 +242,7 @@ def test_upload(self, mock_post): files={'file': file_obj}, data={'folderId': 123}, headers=ANY, - timeout=ANY + timeout=ANY, allow_redirects=False ) def test_classify_media_section(self): diff --git a/rspace_client/tests/http_layer_test.py b/rspace_client/tests/http_layer_test.py index cc2f53e..ffaf011 100644 --- a/rspace_client/tests/http_layer_test.py +++ b/rspace_client/tests/http_layer_test.py @@ -308,3 +308,75 @@ def test_serr_logs_warning_instead_of_printing(self): if __name__ == "__main__": unittest.main() + + +class RebaseAdvertisedLinksTest(unittest.TestCase): + """ + RSpace builds _links from its own configured base URL, which is often not the address the + client used (reverse proxy, container, port-forward). Following such a link verbatim fails, + so absolute URLs are pointed back at the configured server. + """ + + def client(self, url="https://rspace.example.com"): + return ELNClient(url, "key") + + def test_foreign_host_is_rebased_keeping_path_and_query(self): + c = self.client() + self.assertEqual( + "https://rspace.example.com/api/v1/folders/tree/215?pageSize=5&pageNumber=1", + c._rebase_url("http://localhost:8080/api/v1/folders/tree/215?pageSize=5&pageNumber=1")) + + def test_matching_host_is_untouched(self): + c = self.client() + url = "https://rspace.example.com/api/v1/files/7/file" + self.assertEqual(url, c._rebase_url(url)) + + def test_scheme_and_port_differences_are_rebased(self): + c = self.client("http://192.168.1.5:8080") + self.assertEqual("http://192.168.1.5:8080/api/v1/status", + c._rebase_url("https://rspace.internal/api/v1/status")) + + def test_following_a_link_requests_our_host(self): + c = self.client() + with mock.patch.object(c, "_session") as session: + session.get.return_value = mock.MagicMock( + status_code=200, headers={"Content-Type": "application/json"}, + json=mock.MagicMock(return_value={})) + c.retrieve_api_results("http://elsewhere.invalid:9999/api/v1/folders/tree/1?pageNumber=2") + called = session.get.call_args[0][0] + self.assertEqual("https://rspace.example.com/api/v1/folders/tree/1?pageNumber=2", called) + + +class RedirectRefusalTest(unittest.TestCase): + """The apiKey header must never follow a redirect to another host.""" + + def setUp(self): + self.client = ELNClient(RSPACE_URL, "fake-api-key") + + def tearDown(self): + self.client.close() + + @responses.activate + def test_a_redirect_is_an_api_error_and_is_not_followed(self): + responses.add(responses.GET, API_URL + "/status", status=302, + headers={"Location": "https://elsewhere.example/api/v1/status"}) + responses.add(responses.GET, "https://elsewhere.example/api/v1/status", json={}, status=200) + with self.assertRaises(ApiError) as cm: + self.client.retrieve_api_results("/status") + self.assertEqual(302, cm.exception.response_status_code) + self.assertIn("elsewhere.example", str(cm.exception)) + self.assertEqual(1, len(responses.calls)) # the second host was never contacted + + def test_every_session_call_disables_redirects(self): + ok = make_response(200, json_body={}) + with mock.patch.object(self.client._session, "get", return_value=ok) as g, \ + mock.patch.object(self.client._session, "request", return_value=ok) as r, \ + mock.patch.object(self.client._session, "post", return_value=ok) as p: + self.client.retrieve_api_results("/status") + self.client.retrieve_api_results("/forms", request_type="POST") + self.client._post_multipart("/files", files={"file": ("f.txt", b"x")}) + self.client._get_raw_response(API_URL + "/files/1/file") + for mocked in (g, r, p): + for call in mocked.call_args_list: + self.assertFalse(call.kwargs["allow_redirects"]) + diff --git a/rspace_client/tests/inv_attachment_fs_test.py b/rspace_client/tests/inv_attachment_fs_test.py index 0b22707..df7c62a 100644 --- a/rspace_client/tests/inv_attachment_fs_test.py +++ b/rspace_client/tests/inv_attachment_fs_test.py @@ -72,7 +72,7 @@ def test_get_info_folder(self, mock_get): @patch('requests.Session.request', side_effect=mock_requests) def test_remove(self, mock_request): self.fs.remove("IF123") - mock_request.assert_called_with('DELETE', 'https://example.com/api/inventory/v1/files/123', json=None, headers=ANY, timeout=ANY) + mock_request.assert_called_with('DELETE', 'https://example.com/api/inventory/v1/files/123', json=None, headers=ANY, timeout=ANY, allow_redirects=False) @patch('requests.Session.get') def test_download(self, mock_get): @@ -89,7 +89,7 @@ def test_download(self, mock_get): 'https://example.com/api/inventory/v1/files/123/file', headers=ANY, stream=True, - timeout=ANY + timeout=ANY, allow_redirects=False ) @patch('requests.Session.post') @@ -104,7 +104,7 @@ def test_upload(self, mock_post): data={'fileSettings': '{"parentGlobalId": "SS123"}'}, files={'file': file_obj}, headers=ANY, - timeout=ANY + timeout=ANY, allow_redirects=False ) @patch('requests.Session.get', side_effect=mock_requests_get) From 40ae8efbb6c0caff0434fe9586d2a1bf7a3de53c Mon Sep 17 00:00:00 2001 From: Tilo Mathes Date: Mon, 12 Oct 2026 08:17:39 +0200 Subject: [PATCH 2/3] Rebase advertised links onto the configured base path, not just the host _rebase_url swapped scheme and host but kept the server's path, so a client configured with a path prefix (https://proxy/rspace) followed next-page and download links to https://proxy/api/v1/... and lost /rspace. Re-anchor the part from /api/ on under our own base path, the same base relative requests use. Links without /api/ keep their path. Raised in review of PR #60. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 ++-- rspace_client/client_base.py | 10 +++++--- rspace_client/tests/http_layer_test.py | 32 ++++++++++++++++++++++++++ 3 files changed, 41 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e87f580..f17a148 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,8 +10,8 @@ All notable changes to this project will be documented in this file - Security: the client never follows an HTTP redirect, since `requests` would carry the `apiKey` header to the new host. -- Client-level changes: `_links` URLs are rebased onto the configured host so paging and - download links work behind a proxy; `doDelete` tolerates a leading slash (fixes +- Client-level changes: `_links` URLs are rebased onto the configured base URL, path prefix + included, so paging and download links work behind a proxy or port-forward; `doDelete` tolerates a leading slash (fixes `delete_document`); `upload_file` and `upload_attachment_by_global_id` take an optional `filename=`; `list_folder_tree` takes `page_size=`; new `get_workbench_by_id`; Inventory CSV import uses the shared session. diff --git a/rspace_client/client_base.py b/rspace_client/client_base.py index 0333ad9..45e4e03 100644 --- a/rspace_client/client_base.py +++ b/rspace_client/client_base.py @@ -130,12 +130,16 @@ def _rebase_url(self, url): or through a port-forward the advertised host is unreachable. Following such a link verbatim fails with a connection error, so keep the path and query and substitute our own scheme and host. + + Either side may also sit under a path prefix (a proxy serving RSpace at ``/rspace``), so + the part of the path from ``/api/`` on is re-anchored under our own base path, the same + base every relative request is built from. A link without ``/api/`` keeps its path. """ link = urlparse(url) ours = urlparse(self.rspace_url) - if (link.scheme, link.netloc) == (ours.scheme, ours.netloc): - return url - return urlunparse((ours.scheme, ours.netloc, link.path, link.params, link.query, link.fragment)) + api_at = link.path.find("/api/") + path = link.path if api_at < 0 else ours.path.rstrip("/") + link.path[api_at:] + return urlunparse((ours.scheme, ours.netloc, path, link.params, link.query, link.fragment)) def _get_headers(self, content_type="application/json"): headers = {"apiKey": self.api_key} diff --git a/rspace_client/tests/http_layer_test.py b/rspace_client/tests/http_layer_test.py index ffaf011..397d63a 100644 --- a/rspace_client/tests/http_layer_test.py +++ b/rspace_client/tests/http_layer_test.py @@ -346,6 +346,38 @@ def test_following_a_link_requests_our_host(self): called = session.get.call_args[0][0] self.assertEqual("https://rspace.example.com/api/v1/folders/tree/1?pageNumber=2", called) + def test_our_path_prefix_is_kept(self): + # client reaches RSpace through a proxy at /rspace; the server does not know about it + c = self.client("https://proxy.example.com/rspace/") + self.assertEqual( + "https://proxy.example.com/rspace/api/v1/folders/tree/1?pageNumber=2", + c._rebase_url("http://localhost:8080/api/v1/folders/tree/1?pageNumber=2")) + + def test_server_path_prefix_is_replaced_by_ours(self): + c = self.client() + self.assertEqual("https://rspace.example.com/api/inventory/v1/files/3/file", + c._rebase_url("http://internal:8080/rspace/api/inventory/v1/files/3/file")) + + def test_prefixes_on_both_sides(self): + c = self.client("https://proxy.example.com/eln") + self.assertEqual("https://proxy.example.com/eln/api/v1/status", + c._rebase_url("http://internal/rspace/api/v1/status")) + + def test_same_host_with_unprefixed_link_gains_our_prefix(self): + c = self.client("https://proxy.example.com/rspace") + self.assertEqual("https://proxy.example.com/rspace/api/v1/status", + c._rebase_url("https://proxy.example.com/api/v1/status")) + + def test_link_outside_the_api_keeps_its_path(self): + c = self.client("https://proxy.example.com/rspace") + self.assertEqual("https://proxy.example.com/public/apiDocs", + c._rebase_url("http://localhost:8080/public/apiDocs")) + + def test_relative_requests_and_followed_links_agree(self): + c = self.client("https://proxy.example.com/rspace") + self.assertEqual(c._get_api_url() + "/documents", + c._rebase_url("http://localhost:8080/api/v1/documents")) + class RedirectRefusalTest(unittest.TestCase): """The apiKey header must never follow a redirect to another host.""" From 822373956e91e61d7be4cb6ec6510c6554fec75b Mon Sep 17 00:00:00 2001 From: Tilo Mathes Date: Mon, 12 Oct 2026 08:17:39 +0200 Subject: [PATCH 3/3] Unit tests for the client methods the filesystems added or changed Covers the review's test gaps: doDelete single-slash URLs (regression for delete_document), upload_file and upload_attachment_by_global_id filename=, list_folder_tree page_size=, get_workbench_by_id id forms and includeContent, attach_gallery_file_by_global_id body, and CSV parse through the shared session with redirects off and the client timeout. Co-Authored-By: Claude Opus 5.5 --- rspace_client/tests/client_methods_test.py | 128 +++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 rspace_client/tests/client_methods_test.py diff --git a/rspace_client/tests/client_methods_test.py b/rspace_client/tests/client_methods_test.py new file mode 100644 index 0000000..ced8fce --- /dev/null +++ b/rspace_client/tests/client_methods_test.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Unit tests for the ELN and Inventory client methods added or changed for the filesystems: +the URLs, query parameters, bodies and multipart parts they send. No server needed; requests +are intercepted by responses or by mocking the shared session. +""" +import json +import unittest +from io import BytesIO +from unittest import mock + +import responses + +from rspace_client.eln.eln import ELNClient +from rspace_client.inv.inv import InventoryClient +from rspace_client.tests.exceptions_test import make_response + +RSPACE_URL = "https://example.com" +ELN_API = RSPACE_URL + "/api/v1" +INV_API = RSPACE_URL + "/api/inventory/v1" + + +def named_file(name, content=b"content"): + f = BytesIO(content) + f.name = name + return f + + +def sent_body(call): + body = call.request.body + return body if isinstance(body, bytes) else body.encode() + + +class ELNClientMethodsTest(unittest.TestCase): + def setUp(self): + self.client = ELNClient(RSPACE_URL, "fake-api-key") + + def tearDown(self): + self.client.close() + + @responses.activate + def test_delete_document_requests_a_single_slash_url(self): + # regression: doDelete used to build //documents/5 + responses.add(responses.DELETE, ELN_API + "/documents/5", status=204) + self.client.delete_document(5) + self.assertEqual(ELN_API + "/documents/5", responses.calls[0].request.url) + + @responses.activate + def test_doDelete_tolerates_a_leading_slash_on_the_path(self): + responses.add(responses.DELETE, ELN_API + "/forms/3", status=204) + self.client.doDelete("/forms", 3) + self.client.doDelete("forms", "FM3") + self.assertEqual([ELN_API + "/forms/3"] * 2, [c.request.url for c in responses.calls]) + + @responses.activate + def test_upload_file_uses_the_given_filename(self): + responses.add(responses.POST, ELN_API + "/files", json={"id": 1}) + self.client.upload_file(named_file("/tmp/original.bin"), filename="x.txt") + body = sent_body(responses.calls[0]) + self.assertIn(b'filename="x.txt"', body) + self.assertNotIn(b"original.bin", body) + + @responses.activate + def test_upload_file_defaults_to_the_file_objects_name(self): + responses.add(responses.POST, ELN_API + "/files", json={"id": 1}) + self.client.upload_file(named_file("/tmp/original.bin")) + self.assertIn(b'filename="original.bin"', sent_body(responses.calls[0])) + + @responses.activate + def test_list_folder_tree_sends_page_size_only_when_given(self): + responses.add(responses.GET, ELN_API + "/folders/tree/7", json={}) + self.client.list_folder_tree(7, page_size=5) + self.client.list_folder_tree(7) + with_size, without = (c.request.url for c in responses.calls) + self.assertIn("pageSize=5", with_size) + self.assertNotIn("pageSize", without) + + +class InventoryClientMethodsTest(unittest.TestCase): + def setUp(self): + self.client = InventoryClient(RSPACE_URL, "fake-api-key") + + def tearDown(self): + self.client.close() + + @responses.activate + def test_upload_attachment_uses_the_given_filename_and_names_the_parent(self): + responses.add(responses.POST, INV_API + "/files", json={"id": 1}) + self.client.upload_attachment_by_global_id("SA1", named_file("/tmp/original.bin"), + filename="x.txt") + body = sent_body(responses.calls[0]) + self.assertIn(b'filename="x.txt"', body) + self.assertIn(json.dumps({"parentGlobalId": "SA1"}).encode(), body) + + @responses.activate + def test_get_workbench_by_id_accepts_global_and_numeric_ids(self): + url = INV_API + "/workbenches/64" + responses.add(responses.GET, url, json={}) + self.client.get_workbench_by_id("BE64") + self.client.get_workbench_by_id(64) + self.client.get_workbench_by_id(64, include_content=True) + self.assertEqual( + [url + "?includeContent=false", url + "?includeContent=false", + url + "?includeContent=true"], + [c.request.url for c in responses.calls]) + + @responses.activate + def test_attach_gallery_file_posts_both_global_ids(self): + responses.add(responses.POST, INV_API + "/attachments", json={"id": 1}) + self.client.attach_gallery_file_by_global_id("SA1", "GL2") + self.assertEqual({"parentGlobalId": "SA1", "mediaFileGlobalId": "GL2"}, + json.loads(responses.calls[0].request.body)) + + def test_csv_parse_goes_through_the_shared_session(self): + ok = make_response(200, json_body={}) + with mock.patch.object(self.client._session, "post", return_value=ok) as post, \ + mock.patch("requests.post") as bare_post: + self.client.parse_csv_import_file(named_file("samples.csv"), "SAMPLES") + post.assert_called_once() + self.assertEqual(INV_API + "/import/parseFile", post.call_args.args[0]) + self.assertFalse(post.call_args.kwargs["allow_redirects"]) + self.assertEqual(self.client.timeout, post.call_args.kwargs["timeout"]) + bare_post.assert_not_called() + + +if __name__ == "__main__": + unittest.main()