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"]