From f76ba634e0665b8358602b476c148f25283dab6a Mon Sep 17 00:00:00 2001 From: jepegit Date: Wed, 23 Sep 2026 09:19:21 +0200 Subject: [PATCH] List raw cells after confirm and load configured remote URIs. Scenario 2 needs an explicit kind=raw search, needs_metadata, and a prefix-checked load so OtherPath rawdatadir works without opening the pathlib sandbox. --- README.md | 2 +- src/cellpy_mcp/cells.py | 46 +++++++++++---- src/cellpy_mcp/sandbox.py | 47 +++++++++++++-- tests/test_cell_tools.py | 120 ++++++++++++++++++++++++++++++++++++++ tests/test_sandbox.py | 24 +++++++- 5 files changed, 222 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 770edae..4838927 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ Cells and figures: | Tool | What it gives you | |---|---| | `list_instruments` | loaders, and whether each can actually run on this machine | -| `find_cells` | local files by project + number range; empty cellpy → `offer_raw` | +| `find_cells` | files by project + number range; empty cellpy → `offer_raw`; `kind=raw` lists raw (incl. configured remote URIs) and sets `needs_metadata` | | `load_cell` | a handle, cycle count, mass, optional nominal_capacity, summary column names | | `list_cells` | what is loaded | | `describe_plot_families` | the 20 summary families, marked available or missing-columns | diff --git a/src/cellpy_mcp/cells.py b/src/cellpy_mcp/cells.py index 4d482a7..51c57ef 100644 --- a/src/cellpy_mcp/cells.py +++ b/src/cellpy_mcp/cells.py @@ -17,7 +17,7 @@ from typing import Any -from .sandbox import Refused, _is_local_dir +from .sandbox import Refused, _is_local_dir, _is_remote_uri, configured_remote_root __all__ = ["register", "MAX_PREVIEW_ROWS"] @@ -63,11 +63,14 @@ def find_cells( number_max: int, kind: str = "cellpy", ) -> dict: - """List local cellpy or raw files for a project token and number range. + """List cellpy or raw files for a project token and number range. Paths come from cellpy config. `kind="cellpy"` never searches raw files. An empty cellpy result sets `offer_raw` so you can ask the - user before calling again with `kind="raw"`. + user before calling again with `kind="raw"`. Raw results include + `needs_metadata` so you ask for mass and nominal capacity instead + of guessing. A remote `rawdatadir` is listed (URIs); load those + only when they sit under the configured root. """ from cellpy import config, filefinder @@ -76,11 +79,12 @@ def find_cells( setting = "cellpydatadir" if kind == "cellpy" else "rawdatadir" configured = getattr(config.paths, setting, None) - if configured is not None and not _is_local_dir(configured): + remote = configured is not None and not _is_local_dir(configured) + if kind == "cellpy" and remote: return { "found": 0, "cells": [], - "offer_raw": kind == "cellpy", + "offer_raw": True, "remote": True, "reason": ( f"{setting} is a remote URI; this server only walks " @@ -99,8 +103,18 @@ def find_cells( hits = finder(project, number_min, number_max, kind=kind) cells_out = [] for hit in hits: + raw_path = str(hit["path"]) + if remote or _is_remote_uri(raw_path): + cells_out.append( + { + "name": hit["name"], + "number": hit["number"], + "path": raw_path, + } + ) + continue try: - path = sandbox.resolve(hit["path"]) + path = sandbox.resolve(raw_path) except Refused: continue cells_out.append( @@ -111,13 +125,16 @@ def find_cells( } ) empty = not cells_out - return { + result = { "found": len(cells_out), "cells": cells_out, "offer_raw": kind == "cellpy" and empty, - "remote": False, + "remote": remote, **sandbox.describe(), } + if kind == "raw": + result["needs_metadata"] = ["mass", "nominal_capacity"] + return result @server.tool() def load_cell( @@ -138,8 +155,17 @@ def load_cell( """ import cellpy - target = sandbox.resolve(path) - kwargs: dict[str, Any] = {"filename": str(target)} + if _is_remote_uri(path): + root = configured_remote_root(path) + if root is None: + raise Refused( + f"{path!r} is not under a configured remote data " + "directory (rawdatadir / cellpydatadir)." + ) + filename = path.strip() + else: + filename = str(sandbox.resolve(path)) + kwargs: dict[str, Any] = {"filename": filename} if instrument: kwargs["instrument"] = instrument if mass_mg is not None: diff --git a/src/cellpy_mcp/sandbox.py b/src/cellpy_mcp/sandbox.py index d04380b..27f57c2 100644 --- a/src/cellpy_mcp/sandbox.py +++ b/src/cellpy_mcp/sandbox.py @@ -43,6 +43,15 @@ FALLBACK_ROOT = Path.home() / "cellpy_mcp" +def _is_remote_uri(value: object) -> bool: + """True for `scp://…` / `sftp://…` / `ssh://…` — not `C:\\…`.""" + text = str(value or "").strip() + if not text: + return False + head, separator, _rest = text.partition("://") + return bool(separator and len(head) > 1) + + def _is_local_dir(value: object) -> bool: """True when `value` names a plain local directory we can contain paths in. @@ -52,12 +61,40 @@ def _is_local_dir(value: object) -> bool: text = str(value or "").strip() if not text: return False - # `scp://…`, `sftp://…`, `ssh://…`. A Windows drive (`C:\…`) is not a - # scheme, hence the length guard on what precedes the colon. - head, separator, _rest = text.partition("://") - if separator and len(head) > 1: + return not _is_remote_uri(text) + + +def _uri_under_prefix(path: str, prefix: str) -> bool: + """True when *path* is *prefix* or a child of it, with no `..` segments.""" + candidate = path.strip().rstrip("/") + root = prefix.strip().rstrip("/") + if candidate == root: + return True + if not candidate.startswith(root + "/"): return False - return True + rest = candidate[len(root) + 1 :] + return not any(part == ".." for part in rest.split("/")) + + +def configured_remote_root(path: str) -> str | None: + """Configured `rawdatadir` / `cellpydatadir` URI that contains *path*. + + Local pathlib roots are not consulted here. A miss means the caller must + go through `Sandbox.resolve` or be refused. + """ + try: + from cellpy import config + except Exception: # noqa: BLE001 - unconfigured cellpy is a miss + return None + + for setting in ("rawdatadir", "cellpydatadir"): + value = getattr(config.paths, setting, None) + if value is None or not _is_remote_uri(value): + continue + prefix = str(value) + if _uri_under_prefix(path, prefix): + return prefix.rstrip("/") + return None def default_roots() -> list[Path]: diff --git a/tests/test_cell_tools.py b/tests/test_cell_tools.py index 19aab9e..96b832a 100644 --- a/tests/test_cell_tools.py +++ b/tests/test_cell_tools.py @@ -74,6 +74,126 @@ async def steps(call): assert out["offer_raw"] is True +def test_find_cells_raw_lists_hits_and_needs_metadata(drive, root, monkeypatch): + target = root / "20240922_SAL12.res" + target.write_text("x") + + def fake(project, number_min, number_max, *, kind="cellpy", root=None): + return [ + { + "path": str(target), + "name": target.name, + "number": 12, + "kind": kind, + } + ] + + monkeypatch.setattr("cellpy.filefinder.find_by_project", fake, raising=False) + + async def steps(call): + return await call( + "find_cells", + project="SAL", + number_min=10, + number_max=15, + kind="raw", + ) + + out = drive(steps) + assert out["found"] == 1 + assert out["offer_raw"] is False + assert out["needs_metadata"] == ["mass", "nominal_capacity"] + assert out["cells"][0]["number"] == 12 + + +def test_find_cells_raw_remote_returns_uris(drive, monkeypatch): + from cellpy import config + + uri = "scp://host/raw/20240922_SAL12.res" + monkeypatch.setattr(config.paths, "rawdatadir", "scp://host/raw") + + def fake(project, number_min, number_max, *, kind="cellpy", root=None): + return [{"path": uri, "name": "20240922_SAL12.res", "number": 12, "kind": kind}] + + monkeypatch.setattr("cellpy.filefinder.find_by_project", fake, raising=False) + + async def steps(call): + return await call( + "find_cells", + project="SAL", + number_min=10, + number_max=15, + kind="raw", + ) + + out = drive(steps) + assert out["found"] == 1 + assert out["remote"] is True + assert out["cells"][0]["path"] == uri + assert out["needs_metadata"] == ["mass", "nominal_capacity"] + + +def test_load_cell_allows_configured_remote_uri(drive, monkeypatch): + from cellpy import config + + monkeypatch.setattr(config.paths, "rawdatadir", "scp://host/raw") + seen = {} + + class Fake: + cell_name = "demo" + mass = 2.1 + nominal_capacity = 320 + + class data: + class summary: + columns = [] + + def get_cycle_numbers(self): + return [1] + + def fake_get(**kwargs): + seen.update(kwargs) + return Fake() + + monkeypatch.setattr("cellpy.get", fake_get) + + async def steps(call): + return await call( + "load_cell", + path="scp://host/raw/20240922_SAL12.res", + mass_mg=2.1, + nominal_capacity=320, + ) + + out = drive(steps) + assert seen["filename"] == "scp://host/raw/20240922_SAL12.res" + assert seen["mass"] == 2.1 + assert seen["nominal_capacity"] == 320 + assert out["nominal_capacity_was_supplied"] is True + + +def test_load_cell_refuses_unconfigured_remote_uri(drive, monkeypatch): + from cellpy import config + + monkeypatch.setattr(config.paths, "rawdatadir", "scp://host/raw") + + async def steps(call): + return await call("load_cell", path="scp://evil/loot.res") + + assert "configured remote" in drive(steps)["refused"] + + +def test_load_cell_refuses_remote_parent_escape(drive, monkeypatch): + from cellpy import config + + monkeypatch.setattr(config.paths, "rawdatadir", "scp://host/raw") + + async def steps(call): + return await call("load_cell", path="scp://host/raw/../secret.res") + + assert "configured remote" in drive(steps)["refused"] + + def test_find_cells_rejects_unknown_kind(drive): async def steps(call): return await call( diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py index 5b53324..dfe5f20 100644 --- a/tests/test_sandbox.py +++ b/tests/test_sandbox.py @@ -14,7 +14,13 @@ import pytest -from cellpy_mcp.sandbox import Refused, Sandbox, default_roots +from cellpy_mcp.sandbox import ( + Refused, + Sandbox, + _uri_under_prefix, + configured_remote_root, + default_roots, +) pytestmark = pytest.mark.essential @@ -140,6 +146,22 @@ class Paths: assert default_roots() == [local.resolve()] +def test_uri_under_prefix_rejects_siblings_and_dotdot(): + assert _uri_under_prefix("scp://host/raw/a.res", "scp://host/raw") + assert not _uri_under_prefix("scp://host/raw-other/a.res", "scp://host/raw") + assert not _uri_under_prefix("scp://host/raw/../secret.res", "scp://host/raw") + + +def test_configured_remote_root_matches_rawdatadir(monkeypatch): + class Paths: + rawdatadir = "scp://host/raw" + cellpydatadir = "/local/cells" + + monkeypatch.setattr("cellpy.config.paths", Paths(), raising=False) + assert configured_remote_root("scp://host/raw/a.res") == "scp://host/raw" + assert configured_remote_root("scp://evil/a.res") is None + + def test_an_unconfigured_cellpy_still_gets_a_narrow_root(monkeypatch): """Never "/": a server that starts wide open ships wide open."""