From 631051affc29b666e7fed8c07085536ee20bdbf0 Mon Sep 17 00:00:00 2001 From: Tilo Mathes Date: Mon, 21 Sep 2026 14:26:04 +0200 Subject: [PATCH] Add an offline mock of the RSpace ELN and Inventory APIs for the test suite A standard-library HTTP server covering the endpoints the PyFilesystem classes use, backed by an in-memory dataset with the awkward cases a filesystem has to handle (duplicate names, system folders, paginated sections, documents with mixed field types). It counts requests so tests can assert what an operation costs. One server runs for the whole pytest session and `MockServerTestCase` resets its fixtures before each test. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 2 + rspace_client/tests/conftest.py | 25 + rspace_client/tests/mock_rspace/__init__.py | 6 + rspace_client/tests/mock_rspace/fixtures.py | 690 ++++++++++++++++++++ rspace_client/tests/mock_rspace/server.py | 495 ++++++++++++++ rspace_client/tests/mock_server_case.py | 61 ++ 6 files changed, 1279 insertions(+) create mode 100644 rspace_client/tests/conftest.py create mode 100644 rspace_client/tests/mock_rspace/__init__.py create mode 100644 rspace_client/tests/mock_rspace/fixtures.py create mode 100644 rspace_client/tests/mock_rspace/server.py create mode 100644 rspace_client/tests/mock_server_case.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f17a148..81fa09a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ All notable changes to this project will be documented in this file `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`. +- Added `rspace_client/tests/mock_rspace`, an offline mock of the ELN and Inventory REST + endpoints the PyFilesystem classes use, shared by the unit suite. Not installed. ## Unreleased diff --git a/rspace_client/tests/conftest.py b/rspace_client/tests/conftest.py new file mode 100644 index 0000000..32ba096 --- /dev/null +++ b/rspace_client/tests/conftest.py @@ -0,0 +1,25 @@ +""" +pytest hooks for the offline test suite. + +One mock RSpace server (``rspace_client/tests/mock_rspace``) serves the whole +session. It is started here and handed to :mod:`rspace_client.tests.mock_server_case`, +whose ``MockServerTestCase`` resets its fixtures before every test; the old +one-server-per-class arrangement spent most of the suite's wall time in seven +``serve_forever`` shutdowns. +""" +import pytest + +from rspace_client.tests import mock_server_case +from rspace_client.tests.mock_rspace.server import run_in_thread + + +@pytest.fixture(scope="session", autouse=True) +def shared_mock_server(): + server, url = run_in_thread() + mock_server_case.install_shared_server(server, url) + try: + yield server + finally: + mock_server_case.install_shared_server(None, None) + server.shutdown() + server.server_close() diff --git a/rspace_client/tests/mock_rspace/__init__.py b/rspace_client/tests/mock_rspace/__init__.py new file mode 100644 index 0000000..6237c64 --- /dev/null +++ b/rspace_client/tests/mock_rspace/__init__.py @@ -0,0 +1,6 @@ +""" +Mock RSpace API server and a Galaxy-like harness for developing and testing the +RSpace PyFilesystem implementations offline. + +See rspace_client/tests/mock_rspace/README.md for usage. +""" diff --git a/rspace_client/tests/mock_rspace/fixtures.py b/rspace_client/tests/mock_rspace/fixtures.py new file mode 100644 index 0000000..208df60 --- /dev/null +++ b/rspace_client/tests/mock_rspace/fixtures.py @@ -0,0 +1,690 @@ +""" +In-memory fixture dataset for the mock RSpace server, plus serialisers that +render records in the same JSON shapes the real RSpace ELN and Inventory APIs +return (field names taken from the server DTOs and from the recorded fixtures in +rspace_client/tests/data). + +The dataset is deliberately small but realistic: + +ELN (Workspace + Gallery) + FL1 "user1a" home folder + GF2 "Gallery" system folder (the Gallery root; GF id, as on real servers) + GF10 "Images" GL100 microscope.png, GL101 gel.jpg, GF13 "2026-09 run" (GL102 plate1.png) + GF11 "Documents" GL110 protocol.docx, GL111 data.csv, GL112 data.csv (duplicate name on purpose) + GF12 "Chemistry" (empty) + GF14 "Api Imports" default upload target + FL3 "Shared", FL4 "Templates", FL5 "Api Inbox", FL6 "Imports" system folders + FL20 "Project Alpha" + NB30 "Lab notebook 2026" SD40 "2026-09-15 entry", SD41 "2026-09-16 entry" + SD42 "Alpha protocol" custom form with TEXT/ATTACHMENT/NUMBER/DATE/STRING/CHOICE fields + (only the two TEXT fields are browsable) + FL21 "Project Beta" + SD43 "Beta results" + +Inventory + BE1 "WB user1a" workbench: IC203 "Bench box" (SS303), SS304 + IC200 "Freezer -80" top level, attachment IF500; contains IC201 "Rack A" (grid: SS300, SS301), IC202 "Rack B" (SS302) + IC204..IC215 "Shelf 1..12" top level, one subsample each (SS305..SS316) -> makes /containers and /samples paginate + SA1000 "Plasmid pUC19" template IT1, attachment IF501, subsamples SS300 SS301 (SS300 has IF502) + SA1001 "E. coli DH5a" template IT2, subsamples SS302 SS303 + SA1002 "Buffer stock" subsample SS304 + SA1003..SA1014 "Sample N" one subsample each (15 samples in total) + IT1 "Plasmid" (IF503), IT2 "Bacterial strain" +""" +from __future__ import annotations + +import itertools +import mimetypes +import re +import threading +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional + +_T0 = datetime(2026, 9, 1, 9, 0, 0, tzinfo=timezone.utc) +FILE_ID_TOKEN = re.compile(r"") +ATTACH_REF = re.compile(r'attachOnText_(\d+)"') + + +IMAGE_REF = re.compile(r"sourceType=IMAGE&sourceId=(\d+)") + + +def render_attachment(file: dict, field_id: int = 0) -> str: + """The HTML a real RSpace server stores in place of a token (text fields).""" + if str(file.get("contentType", "")).startswith("image/"): + return ( + f'\n

image {file[

' + ) + return ( + '\n
\n' + f' \n' + f'

{file["name"]}

\n' + f'
\n' + ' \n
\n
' + ) + +OWNER = { + "id": 1, + "username": "user1a", + "firstName": "User", + "lastName": "One", + "email": "user1a@example.com", +} + +def iso(offset_minutes: int = 0) -> str: + """ISO-8601 timestamp in the format RSpace uses, offset from a fixed epoch.""" + return (_T0 + timedelta(minutes=offset_minutes)).strftime("%Y-%m-%dT%H:%M:%S.000Z") + + +def fake_blob(name: str, text: Optional[str] = None) -> bytes: + """Deterministic file bytes; images get a real PNG/JPEG signature so mime sniffing works.""" + lower = name.lower() + if text is not None: + return text.encode("utf-8") + if lower.endswith(".png"): + return b"\x89PNG\r\n\x1a\n" + (b"mock-png-payload " * 64) + if lower.endswith((".jpg", ".jpeg")): + return b"\xff\xd8\xff\xe0" + (b"mock-jpeg-payload " * 48) + if lower.endswith(".pdf"): + return b"%PDF-1.4\n" + (b"% mock pdf\n" * 40) + b"%%EOF\n" + return (f"mock content of {name}\n" * 20).encode("utf-8") + + +def guess_type(name: str, fallback: str = "application/octet-stream") -> str: + return mimetypes.guess_type(name)[0] or fallback + + +def ext_of(name: str) -> str: + return name.rsplit(".", 1)[-1].lower() if "." in name else "" + + +class MockData: + """All mock state. Mutations are guarded by `lock` in the server.""" + + def __init__(self) -> None: + self.lock = threading.Lock() + self.reset() + + # ------------------------------------------------------------------ build + + def reset(self) -> None: + self._ids = itertools.count(10000) + # ELN + self.folders: Dict[int, dict] = {} # FL / GF / NB + self.gallery_files: Dict[int, dict] = {} # GL + self.documents: Dict[int, dict] = {} # SD + self.children: Dict[int, List[str]] = {} # folder id -> [globalId, ...] + self.forms = {1: "Basic Document", 2: "Experiment"} + # Inventory + self.containers: Dict[int, dict] = {} # IC and BE (benches) + self.samples: Dict[int, dict] = {} # SA + self.subsamples: Dict[int, dict] = {} # SS + self.templates: Dict[int, dict] = {} # IT + self.instruments: Dict[int, dict] = {} # IN + self.fields: Dict[int, dict] = {} # SF (template-defined record fields) + self.inv_files: Dict[int, dict] = {} # IF + # bytes for GL and IF, keyed by globalId + self.blobs: Dict[str, bytes] = {} + self._build() + + def new_id(self) -> int: + return next(self._ids) + + # ELN builders ----------------------------------------------------------- + + def _folder(self, fid: int, name: str, parent: Optional[int], prefix: str = "FL", + notebook: bool = False, media_type: Optional[str] = None, + system: bool = False, t: int = 0) -> dict: + f = { + "id": fid, "globalId": f"{prefix}{fid}", "name": name, "parentFolderId": parent, + "notebook": notebook, "mediaType": media_type, "system": system, + "created": iso(t), "lastModified": iso(t), + } + self.folders[fid] = f + self.children.setdefault(fid, []) + if parent is not None: + self.children.setdefault(parent, []).append(f["globalId"]) + return f + + def _gfile(self, gid: int, name: str, parent: int, t: int = 0, text: Optional[str] = None, + caption: str = "") -> dict: + blob = fake_blob(name, text) + g = { + "id": gid, "globalId": f"GL{gid}", "name": name, "caption": caption, + "contentType": guess_type(name), "created": iso(t), "size": len(blob), + "version": 1, "parentFolderId": parent, + } + self.gallery_files[gid] = g + self.blobs[g["globalId"]] = blob + self.children.setdefault(parent, []).append(g["globalId"]) + return g + + def _doc(self, did: int, name: str, parent: int, form_id: int, fields: List[dict], + t: int = 0, signed: bool = False) -> dict: + for i, fld in enumerate(fields): + fld.setdefault("columnIndex", i + 1) + fld.setdefault("attachment_file_ids", []) + fld["globalId"] = f"FD{fld['id']}" + fld["lastModified"] = iso(t) + d = { + "id": did, "globalId": f"SD{did}", "name": name, "parentFolderId": parent, + "formId": form_id, "tags": "", "created": iso(t), "lastModified": iso(t), + "fields": fields, "signed": signed, + } + self.documents[did] = d + self.children.setdefault(parent, []).append(d["globalId"]) + return d + + # Inventory builders ------------------------------------------------------- + + def _container(self, cid: int, name: str, ctype: str = "LIST", parent: Optional[str] = None, + prefix: str = "IC", grid: Optional[tuple] = None, t: int = 0) -> dict: + c = { + "id": cid, "globalId": f"{prefix}{cid}", "name": name, "type": "CONTAINER", + "cType": ctype, "description": "", "created": iso(t), "lastModified": iso(t), + "parentGlobalId": parent, "gridLayout": ( + {"columnsNumber": grid[0], "rowsNumber": grid[1]} if grid else None), + "attachment_ids": [], "content": [], # content: list of child globalIds in location order + "canStoreContainers": ctype != "GRID", "canStoreSamples": True, + } + self.containers[cid] = c + if parent is not None: + self.containers[int(parent[2:])]["content"].append(c["globalId"]) + return c + + def _sample(self, sid: int, name: str, template: Optional[int] = None, t: int = 0) -> dict: + s = { + "id": sid, "globalId": f"SA{sid}", "name": name, "type": "SAMPLE", "description": "", + "created": iso(t), "lastModified": iso(t), "templateId": template, + "quantity": {"numericValue": 10.0, "unitId": 3}, "subsample_ids": [], + "attachment_ids": [], "tags": [], "fields": [], + } + self.samples[sid] = s + return s + + def _subsample(self, ssid: int, name: str, sample: int, parent: Optional[str], t: int = 0) -> dict: + ss = { + "id": ssid, "globalId": f"SS{ssid}", "name": name, "type": "SUBSAMPLE", + "created": iso(t), "lastModified": iso(t), "sampleId": sample, + "parentGlobalId": parent, "quantity": {"numericValue": 1.0, "unitId": 3}, + "attachment_ids": [], "tags": [], + } + self.subsamples[ssid] = ss + self.samples[sample]["subsample_ids"].append(ssid) + if parent is not None: + self.containers[int(parent[2:])]["content"].append(ss["globalId"]) + return ss + + def _template(self, tid: int, name: str, t: int = 0) -> dict: + tpl = { + "id": tid, "globalId": f"IT{tid}", "name": name, "type": "SAMPLE_TEMPLATE", + "description": "", "created": iso(t), "lastModified": iso(t), + "attachment_ids": [], "tags": [], "defaultUnitId": 3, "subSampleAlias": "subsample", + } + self.templates[tid] = tpl + return tpl + + def _sample_field(self, record: dict, fid: int, name: str, ftype: str) -> dict: + """A template-defined field (SF). Only an 'attachment' field holds a file, and one.""" + field = {"id": fid, "globalId": f"SF{fid}", "name": name, "type": ftype, + "content": None, "attachment_id": None} + record.setdefault("fields", []).append(field) + self.fields[fid] = field + return field + + def _instrument(self, iid: int, name: str, parent: Optional[str] = None, t: int = 0) -> dict: + ins = { + "id": iid, "globalId": f"IN{iid}", "name": name, "type": "INSTRUMENT", + "description": "", "created": iso(t), "lastModified": iso(t), + "parentGlobalId": parent, "attachment_ids": [], "tags": [], + } + self.instruments[iid] = ins + if parent is not None: + self.containers[int(parent[2:])]["content"].append(ins["globalId"]) + return ins + + def _ifile(self, fid: int, name: str, parent_gid: str, t: int = 0, text: Optional[str] = None) -> dict: + blob = fake_blob(name, text) + a = { + "id": fid, "globalId": f"IF{fid}", "name": name, "parentGlobalId": parent_gid, + "mediaFileGlobalId": None, "type": "GENERAL", "contentMimeType": guess_type(name), + "extension": ext_of(name), "size": len(blob), "created": iso(t), + "createdBy": OWNER["username"], "deleted": False, + } + self.inv_files[fid] = a + self.blobs[a["globalId"]] = blob + parent = self.record_by_global_id(parent_gid) + if parent_gid.startswith("SF"): # a field holds one file; a new one replaces it + previous = parent.get("attachment_id") + if previous is not None: + self.inv_files[previous]["deleted"] = True + parent["attachment_id"] = fid + else: + parent["attachment_ids"].append(fid) + return a + + def _build(self) -> None: + # ---- ELN tree + self._folder(1, "user1a", None) + self._folder(2, "Gallery", 1, prefix="GF", system=True) # real servers use a GF id for the Gallery root + self._folder(3, "Shared", 1, system=True) + self._folder(4, "Templates", 1, system=True) + self._folder(5, "Api Inbox", 1, system=True) + self._folder(6, "Imports", 1, system=True) + self._folder(10, "Images", 2, prefix="GF", media_type="Images", t=1) + self._folder(11, "Documents", 2, prefix="GF", media_type="Documents", t=2) + self._folder(12, "Chemistry", 2, prefix="GF", media_type="Chemistry", t=3) + self._folder(13, "2026-09 run", 10, prefix="GF", media_type="Images", t=4) + self._folder(14, "Api Imports", 2, prefix="GF", media_type="Documents", t=5) + self._folder(20, "Project Alpha", 1, t=10) + self._folder(21, "Project Beta", 1, t=11) + self._folder(30, "Lab notebook 2026", 20, prefix="NB", notebook=True, t=12) + + self._gfile(100, "microscope.png", 10, t=20, caption="Confocal, 40x") + self._gfile(101, "gel.jpg", 10, t=21) + self._gfile(102, "plate1.png", 13, t=22) + self._gfile(110, "protocol.docx", 11, t=23) + self._gfile(111, "data.csv", 11, t=24, text="well,od600\nA1,0.41\nA2,0.39\n") + self._gfile(112, "data.csv", 11, t=25, text="well,od600\nB1,0.52\nB2,0.48\n") + + self._doc(40, "2026-09-15 entry", 30, 1, [ + {"id": 400, "name": "Data", "type": "text", + "content": "

Imaging session, see image.

"}, + ], t=30) + self._doc(41, "2026-09-16 entry", 30, 1, [ + {"id": 410, "name": "Data", "type": "text", "content": "

No images yet.

"}, + ], t=31) + self._doc(42, "Alpha protocol", 20, 2, [ + {"id": 420, "name": "Objective", "type": "text", + "content": "

Express pUC19 in DH5a and check yield.

"}, + {"id": 421, "name": "Results", "type": "text", + "content": "

See attached data and plate image.

"}, + # A legacy Attachment field: the form editor cannot create one and nothing links + # media to it, so it is always empty and the filesystem does not list it. + {"id": 422, "name": "Raw data", "type": "attachment", "content": ""}, + {"id": 423, "name": "Replicates", "type": "number", "content": "3"}, + {"id": 424, "name": "Run date", "type": "date", "content": "2026-09-15"}, + {"id": 425, "name": "Operator", "type": "string", "content": "user1a"}, + {"id": 426, "name": "Status", "type": "choice", "content": "in progress"}, + ], t=32) + self._doc(44, "Signed protocol", 20, 1, [ + # holds one linked file, so that unlinking from a signed document can be tested + {"id": 440, "name": "Data", "type": "text", "content": "

witnessed

"}, + ], t=34, signed=True) + self._doc(43, "Beta results", 21, 1, [ + {"id": 430, "name": "Data", "type": "text", "content": "

Second run.

"}, + ], t=33) + + # ---- Inventory + self._container(1, "WB user1a", ctype="WORKBENCH", prefix="BE") + self._container(200, "Freezer -80", t=40) + self._container(201, "Rack A", ctype="GRID", parent="IC200", grid=(2, 2), t=41) + self._container(202, "Rack B", parent="IC200", t=42) + self._container(203, "Bench box", parent="BE1", t=43) + for i in range(12): + self._container(204 + i, f"Shelf {i + 1}", t=44 + i) + + self._template(1, "Plasmid", t=50) + self._template(2, "Bacterial strain", t=51) + + self._sample(1000, "Plasmid pUC19", template=1, t=60) + self._sample(1001, "E. coli DH5a", template=2, t=61) + self._sample(1002, "Buffer stock", t=62) + self._subsample(300, "Plasmid pUC19 #1", 1000, "IC201", t=63) + self._subsample(301, "Plasmid pUC19 #2", 1000, "IC201", t=64) + self._subsample(302, "E. coli DH5a #1", 1001, "IC202", t=65) + self._subsample(303, "E. coli DH5a #2", 1001, "IC203", t=66) + self._subsample(304, "Buffer stock #1", 1002, "BE1", t=67) + for i in range(12): # one extra sample per shelf IC204..IC215 + sid = 1003 + i + self._sample(sid, f"Sample {sid - 1000}", t=70 + i) + self._subsample(305 + i, f"Sample {sid - 1000} #1", sid, f"IC{204 + i}", t=70 + i) + + self._sample_field(self.samples[1000], 700, "Safety Data", "attachment") + self._sample_field(self.samples[1000], 701, "Concentration", "number") + self._sample_field(self.samples[1001], 702, "Datasheet", "attachment") # deliberately empty + self._instrument(2, "Confocal microscope", parent="BE1", t=52) + self._ifile(504, "calibration.pdf", "IN2", t=84) + self._ifile(505, "msds.pdf", "SF700", t=85) # a file held by an attachment field + self._ifile(500, "freezer_manual.pdf", "IC200", t=80) + self._ifile(501, "plasmid_map.gb", "SA1000", t=81, text="LOCUS pUC19 2686 bp DNA circular\n") + self._ifile(502, "qc_gel.png", "SS300", t=82) + self._ifile(503, "template_notes.txt", "IT1", t=83, text="Notes for the plasmid template.\n") + + # ------------------------------------------------------------------ lookups + + def record_by_global_id(self, gid: str) -> dict: + prefix, num = gid[:2], int(gid[2:]) + table = { + "FL": self.folders, "GF": self.folders, "NB": self.folders, "GL": self.gallery_files, + "SD": self.documents, "IC": self.containers, "BE": self.containers, + "SA": self.samples, "SS": self.subsamples, "IT": self.templates, + "IN": self.instruments, "IF": self.inv_files, "SF": self.fields, + }.get(prefix) + if table is None or num not in table: + raise KeyError(gid) + return table[num] + + def gallery_root_id(self) -> int: + return 2 + + def home_id(self) -> int: + return 1 + + def is_gallery_folder(self, fid: int) -> bool: + f = self.folders.get(fid) + return bool(f) and (f["globalId"].startswith("GF") or fid == self.gallery_root_id()) + + def top_level_containers(self) -> List[dict]: + return [c for c in self.containers.values() + if c["parentGlobalId"] is None and c["cType"] != "WORKBENCH"] + + def benches(self) -> List[dict]: + return [c for c in self.containers.values() if c["cType"] == "WORKBENCH"] + + # ------------------------------------------------------------------ mutations + + def add_folder(self, name: str, parent: Optional[int], notebook: bool) -> dict: + parent = parent if parent is not None else self.home_id() + if parent not in self.folders: + raise KeyError(f"FL{parent}") + prefix = "GF" if self.is_gallery_folder(parent) else ("NB" if notebook else "FL") + media = self.folders[parent].get("mediaType") if prefix == "GF" else None + return self._folder(self.new_id(), name, parent, prefix=prefix, notebook=notebook and prefix == "NB", + media_type=media, t=100) + + def remove_folder(self, fid: int) -> None: + f = self.folders[fid] + if self.children.get(fid): + raise ValueError("folder is not empty") + parent = f["parentFolderId"] + if parent is not None: + self.children[parent].remove(f["globalId"]) + del self.folders[fid] + del self.children[fid] + + def add_gallery_file(self, name: str, parent: Optional[int], content: bytes, + content_type: Optional[str], caption: str = "") -> dict: + parent = parent if parent is not None else 14 # "Api Imports" + if not self.is_gallery_folder(parent) or parent == self.gallery_root_id(): + raise ValueError(f"folder {parent} is not a Gallery folder that can hold files") + gid = self.new_id() + g = { + "id": gid, "globalId": f"GL{gid}", "name": name, "caption": caption, + "contentType": content_type or guess_type(name), "created": iso(200), "size": len(content), + "version": 1, "parentFolderId": parent, + } + self.gallery_files[gid] = g + self.blobs[g["globalId"]] = content + self.children.setdefault(parent, []).append(g["globalId"]) + return g + + def update_document(self, did: int, payload: dict) -> dict: + d = self.documents[did] + if payload.get("name"): + d["name"] = payload["name"] + if "tags" in payload and payload["tags"] is not None: + d["tags"] = payload["tags"] + by_id = {f["id"]: f for f in d["fields"]} + for upd in payload.get("fields") or []: + fld = by_id.get(upd.get("id")) + if fld is None: + raise KeyError(f"field {upd.get('id')} not in document SD{did}") + if "content" in upd: + content = upd["content"] + if fld["type"] == "text": # like the server: tokens become attachment markup on save + content = FILE_ID_TOKEN.sub( + lambda m: render_attachment(self.gallery_files[int(m.group(1))], fld["id"]) + if int(m.group(1)) in self.gallery_files else "", content) + fld["content"] = content + fld["lastModified"] = iso(201) + d["lastModified"] = iso(201) + return d + + def link_gallery_file(self, parent_gid: str, media_gid: str) -> dict: + """The server's "link from Gallery": no bytes move, and the resulting Inventory file + names the Gallery file it points at. Benches and sample templates are refused, as the + real server refuses them.""" + if parent_gid[:2] in ("BE",): + raise ValueError(f"Unsupported global id type: {parent_gid}") + if parent_gid[:2] in ("IT",): + raise ValueError("Sample Templates don't support file attachments yet") + gallery = next((f for f in self.gallery_files.values() + if f.get("globalId") == media_gid), None) + if gallery is None: + raise ValueError(f"no such Gallery file: {media_gid}") + linked = self.add_inv_file(parent_gid, gallery["name"], + self.blobs.get(media_gid, b""), gallery.get("contentType")) + linked["mediaFileGlobalId"] = media_gid + return linked + + def add_inv_file(self, parent_gid: str, name: str, content: bytes, content_type: Optional[str]) -> dict: + if parent_gid[:2] not in ("IC", "SA", "SS", "IT", "IN", "SF"): + raise ValueError(f"{parent_gid} cannot hold attachments") + parent = self.record_by_global_id(parent_gid) + fid = self.new_id() + a = { + "id": fid, "globalId": f"IF{fid}", "name": name, "parentGlobalId": parent_gid, + "mediaFileGlobalId": None, "type": "GENERAL", "contentMimeType": content_type or guess_type(name), + "extension": ext_of(name), "size": len(content), "created": iso(202), + "createdBy": OWNER["username"], "deleted": False, + } + self.inv_files[fid] = a + self.blobs[a["globalId"]] = content + if parent_gid.startswith("SF"): # a field holds one file; a new one replaces it + previous = parent.get("attachment_id") + if previous is not None: + self.inv_files[previous]["deleted"] = True + parent["attachment_id"] = fid + else: + parent["attachment_ids"].append(fid) + return a + + def remove_inv_file(self, fid: int) -> None: + a = self.inv_files.pop(fid) + parent = self.record_by_global_id(a["parentGlobalId"]) + if a["parentGlobalId"].startswith("SF"): + if parent.get("attachment_id") == fid: + parent["attachment_id"] = None + else: + parent["attachment_ids"].remove(fid) + self.blobs.pop(a["globalId"], None) + + +# ============================================================ API serialisers +# Each takes the record and the API base URL (e.g. http://host:port/api/v1) so +# that _links are absolute and share the client's origin. + +def _self(base: str, path: str) -> List[dict]: + return [{"link": f"{base}{path}", "rel": "self"}] + + +def ser_folder(f: dict, base: str) -> dict: + return { + "id": f["id"], "globalId": f["globalId"], "name": f["name"], "created": f["created"], + "lastModified": f["lastModified"], "parentFolderId": f["parentFolderId"], + "notebook": f["notebook"], "mediaType": f["mediaType"], "pathToRootFolder": None, + "_links": _self(base, f"/folders/{f['id']}"), + } + + +def ser_gallery_file(g: dict, base: str) -> dict: + return { + "id": g["id"], "globalId": g["globalId"], "name": g["name"], "caption": g["caption"], + "contentType": g["contentType"], "created": g["created"], "size": g["size"], + "version": g["version"], "parentFolderId": g["parentFolderId"], + "_links": _self(base, f"/files/{g['id']}") + [ + {"link": f"{base}/files/{g['id']}/file", "rel": "enclosure"}], + } + + +def ser_tree_item(gid: str, data: MockData, base: str) -> dict: + rec = data.record_by_global_id(gid) + prefix = gid[:2] + if prefix in ("FL", "GF"): + typ, link = "FOLDER", f"/folders/{rec['id']}" + elif prefix == "NB": + typ, link = "NOTEBOOK", f"/folders/{rec['id']}" + elif prefix == "SD": + typ, link = "DOCUMENT", f"/documents/{rec['id']}" + else: + typ, link = "MEDIA_FILE", f"/files/{rec['id']}" + item = { + "id": rec["id"], "globalId": gid, "name": rec["name"], "created": rec["created"], + "lastModified": rec.get("lastModified", rec["created"]), "parentFolderId": rec.get("parentFolderId"), + "type": typ, "owner": OWNER, "_links": _self(base, link), + } + if typ == "FOLDER": # real servers flag system and shared folders on folder tree items + item["systemFolder"] = bool(rec.get("system")) + item["sharedFolder"] = rec.get("name") == "Shared" + return item + + +def field_file_ids(fld: dict) -> List[int]: + if fld["type"] == "text": + content = fld.get("content") or "" + ids = ([int(m) for m in FILE_ID_TOKEN.findall(content)] + [int(m) for m in ATTACH_REF.findall(content)] + + [int(m) for m in IMAGE_REF.findall(content)]) + return list(dict.fromkeys(ids)) # unique, in order + if fld["type"] == "attachment": + return list(fld.get("attachment_file_ids") or []) + return [] + + +def ser_document(d: dict, data: MockData, base: str) -> dict: + fields = [] + for fld in d["fields"]: + files = [ser_gallery_file(data.gallery_files[i], base) + for i in field_file_ids(fld) if i in data.gallery_files] + fields.append({ + "id": fld["id"], "globalId": fld["globalId"], "name": fld["name"], "type": fld["type"], + "content": fld.get("content", ""), "lastModified": fld["lastModified"], + "columnIndex": fld["columnIndex"], "files": files, "listOfMaterials": [], + "_links": [], + }) + return { + "id": d["id"], "globalId": d["globalId"], "name": d["name"], "created": d["created"], + "lastModified": d["lastModified"], "parentFolderId": d["parentFolderId"], "tags": d["tags"], + "form": {"id": d["formId"], "globalId": f"FM{d['formId']}", "name": data.forms[d["formId"]]}, + "owner": OWNER, "signed": d.get("signed", False), "fields": fields, + "_links": _self(base, f"/documents/{d['id']}"), + } + + +def ser_inv_file(a: dict, base: str) -> dict: + out = dict(a) + out["_links"] = _self(base, f"/files/{a['id']}") + [ + {"link": f"{base}/files/{a['id']}/file", "rel": "enclosure"}] + return out + + +def ser_summary(gid: str, data: MockData) -> dict: + rec = data.record_by_global_id(gid) + return {"id": rec["id"], "globalId": gid, "name": rec["name"], "type": rec["type"], + "created": rec["created"], "lastModified": rec["lastModified"]} + + +def ser_fields(rec: dict, data: MockData, base: str) -> List[dict]: + out = [] + for f in rec.get("fields", []): + item = {k: f[k] for k in ("id", "globalId", "name", "type", "content")} + att = f.get("attachment_id") + item["attachment"] = ser_inv_file(data.inv_files[att], base) if att is not None else None + out.append(item) + return out + + +def _attachments(rec: dict, data: MockData, base: str) -> List[dict]: + return [ser_inv_file(data.inv_files[i], base) for i in rec["attachment_ids"] if i in data.inv_files] + + +def _parent_containers(gid: Optional[str], data: MockData) -> List[dict]: + chain = [] + while gid is not None: + c = data.record_by_global_id(gid) + chain.append({"id": c["id"], "globalId": gid, "name": c["name"], "cType": c["cType"]}) + gid = c.get("parentGlobalId") + return chain + + +def ser_container(c: dict, data: MockData, base: str, include_content: bool) -> dict: + content = c["content"] + locations = [] + for idx, gid in enumerate(content): + loc = {"id": c["id"] * 1000 + idx, "coordX": (idx % 2) + 1, "coordY": (idx // 2) + 1} + if c["cType"] != "GRID": + loc["coordX"], loc["coordY"] = 1, idx + 1 + loc["content"] = ser_summary(gid, data) if include_content else None + locations.append(loc) + counts = {"totalCount": len(content), + "containerCount": sum(1 for g in content if g.startswith("IC")), + "subSampleCount": sum(1 for g in content if g.startswith("SS"))} + return { + "id": c["id"], "globalId": c["globalId"], "name": c["name"], "type": "CONTAINER", + "cType": c["cType"], "description": c["description"], "created": c["created"], + "lastModified": c["lastModified"], "owner": OWNER, "tags": [], "extraFields": [], + "barcode": None, "deleted": False, "gridLayout": c["gridLayout"], + "canStoreContainers": c["canStoreContainers"], "canStoreSamples": c["canStoreSamples"], + "locationsCount": len(content), "contentSummary": counts, "locations": locations, + "parentContainers": _parent_containers(c["parentGlobalId"], data), + "parentLocation": None, "attachments": _attachments(c, data, base), + "_links": _self(base, f"/containers/{c['id']}"), + } + + +def ser_subsample(ss: dict, data: MockData, base: str, brief: bool = False) -> dict: + sample = data.samples[ss["sampleId"]] + out = { + "id": ss["id"], "globalId": ss["globalId"], "name": ss["name"], "type": "SUBSAMPLE", + "created": ss["created"], "lastModified": ss["lastModified"], "owner": OWNER, + "quantity": ss["quantity"], "tags": ss["tags"], "deleted": False, + "storedInContainer": ss["parentGlobalId"] is not None, + "parentContainers": _parent_containers(ss["parentGlobalId"], data), + "parentLocation": None, "attachments": _attachments(ss, data, base), + "_links": _self(base, f"/subSamples/{ss['id']}"), + } + if not brief: + out["sample"] = {"id": sample["id"], "globalId": sample["globalId"], "name": sample["name"]} + out["extraFields"] = [] + out["notes"] = [] + return out + + +def ser_sample(s: dict, data: MockData, base: str) -> dict: + tpl = data.templates.get(s["templateId"]) if s["templateId"] else None + return { + "id": s["id"], "globalId": s["globalId"], "name": s["name"], "type": "SAMPLE", + "description": s["description"], "created": s["created"], "lastModified": s["lastModified"], + "owner": OWNER, "tags": s["tags"], "quantity": s["quantity"], "deleted": False, + "templateId": s["templateId"], + "template": ({"id": tpl["id"], "globalId": tpl["globalId"], "name": tpl["name"]} if tpl else None), + "subSamplesCount": len(s["subsample_ids"]), + "subSamples": [ser_subsample(data.subsamples[i], data, base, brief=True) for i in s["subsample_ids"]], + "fields": ser_fields(s, data, base), "extraFields": [], "attachments": _attachments(s, data, base), + "_links": _self(base, f"/samples/{s['id']}"), + } + + +def ser_instrument(ins: dict, data: MockData, base: str) -> dict: + return { + "id": ins["id"], "globalId": ins["globalId"], "name": ins["name"], "type": "INSTRUMENT", + "description": ins["description"], "created": ins["created"], "lastModified": ins["lastModified"], + "owner": OWNER, "tags": ins["tags"], "deleted": False, "fields": [], "extraFields": [], + "attachments": _attachments(ins, data, base), + "_links": _self(base, f"/instruments/{ins['id']}"), + } + + +def ser_template(t: dict, data: MockData, base: str) -> dict: + return { + "id": t["id"], "globalId": t["globalId"], "name": t["name"], "type": "SAMPLE_TEMPLATE", + "description": t["description"], "created": t["created"], "lastModified": t["lastModified"], + "owner": OWNER, "tags": t["tags"], "deleted": False, "defaultUnitId": t["defaultUnitId"], + "subSampleAlias": t["subSampleAlias"], "fields": [], "extraFields": [], + "attachments": _attachments(t, data, base), + "_links": _self(base, f"/sampleTemplates/{t['id']}"), + } diff --git a/rspace_client/tests/mock_rspace/server.py b/rspace_client/tests/mock_rspace/server.py new file mode 100644 index 0000000..58cccb0 --- /dev/null +++ b/rspace_client/tests/mock_rspace/server.py @@ -0,0 +1,495 @@ +""" +A small, dependency-free mock of the RSpace ELN (/api/v1) and Inventory +(/api/inventory/v1) REST APIs, covering just the endpoints the PyFilesystem +implementations use. Backed by the in-memory dataset in fixtures.py. + +Run standalone: + python -m rspace_client.tests.mock_rspace.server --port 8765 + +Or embed in a test / the harness: + from rspace_client.tests.mock_rspace.server import run_in_thread + server, url = run_in_thread() # url like http://127.0.0.1:54321 + +Control endpoints (not part of RSpace, not counted in stats): + GET /__mock/stats -> {"requests": N, "by_path": {...}} + POST /__mock/reset -> reset dataset and counters + +Behavioural notes that mirror the real client's expectations: + * every JSON response (including errors) carries Content-Type application/json, + because ClientBase._handle_response indexes that header on failure; + * DELETE returns 204 with no body and no JSON content type, otherwise the + client would try to .json() an empty body; + * paginated listings key the collection by endpoint name (containers, samples, + templates) and emit absolute _links (self/next/prev) on the request's origin, + which ClientBase._stream follows. +""" +from __future__ import annotations + +import argparse +import json +import re +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Callable, Dict, List, Optional, Tuple +from urllib.parse import parse_qs, urlencode, urlparse + +try: # package import (python -m rspace_client.tests.mock_rspace.server, or from tests) + from . import fixtures as fx +except ImportError: # run as a plain script from its own directory + import fixtures as fx # type: ignore + +ELN_PREFIX = "/api/v1" +INV_PREFIX = "/api/inventory/v1" + +Route = Tuple[str, "re.Pattern[str]", Callable] + + +class ApiError(Exception): + def __init__(self, status: int, message: str, errors: Optional[List[str]] = None): + super().__init__(message) + self.status, self.message, self.errors = status, message, errors or [] + + +def parse_multipart(content_type: str, body: bytes) -> Dict[str, Tuple[Optional[str], bytes, str]]: + """Byte-exact multipart/form-data parser. Returns name -> (filename, bytes, part content-type).""" + m = re.search(r'boundary="?([^";]+)"?', content_type or "") + if not m: + raise ApiError(400, "multipart/form-data with boundary expected") + delim = b"--" + m.group(1).encode() + parts: Dict[str, Tuple[Optional[str], bytes, str]] = {} + for chunk in body.split(delim): + if chunk.startswith(b"\r\n"): + chunk = chunk[2:] + if not chunk or chunk.startswith(b"--"): + continue + header_blob, _, payload = chunk.partition(b"\r\n\r\n") + if payload.endswith(b"\r\n"): + payload = payload[:-2] + headers: Dict[str, str] = {} + for line in header_blob.split(b"\r\n"): + k, _, v = line.partition(b":") + headers[k.strip().lower().decode()] = v.strip().decode(errors="replace") + disp = headers.get("content-disposition", "") + name = re.search(r'name="([^"]*)"', disp) + filename = re.search(r'filename="([^"]*)"', disp) + parts[name.group(1) if name else ""] = ( + filename.group(1) if filename else None, payload, + headers.get("content-type", "application/octet-stream")) + return parts + + +class MockRSpaceServer(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + #: What the server writes into its ``_links``. A real RSpace advertises its own configured + #: base URL, which is often not the address the client used (proxy, container, port-forward), + #: so the mock does the same and clients have to cope with it. + advertised_base = "http://rspace.advertised.example:8080" + + def __init__(self, address: Tuple[str, int], data: Optional[fx.MockData] = None, verbose: bool = False): + super().__init__(address, MockRSpaceHandler) + self.data = data or fx.MockData() + self.verbose = verbose + self.stats_lock = threading.Lock() + self.stats: Dict = {"requests": 0, "by_path": {}} + + @property + def url(self) -> str: + host, port = self.server_address[:2] + return f"http://{host}:{port}" + + def reset(self) -> None: + with self.data.lock: + self.data.reset() + with self.stats_lock: + self.stats = {"requests": 0, "by_path": {}} + + +class MockRSpaceHandler(BaseHTTPRequestHandler): + server_version = "MockRSpace/0.1" + server: MockRSpaceServer # type: ignore[assignment] + + # ------------------------------------------------------------ plumbing + + def log_message(self, fmt, *args): # quiet unless --verbose + if self.server.verbose: + super().log_message(fmt, *args) + + def _host_base(self) -> str: + # deliberately not the Host header: see MockRSpaceServer.advertised_base + return self.server.advertised_base + + def _send_json(self, obj, status: int = 200) -> None: + body = json.dumps(obj).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _send_bytes(self, data: bytes, content_type: str, filename: Optional[str] = None) -> None: + self.send_response(200) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(data))) + if filename: + self.send_header("Content-Disposition", f'attachment; filename="{filename}"') + self.end_headers() + self.wfile.write(data) + + def _send_no_content(self) -> None: + self.send_response(204) + self.send_header("Content-Length", "0") + self.end_headers() + + def _read_body(self) -> bytes: + length = int(self.headers.get("Content-Length") or 0) + return self.rfile.read(length) if length else b"" + + def _json_body(self) -> dict: + raw = self._read_body() + try: + return json.loads(raw or b"{}") + except json.JSONDecodeError: + raise ApiError(400, "request body is not valid JSON") + + def _count(self, path: str) -> None: + with self.server.stats_lock: + self.server.stats["requests"] += 1 + key = re.sub(r"/\d+", "/{id}", path) + self.server.stats["by_path"][key] = self.server.stats["by_path"].get(key, 0) + 1 + + # ------------------------------------------------------------ dispatch + + def do_GET(self): + self._dispatch("GET") + + def do_POST(self): + self._dispatch("POST") + + def do_PUT(self): + self._dispatch("PUT") + + def do_DELETE(self): + self._dispatch("DELETE") + + def _dispatch(self, method: str) -> None: + parsed = urlparse(self.path) + path, query = parsed.path, {k: v[0] for k, v in parse_qs(parsed.query).items()} + try: + if path.startswith("/__mock/"): + return self._mock_control(method, path) + self._count(path) + if not self.headers.get("apiKey"): + raise ApiError(401, "Unauthorized: apiKey header is missing") + for prefix, routes in ((INV_PREFIX, INV_ROUTES), (ELN_PREFIX, ELN_ROUTES)): + if path.startswith(prefix + "/") or path == prefix: + sub = path[len(prefix):] or "/" + for m, pattern, handler in routes: + match = pattern.fullmatch(sub) + if m == method and match: + base = self._host_base() + prefix + return handler(self, base, query, **match.groupdict()) + raise ApiError(404, f"no mock route for {method} {path}") + raise ApiError(404, f"unknown path {path}") + except ApiError as e: + self._send_json({"status": e.status, "message": e.message, "errors": e.errors}, e.status) + except KeyError as e: + self._send_json({"status": 404, "message": f"resource not found: {e}", "errors": []}, 404) + except ValueError as e: + self._send_json({"status": 400, "message": str(e), "errors": []}, 400) + + def _mock_control(self, method: str, path: str) -> None: + if path == "/__mock/stats" and method == "GET": + with self.server.stats_lock: + return self._send_json(json.loads(json.dumps(self.server.stats))) + if path == "/__mock/reset" and method == "POST": + self.server.reset() + return self._send_json({"reset": True}) + raise ApiError(404, f"unknown control endpoint {path}") + + # ------------------------------------------------------------ helpers + + def _paginate(self, items: List[dict], query: dict, base: str, endpoint: str, key: str) -> dict: + page = int(query.get("pageNumber", 0)) + size = max(1, int(query.get("pageSize", 20))) + total = len(items) + start = page * size + chunk = items[start:start + size] + + def link(p: int, rel: str) -> dict: + q = dict(query) + q["pageNumber"], q["pageSize"] = p, size + return {"link": f"{base}/{endpoint}?{urlencode(q)}", "rel": rel} + + links = [link(page, "self")] + if start + size < total: + links.append(link(page + 1, "next")) + if page > 0: + links.append(link(page - 1, "prev")) + return {key: chunk, "totalHits": total, "pageNumber": page, "_links": links} + + +# ================================================================ ELN routes + +def eln_status(h: MockRSpaceHandler, base, query): + h._send_json({"message": "OK", "rspaceVersion": "mock-0.1"}) + + +def eln_folder_tree(h: MockRSpaceHandler, base, query, id=None): + data = h.server.data + fid = int(id) if id is not None else data.home_id() + if fid not in data.folders: + raise KeyError(f"FL{fid}") + gids = list(data.children.get(fid, [])) + types = query.get("typesToInclude") + if types: + wanted = {t.strip().lower() for t in types.split(",")} + keep = {"folder": ("FL", "GF"), "notebook": ("NB",), "document": ("SD",)} + allowed = tuple(p for t in wanted for p in keep.get(t, ())) + gids = [g for g in gids if g.startswith(allowed)] + records = [fx.ser_tree_item(g, data, base) for g in gids] + out = h._paginate(records, query, base, f"folders/tree/{fid}", "records") + out["folderId"] = fid + h._send_json(out) + + +def eln_get_folder(h: MockRSpaceHandler, base, query, id): + h._send_json(fx.ser_folder(h.server.data.folders[int(id)], base)) + + +def eln_create_folder(h: MockRSpaceHandler, base, query): + body = h._json_body() + if not body.get("name"): + raise ApiError(400, "name is required", ["name: must not be empty"]) + with h.server.data.lock: + f = h.server.data.add_folder(body["name"], body.get("parentFolderId"), bool(body.get("notebook"))) + h._send_json(fx.ser_folder(f, base), 201) + + +def eln_delete_folder(h: MockRSpaceHandler, base, query, id): + data = h.server.data + with data.lock: + f = data.folders[int(id)] + if f.get("system"): + raise ApiError(403, f"{f['globalId']} is a system folder and cannot be deleted") + data.remove_folder(int(id)) + h._send_no_content() + + +def eln_get_file(h: MockRSpaceHandler, base, query, id): + h._send_json(fx.ser_gallery_file(h.server.data.gallery_files[int(id)], base)) + + +def eln_download_file(h: MockRSpaceHandler, base, query, id): + g = h.server.data.gallery_files[int(id)] + h._send_bytes(h.server.data.blobs[g["globalId"]], g["contentType"], g["name"]) + + +def eln_upload_file(h: MockRSpaceHandler, base, query): + parts = parse_multipart(h.headers.get("Content-Type", ""), h._read_body()) + if "file" not in parts: + raise ApiError(400, "multipart field 'file' is required") + filename, content, ctype = parts["file"] + folder = parts.get("folderId") + folder_id = int(folder[1].decode()) if folder and folder[1] else None + caption = parts["caption"][1].decode() if "caption" in parts else "" + with h.server.data.lock: + name = filename or f"upload-{h.server.data.new_id()}.bin" + g = h.server.data.add_gallery_file(name, folder_id, content, + None if ctype == "application/octet-stream" else ctype, caption) + h._send_json(fx.ser_gallery_file(g, base), 201) + + +def eln_get_document(h: MockRSpaceHandler, base, query, id): + d = h.server.data.documents[int(id)] + h._send_json(fx.ser_document(d, h.server.data, base)) + + +def eln_update_document(h: MockRSpaceHandler, base, query, id): + body = h._json_body() + with h.server.data.lock: + d = h.server.data.update_document(int(id), body) + h._send_json(fx.ser_document(d, h.server.data, base)) + + +ELN_ROUTES: List[Route] = [ + ("GET", re.compile(r"/status"), eln_status), + ("GET", re.compile(r"/folders/tree"), eln_folder_tree), + ("GET", re.compile(r"/folders/tree/(?P\d+)"), eln_folder_tree), + ("GET", re.compile(r"/folders/(?P\d+)"), eln_get_folder), + ("POST", re.compile(r"/folders"), eln_create_folder), + ("DELETE", re.compile(r"/folders/(?P\d+)"), eln_delete_folder), + ("GET", re.compile(r"/files/(?P\d+)"), eln_get_file), + ("GET", re.compile(r"/files/(?P\d+)/file"), eln_download_file), + ("POST", re.compile(r"/files"), eln_upload_file), + ("GET", re.compile(r"/documents/(?P\d+)"), eln_get_document), + ("PUT", re.compile(r"/documents/(?P\d+)"), eln_update_document), +] + + +# ============================================================ Inventory routes + +def _truthy(v: Optional[str]) -> bool: + return str(v).lower() in ("true", "1", "yes") + + +def inv_workbenches(h: MockRSpaceHandler, base, query): + data = h.server.data + h._send_json({"containers": [fx.ser_container(b, data, base, include_content=False) for b in data.benches()]}) + + +def inv_list_containers(h: MockRSpaceHandler, base, query): + data = h.server.data + items = [fx.ser_container(c, data, base, include_content=False) for c in data.top_level_containers()] + h._send_json(h._paginate(items, query, base, "containers", "containers")) + + +def inv_get_container(h: MockRSpaceHandler, base, query, id): + data = h.server.data + container = data.containers[int(id)] + if container["cType"] == "WORKBENCH": # mirrors the real server + raise ApiError(422, f"Container with id {id} is a workbench", [f"Container with id {id} is a workbench"]) + h._send_json(fx.ser_container(container, data, base, _truthy(query.get("includeContent")))) + + +def inv_get_workbench(h: MockRSpaceHandler, base, query, id): + data = h.server.data + bench = data.containers[int(id)] + if bench["cType"] != "WORKBENCH": + raise ApiError(404, f"no workbench with id {id}") + h._send_json(fx.ser_container(bench, data, base, _truthy(query.get("includeContent")))) + + +def inv_list_samples(h: MockRSpaceHandler, base, query): + data = h.server.data + items = [fx.ser_sample(s, data, base) for s in data.samples.values()] + h._send_json(h._paginate(items, query, base, "samples", "samples")) + + +def inv_get_sample(h: MockRSpaceHandler, base, query, id): + data = h.server.data + h._send_json(fx.ser_sample(data.samples[int(id)], data, base)) + + +def inv_get_subsample(h: MockRSpaceHandler, base, query, id): + data = h.server.data + h._send_json(fx.ser_subsample(data.subsamples[int(id)], data, base)) + + +def inv_list_templates(h: MockRSpaceHandler, base, query): + data = h.server.data + items = [fx.ser_template(t, data, base) for t in data.templates.values()] + h._send_json(h._paginate(items, query, base, "sampleTemplates", "templates")) + + +def inv_get_template(h: MockRSpaceHandler, base, query, id): + data = h.server.data + h._send_json(fx.ser_template(data.templates[int(id)], data, base)) + + +def inv_get_instrument(h: MockRSpaceHandler, base, query, id): + data = h.server.data + h._send_json(fx.ser_instrument(data.instruments[int(id)], data, base)) + + +def inv_get_file(h: MockRSpaceHandler, base, query, id): + h._send_json(fx.ser_inv_file(h.server.data.inv_files[int(id)], base)) + + +def inv_download_file(h: MockRSpaceHandler, base, query, id): + a = h.server.data.inv_files[int(id)] + h._send_bytes(h.server.data.blobs[a["globalId"]], a["contentMimeType"], a["name"]) + + +def inv_upload_file(h: MockRSpaceHandler, base, query): + parts = parse_multipart(h.headers.get("Content-Type", ""), h._read_body()) + if "file" not in parts or "fileSettings" not in parts: + raise ApiError(400, "multipart fields 'file' and 'fileSettings' are required") + filename, content, ctype = parts["file"] + try: + settings = json.loads(parts["fileSettings"][1].decode()) + except json.JSONDecodeError: + raise ApiError(400, "fileSettings must be JSON") + parent = settings.get("parentGlobalId") + if not parent: + raise ApiError(400, "fileSettings.parentGlobalId is required") + with h.server.data.lock: + name = filename or f"upload-{h.server.data.new_id()}.bin" + a = h.server.data.add_inv_file(parent, name, content, + None if ctype == "application/octet-stream" else ctype) + h._send_json(fx.ser_inv_file(a, base), 201) + + +def inv_attach_gallery_file(h: MockRSpaceHandler, base, query): + body = h._json_body() + parent, media = body.get("parentGlobalId"), body.get("mediaFileGlobalId") + if not parent or not media: + raise ApiError(400, "parentGlobalId and mediaFileGlobalId are required") + with h.server.data.lock: + try: + a = h.server.data.link_gallery_file(parent, media) + except ValueError as exc: + raise ApiError(422, str(exc)) + h._send_json(fx.ser_inv_file(a, base), 201) + + +def inv_delete_file(h: MockRSpaceHandler, base, query, id): + with h.server.data.lock: + h.server.data.remove_inv_file(int(id)) + h._send_no_content() + + +INV_ROUTES: List[Route] = [ + ("GET", re.compile(r"/workbenches"), inv_workbenches), + ("GET", re.compile(r"/workbenches/(?P\d+)"), inv_get_workbench), + ("GET", re.compile(r"/containers"), inv_list_containers), + ("GET", re.compile(r"/containers/(?P\d+)"), inv_get_container), + ("GET", re.compile(r"/samples"), inv_list_samples), + ("GET", re.compile(r"/samples/(?P\d+)"), inv_get_sample), + ("GET", re.compile(r"/subSamples/(?P\d+)"), inv_get_subsample), + ("GET", re.compile(r"/sampleTemplates"), inv_list_templates), + ("GET", re.compile(r"/sampleTemplates/(?P\d+)"), inv_get_template), + ("GET", re.compile(r"/instruments/(?P\d+)"), inv_get_instrument), + ("GET", re.compile(r"/files/(?P\d+)"), inv_get_file), + ("GET", re.compile(r"/files/(?P\d+)/file"), inv_download_file), + ("POST", re.compile(r"/files"), inv_upload_file), + ("POST", re.compile(r"/attachments"), inv_attach_gallery_file), + ("DELETE", re.compile(r"/files/(?P\d+)"), inv_delete_file), +] + + +# ================================================================ entry points + +def make_server(host: str = "127.0.0.1", port: int = 8765, verbose: bool = False) -> MockRSpaceServer: + return MockRSpaceServer((host, port), verbose=verbose) + + +def run_in_thread(host: str = "127.0.0.1", port: int = 0, verbose: bool = False) -> Tuple[MockRSpaceServer, str]: + """Start the server on a background daemon thread. port=0 picks a free port. Returns (server, url).""" + server = make_server(host, port, verbose) + threading.Thread(target=server.serve_forever, name="mock-rspace", daemon=True).start() + return server, server.url + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description="Mock RSpace ELN + Inventory API server") + ap.add_argument("--host", default="127.0.0.1") + ap.add_argument("--port", type=int, default=8765) + ap.add_argument("--verbose", "-v", action="store_true", help="log every request") + args = ap.parse_args(argv) + server = make_server(args.host, args.port, args.verbose) + print(f"Mock RSpace listening on {server.url} (ELN: {ELN_PREFIX}, Inventory: {INV_PREFIX})") + print("Use any non-empty apiKey. Ctrl-C to stop.") + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/rspace_client/tests/mock_server_case.py b/rspace_client/tests/mock_server_case.py new file mode 100644 index 0000000..69e133f --- /dev/null +++ b/rspace_client/tests/mock_server_case.py @@ -0,0 +1,61 @@ +""" +Shared fixture for the offline filesystem tests. + +The embedded mock RSpace (``rspace_client/tests/mock_rspace``) is started once per +session by ``conftest.py`` and installed here; every ``MockServerTestCase`` subclass +talks to that one server and begins each test from the stock fixtures +(``server.reset()``). Under plain ``python -m unittest`` no conftest runs, so the +first class to need a server starts one lazily and keeps it for the process. +""" +import atexit +import unittest +from typing import Optional, Tuple + +import requests + +from rspace_client.tests.mock_rspace.server import MockRSpaceServer, run_in_thread + +_shared: Tuple[Optional[MockRSpaceServer], Optional[str]] = (None, None) +_owned: Tuple[Optional[MockRSpaceServer], Optional[str]] = (None, None) + + +def install_shared_server(server: Optional[MockRSpaceServer], url: Optional[str]) -> None: + """Called by the pytest session fixture with the session's server, then with None at teardown.""" + global _shared + _shared = (server, url) + + +def _stop_owned() -> None: + server, _ = _owned + if server is not None: + server.shutdown() + server.server_close() + + +def shared_server() -> Tuple[MockRSpaceServer, str]: + """The session server if pytest installed one, else a process-wide fallback started on demand.""" + global _owned + if _shared[0] is not None: + return _shared # type: ignore[return-value] + if _owned[0] is None: + _owned = run_in_thread() + atexit.register(_stop_owned) + return _owned # type: ignore[return-value] + + +class MockServerTestCase(unittest.TestCase): + + server: MockRSpaceServer + url: str + + @classmethod + def setUpClass(cls): + cls.server, cls.url = shared_server() + + def setUp(self): + self.server.reset() + + def calls(self) -> int: + """API requests the mock has served since the last reset, for the tests that + assert a listing costs one call per directory.""" + return requests.get(f"{self.url}/__mock/stats").json()["requests"]