506 lines
18 KiB
Python
506 lines
18 KiB
Python
"""Test doubles shared across the suite.
|
|
|
|
Three fakes live here, each at a different seam:
|
|
|
|
* :class:`FakeElabftwClient` duck-types the ``ElabftwClient`` adapter interface.
|
|
It is a stateful in-memory eLabFTW instance used by workflow/service tests,
|
|
so saga behaviour can be asserted against real state changes.
|
|
* :class:`StubResolver` duck-types ``ResourceResolver`` with a plain dict of
|
|
mappings, keeping workflow tests independent of the SQLite mapping store.
|
|
* :func:`stub_elabapi` installs fake ``elabapi_python`` API classes into the
|
|
``mcp_server.elabftw`` module namespace. Convention under test: the adapter
|
|
resolves vendored API classes as module globals of ``mcp_server.elabftw``
|
|
(``from elabapi_python import ExperimentsApi`` ...) and instantiates them
|
|
with an api_client. If the implementation wires this differently, only the
|
|
injection point in these stubs needs to move.
|
|
|
|
Scenario builders (``make_experiment``, ``make_item`` ...) produce the standard
|
|
fixtures referenced throughout the tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from mcp_server.errors import ClarificationNeededError, NotFoundError, PermissionDeniedError
|
|
from mcp_server.models import (
|
|
AllocationPolicy,
|
|
MappingEntry,
|
|
MappingProposal,
|
|
)
|
|
from mcp_server.schemas import (
|
|
ContainerLink,
|
|
ElabftwComment,
|
|
ElabftwExperiment,
|
|
ElabftwInfo,
|
|
ElabftwItem,
|
|
ElabftwStep,
|
|
ElabftwTemplate,
|
|
StorageUnit,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Annotation snippets (the labvoice:v1 format from planning §4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ANNOTATION_ETHANOL_FIFO = (
|
|
"<!-- labvoice:v1\n"
|
|
'{"consumables": [{"resource_key": "ethanol_absolute", "quantity": 2.0,'
|
|
' "unit": "mL", "allocation": "fifo"}]}\n'
|
|
"-->"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fake eLabFTW state (adapter-level)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class FakeStep:
|
|
id: int
|
|
body: str
|
|
finished: bool = False
|
|
ordering: int | None = None
|
|
|
|
|
|
@dataclass
|
|
class FakeComment:
|
|
id: int
|
|
body: str
|
|
created_at: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class FakeContainer:
|
|
sub_id: int
|
|
qty_stored: float
|
|
unit: str
|
|
|
|
|
|
@dataclass
|
|
class FakeItem:
|
|
id: int
|
|
title: str
|
|
body: str = ""
|
|
extra_fields: dict = field(default_factory=dict)
|
|
containers: list[FakeContainer] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class FakeExperiment:
|
|
id: int
|
|
title: str
|
|
status: str = "running"
|
|
steps: list[FakeStep] = field(default_factory=list)
|
|
comments: list[FakeComment] = field(default_factory=list)
|
|
linked_items: list[int] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class FakeTemplate:
|
|
id: int
|
|
title: str
|
|
description: str = ""
|
|
tags: list[str] = field(default_factory=list)
|
|
steps: list[FakeStep] = field(default_factory=list)
|
|
|
|
|
|
def make_template(**overrides) -> FakeTemplate:
|
|
"""Standard template: two plain steps, description with two paragraphs."""
|
|
tpl = FakeTemplate(
|
|
id=7,
|
|
title="PCR cleanup",
|
|
description="Purify an amplicon after PCR.\n\nLonger protocol notes follow here.",
|
|
tags=["dna", "cleanup"],
|
|
steps=[
|
|
FakeStep(id=41, body="Add binding buffer", ordering=1),
|
|
FakeStep(id=42, body="Elute", ordering=2),
|
|
],
|
|
)
|
|
for key, value in overrides.items():
|
|
setattr(tpl, key, value)
|
|
return tpl
|
|
|
|
|
|
def make_experiment(**overrides) -> FakeExperiment:
|
|
"""Standard experiment: step 9 (annotated) and step 10 (plain), both unfinished."""
|
|
exp = FakeExperiment(
|
|
id=123,
|
|
title="Plasmid prep",
|
|
steps=[
|
|
FakeStep(
|
|
id=9,
|
|
body=f"Add ethanol to the tube\n{ANNOTATION_ETHANOL_FIFO}",
|
|
ordering=1,
|
|
),
|
|
FakeStep(id=10, body="Incubate 30 min", ordering=2),
|
|
],
|
|
linked_items=[12],
|
|
)
|
|
for key, value in overrides.items():
|
|
setattr(exp, key, value)
|
|
return exp
|
|
|
|
|
|
def make_item(**overrides) -> FakeItem:
|
|
"""Standard resource: ethanol with the marker comment and two containers."""
|
|
item = FakeItem(
|
|
id=12,
|
|
title="Absolute Ethanol",
|
|
body="<p>Absolutely ethanol.</p>\n<!-- labvoice:resource-key=ethanol_absolute -->",
|
|
containers=[
|
|
FakeContainer(sub_id=31, qty_stored=50.0, unit="mL"),
|
|
FakeContainer(sub_id=32, qty_stored=10.0, unit="mL"),
|
|
],
|
|
)
|
|
for key, value in overrides.items():
|
|
setattr(item, key, value)
|
|
return item
|
|
|
|
|
|
class FakeElabftwClient:
|
|
"""In-memory stand-in for the async eLabFTW adapter.
|
|
|
|
Mirrors the ``ElabftwClient`` method surface. All mutations are recorded
|
|
in ``self.calls`` as ``(method_name, kwargs)`` tuples so tests can assert
|
|
exactly what changed (and that nothing changed when a workflow must abort).
|
|
``read_only=True`` makes every mutating method raise
|
|
``PermissionDeniedError`` like a canwrite=false API key would.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
experiments: list[FakeExperiment] | None = None,
|
|
items: list[FakeItem] | None = None,
|
|
templates: list[FakeTemplate] | None = None,
|
|
storage_units: list[StorageUnit] | None = None,
|
|
read_only: bool = False,
|
|
) -> None:
|
|
self.experiments = {e.id: e for e in (experiments or [])}
|
|
self.items = {i.id: i for i in (items or [])}
|
|
self.templates = {t.id: t for t in (templates or [])}
|
|
self.storage_units = storage_units or [
|
|
StorageUnit(id=1, label="Bench fridge / Shelf 2")
|
|
]
|
|
self.read_only = read_only
|
|
self.calls: list[tuple[str, dict]] = []
|
|
self._next_comment_id = 500
|
|
self._next_experiment_id = 1000
|
|
|
|
# -- call recording ----------------------------------------------------
|
|
|
|
def _record(self, method: str, **kwargs) -> None:
|
|
self.calls.append((method, kwargs))
|
|
|
|
def calls_of(self, method: str) -> list[dict]:
|
|
return [kwargs for name, kwargs in self.calls if name == method]
|
|
|
|
def _deny_if_read_only(self) -> None:
|
|
if self.read_only:
|
|
raise PermissionDeniedError(
|
|
"The API key cannot modify eLabFTW; ask an admin for a writable key."
|
|
)
|
|
|
|
# -- reads ---------------------------------------------------------------
|
|
|
|
async def get_info(self) -> ElabftwInfo:
|
|
self._record("get_info")
|
|
return ElabftwInfo(elabftw_version="5.0.0", api_version="2")
|
|
|
|
async def search_experiments(self, *, q=None, tags=None, limit=15, offset=0):
|
|
self._record("search_experiments", q=q, tags=tags, limit=limit, offset=offset)
|
|
hits = list(self.experiments.values())
|
|
if q:
|
|
hits = [e for e in hits if q.lower() in e.title.lower()]
|
|
if tags:
|
|
hits = [e for e in hits if any(t in tags for t in getattr(e, "tags", []))]
|
|
return [
|
|
ElabftwExperiment(id=e.id, title=e.title, status=e.status)
|
|
for e in hits[offset : offset + limit]
|
|
]
|
|
|
|
async def get_experiment(self, experiment_id: int) -> ElabftwExperiment:
|
|
self._record("get_experiment", experiment_id=experiment_id)
|
|
exp = self._entity("experiments", experiment_id)
|
|
return ElabftwExperiment(id=exp.id, title=exp.title, status=exp.status)
|
|
|
|
async def create_experiment(self, template_id: int, title=None):
|
|
self._record("create_experiment", template_id=template_id, title=title)
|
|
self._deny_if_read_only()
|
|
if template_id not in self.templates:
|
|
raise NotFoundError(f"Template {template_id} not found.")
|
|
tpl = self.templates[template_id]
|
|
new_id = self._next_experiment_id
|
|
self._next_experiment_id += 1
|
|
exp = FakeExperiment(
|
|
id=new_id,
|
|
title=title or tpl.title,
|
|
steps=[
|
|
FakeStep(id=s.id, body=s.body, ordering=s.ordering) for s in tpl.steps
|
|
],
|
|
)
|
|
self.experiments[new_id] = exp
|
|
return ElabftwExperiment(id=exp.id, title=exp.title, status=exp.status)
|
|
|
|
async def get_steps(self, entity_type, entity_id):
|
|
self._record("get_steps", entity_type=entity_type, entity_id=entity_id)
|
|
steps = self._entity(entity_type, entity_id).steps
|
|
return [
|
|
ElabftwStep(id=s.id, body=s.body, finished=s.finished, ordering=s.ordering)
|
|
for s in steps
|
|
]
|
|
|
|
async def finish_step(self, entity_type, entity_id, step_id):
|
|
self._record(
|
|
"finish_step", entity_type=entity_type, entity_id=entity_id, step_id=step_id
|
|
)
|
|
self._deny_if_read_only()
|
|
for step in self._entity(entity_type, entity_id).steps:
|
|
if step.id == step_id:
|
|
if step.finished:
|
|
return ElabftwStep(
|
|
id=step.id, body=step.body, finished=True, ordering=step.ordering
|
|
)
|
|
step.finished = True
|
|
return ElabftwStep(
|
|
id=step.id, body=step.body, finished=True, ordering=step.ordering
|
|
)
|
|
raise NotFoundError(f"Step {step_id} not found on {entity_type} {entity_id}.")
|
|
|
|
async def list_comments(self, entity_type, entity_id):
|
|
self._record("list_comments", entity_type=entity_type, entity_id=entity_id)
|
|
return [
|
|
ElabftwComment(id=c.id, body=c.body, created_at=c.created_at)
|
|
for c in self._entity(entity_type, entity_id).comments
|
|
]
|
|
|
|
async def post_comment(self, entity_type, entity_id, body):
|
|
self._record(
|
|
"post_comment", entity_type=entity_type, entity_id=entity_id, body=body
|
|
)
|
|
self._deny_if_read_only()
|
|
comment = FakeComment(id=self._next_comment_id, body=body)
|
|
self._next_comment_id += 1
|
|
self._entity(entity_type, entity_id).comments.append(comment)
|
|
return ElabftwComment(id=comment.id, body=comment.body, created_at=None)
|
|
|
|
async def search_items(self, *, q=None, limit=15, offset=0):
|
|
self._record("search_items", q=q, limit=limit, offset=offset)
|
|
hits = list(self.items.values())
|
|
if q:
|
|
hits = [i for i in hits if q.lower() in i.title.lower()]
|
|
return [self._item_model(i) for i in hits[offset : offset + limit]]
|
|
|
|
async def get_item(self, item_id: int) -> ElabftwItem:
|
|
self._record("get_item", item_id=item_id)
|
|
return self._item_model(self._entity("items", item_id))
|
|
|
|
async def get_containers(self, entity_type, entity_id):
|
|
self._record("get_containers", entity_type=entity_type, entity_id=entity_id)
|
|
item = self._entity("items", entity_id)
|
|
return [
|
|
ContainerLink(id=c.sub_id, qty_stored=c.qty_stored, unit=c.unit)
|
|
for c in item.containers
|
|
]
|
|
|
|
async def set_container_quantity(self, entity_type, entity_id, subitem_id, qty_stored):
|
|
self._record(
|
|
"set_container_quantity",
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
subitem_id=subitem_id,
|
|
qty_stored=qty_stored,
|
|
)
|
|
self._deny_if_read_only()
|
|
for container in self._entity("items", entity_id).containers:
|
|
if container.sub_id == subitem_id:
|
|
container.qty_stored = qty_stored
|
|
return
|
|
raise NotFoundError(f"Container {subitem_id} not found on item {entity_id}.")
|
|
|
|
async def list_storage_units(self):
|
|
self._record("list_storage_units")
|
|
return list(self.storage_units)
|
|
|
|
async def list_templates(self, *, q=None, limit=25, offset=0):
|
|
self._record("list_templates", q=q, limit=limit, offset=offset)
|
|
hits = list(self.templates.values())
|
|
if q:
|
|
hits = [t for t in hits if q.lower() in t.title.lower()]
|
|
return [
|
|
ElabftwTemplate(id=t.id, title=t.title, description=t.description, tags=t.tags)
|
|
for t in hits[offset : offset + limit]
|
|
]
|
|
|
|
async def get_template(self, template_id: int) -> ElabftwTemplate:
|
|
self._record("get_template", template_id=template_id)
|
|
if template_id not in self.templates:
|
|
raise NotFoundError(f"Template {template_id} not found.")
|
|
t = self.templates[template_id]
|
|
return ElabftwTemplate(id=t.id, title=t.title, description=t.description, tags=t.tags)
|
|
|
|
# -- helpers -------------------------------------------------------------
|
|
|
|
def _entity(self, entity_type, entity_id):
|
|
if entity_type == "experiments":
|
|
if entity_id not in self.experiments:
|
|
raise NotFoundError(f"Experiment {entity_id} not found.")
|
|
return self.experiments[entity_id]
|
|
if entity_id not in self.items:
|
|
raise NotFoundError(f"Item {entity_id} not found.")
|
|
return self.items[entity_id]
|
|
|
|
def _item_model(self, item: FakeItem) -> ElabftwItem:
|
|
return ElabftwItem(
|
|
id=item.id,
|
|
title=item.title,
|
|
body=item.body,
|
|
**{"extra_fields": item.extra_fields} if item.extra_fields else {},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stub resolver (resolver-level)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class StubResolver:
|
|
"""Dict-backed ``ResourceResolver`` stand-in for workflow tests."""
|
|
|
|
def __init__(
|
|
self,
|
|
mappings: dict[str, int] | None = None,
|
|
*,
|
|
fail_on: set[str] | None = None,
|
|
options_by_key: dict[str, list[str]] | None = None,
|
|
) -> None:
|
|
self.mappings = dict(mappings or {})
|
|
self.fail_on = fail_on or set()
|
|
self.options_by_key = options_by_key or {}
|
|
self.resolved: list[str] = []
|
|
|
|
async def resolve(self, resource_key: str) -> int:
|
|
self.resolved.append(resource_key)
|
|
if resource_key in self.fail_on or resource_key not in self.mappings:
|
|
raise ClarificationNeededError(
|
|
f"I don't know which resource '{resource_key}' refers to.",
|
|
options=self.options_by_key.get(resource_key, []),
|
|
)
|
|
return self.mappings[resource_key]
|
|
|
|
async def find_candidates(self, resource_key: str) -> list[MappingProposal]:
|
|
return [
|
|
MappingProposal(resource_key=resource_key, item_id=item_id)
|
|
for item_id in sorted(self.mappings.values())
|
|
]
|
|
|
|
async def upsert_mapping(self, resource_key: str, item_id: int) -> MappingEntry:
|
|
self.mappings[resource_key] = item_id
|
|
return MappingEntry(resource_key=resource_key, item_id=item_id)
|
|
|
|
async def list_mappings(self) -> list[MappingEntry]:
|
|
return [
|
|
MappingEntry(resource_key=k, item_id=v)
|
|
for k, v in sorted(self.mappings.items())
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# elabapi_python stubs (client construction seam, adapter tests only)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class StubElabapi:
|
|
"""Fake vendored elabapi_python client hub.
|
|
|
|
Each generated ``*Api`` class is replaced by a thin proxy that forwards
|
|
every method call to :meth:`handle`, which consults routes registered by
|
|
the test. Unrouted calls fail loudly, so adapter tests only ever pass
|
|
against the endpoints they explicitly define.
|
|
"""
|
|
|
|
API_CLASSES = (
|
|
"InfoApi",
|
|
"InstanceApi",
|
|
"ExperimentsApi",
|
|
"ExperimentsTemplatesApi",
|
|
"StepsApi",
|
|
"CommentsApi",
|
|
"ItemsApi",
|
|
"ContainersApi",
|
|
"StorageUnitsApi",
|
|
"LinksToItemsApi",
|
|
"UnfinishedStepsApi",
|
|
"TodolistApi",
|
|
)
|
|
|
|
def __init__(self) -> None:
|
|
self.routes: dict[tuple[str, str], object] = {}
|
|
self.calls: list[tuple[str, str, dict]] = []
|
|
self.thread_ids: list[tuple[str, int]] = []
|
|
|
|
def route(self, api: str, method: str, handler) -> None:
|
|
self.routes[(api, method)] = handler
|
|
|
|
async def handle(self, api: str, method: str, args: tuple, kwargs: dict):
|
|
self.calls.append((api, method, kwargs))
|
|
self.thread_ids.append((f"{api}.{method}", __import__("threading").get_ident()))
|
|
if (api, method) not in self.routes and (api, "*") not in self.routes:
|
|
raise AssertionError(f"Unexpected elabapi call: {api}.{method}{args}{kwargs}")
|
|
handler = self.routes.get((api, method)) or self.routes[(api, "*")]
|
|
return handler(*args, **kwargs)
|
|
|
|
|
|
def make_stub_elabapi_cls(hub: StubElabapi, api_name: str):
|
|
class _ApiProxy:
|
|
def __init__(self, api_client=None, *args, **kwargs):
|
|
self.api_client = api_client
|
|
|
|
def __getattr__(self, method: str):
|
|
async def _call(*args, **kwargs):
|
|
return await hub.handle(api_name, method, args, kwargs)
|
|
|
|
_call.__name__ = method
|
|
return _call
|
|
|
|
_ApiProxy.__name__ = f"Stub{api_name}"
|
|
return _ApiProxy
|
|
|
|
|
|
def stub_elabapi(monkeypatch, hub: StubElabapi) -> StubElabapi:
|
|
"""Patch elabapi API classes (and Configuration/ApiClient) into mcp_server.elabftw."""
|
|
import elabapi_python.rest as elabapi_rest
|
|
import mcp_server.elabftw as elabftw_module
|
|
|
|
for api_name in hub.API_CLASSES:
|
|
monkeypatch.setattr(
|
|
elabftw_module,
|
|
api_name,
|
|
make_stub_elabapi_cls(hub, api_name),
|
|
raising=False,
|
|
)
|
|
|
|
class _StubConfiguration:
|
|
def __init__(self, host=None, api_key=None, username=None, password=None, **kw):
|
|
self.host = host
|
|
self.api_key = api_key
|
|
self.username = username
|
|
self.password = password
|
|
|
|
class _StubApiClient:
|
|
def __init__(self, configuration=None, **kw):
|
|
self.configuration = configuration
|
|
|
|
def set_default_header(self, name, value):
|
|
pass
|
|
|
|
monkeypatch.setattr(elabftw_module, "Configuration", _StubConfiguration, raising=False)
|
|
monkeypatch.setattr(elabftw_module, "ApiClient", _StubApiClient, raising=False)
|
|
monkeypatch.setattr(
|
|
elabftw_module, "ApiException", elabapi_rest.ApiException, raising=False
|
|
)
|
|
return hub
|