diff --git a/CHANGELOG.md b/CHANGELOG.md index 54bf3dc..ff9b9f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,26 @@ uses Semantic Versioning for public releases. ## Unreleased ### Added +- `transfer_ls`, `transfer_put`, `transfer_get` and `transfer_status` expose + data movement as tools, in the deferred pool; the core surface stays at 33. + Three verbs rather than one `transfer(op=...)` dispatcher, because a model + picks better from schemas that name their own arguments and an upload is not + the same authorization decision as a download. `transfer_status` is the + fourth because the other two return as soon as Globus accepts the task, and + without it the returned `task_id` names something nothing can read back. They + register only when some endpoint declares a `globus_transfer` block: a tool + whose only possible answer is "not configured" is still a tool the model can + call, so an install that moves no files shows no sign of them, and the + namespace-coverage check excuses exactly those four names rather than + whatever happens to be unregistered. A refused path comes back as a result + with its reason rather than a traceback, and nothing is submitted before the + paths are checked, so a rejected request costs no network and leaves no + half-made task. `validate_hpc_setup` (behind `doctor`) gains a transfer + check: it passes when nothing is configured, since transfers are opt-in the + way HPC is, and fails on a configured transfer that cannot work -- no write + root, no SDK, no consent, or a write root the collection will not list. The + write root is readable as well as writable, which the doctor probe found the + hard way: somewhere you may put a file is somewhere you may look at one. - Files can now move between this machine and an HPC collection. Compute has always run there and nothing could get a file there or back: a mesh had to be staged by hand before a remote tool could see it, and a subset or export a diff --git a/config.yaml.example b/config.yaml.example index efd1778..16b2d8d 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -39,6 +39,32 @@ hpc: path_prefixes: - "/lcrc/" - "/home//" + # Optional: where this endpoint's files live, in Globus Transfer's terms. + # Omit the block entirely and the endpoint runs code but moves no files -- + # the transfer tools are not registered at all. A Compute endpoint UUID + # says nothing about which collection serves that filesystem, so the + # collection is named explicitly rather than inferred. + globus_transfer: + # The Globus collection that serves this cluster's filesystem. + remote_collection_id: "00000000-0000-0000-0000-000000000000" + # This machine's collection (Globus Connect Personal). Without it + # nothing here is one end of a transfer. + local_collection_id: "00000000-0000-0000-0000-000000000000" + # Uploads must resolve inside this. It is a boundary, not a default + # directory: a path that escapes it is refused, not relocated. + remote_write_root: "/lcrc/group/e3sm/" + # Optional: widens where downloads may read from without widening + # where uploads may write. The write root stays readable either way. + remote_read_root: "/lcrc/group/e3sm" + # Optional: a collection that exposes a subtree calls that subtree "/". + # Longest match wins; a path under none of these is passed to Globus + # unchanged, because a wrong translation names a real file nobody + # asked for while an unrecognized one is merely rejected. + collection_roots: + - "/lcrc/group/e3sm" + # Optional: bounds where downloads may land on this machine, checked + # after symlinks are resolved. + local_root: "/Users//Downloads" # ---- Legacy single-endpoint form (still supported) ---- # Keep ``hpc.endpoints`` empty and set this if you only ever submit to one diff --git a/src/uxarray_mcp/registry.py b/src/uxarray_mcp/registry.py index 1679df5..ecafedf 100644 --- a/src/uxarray_mcp/registry.py +++ b/src/uxarray_mcp/registry.py @@ -138,6 +138,35 @@ "hpc": ("check_remote_yac",), } +# Deferred too, but only when some endpoint declares a ``globus_transfer`` +# block. An install that moves no files should show no sign of these: a tool +# whose only possible answer is "not configured" is still a tool the model can +# call, and calling it is the wrong lesson to have taught it. +_CONDITIONAL_TOOLS: dict[str, tuple[str, ...]] = { + "transfer": ( + "transfer_ls", + "transfer_put", + "transfer_get", + "transfer_status", + ), +} + +_CONDITIONAL_NAMES: frozenset[str] = frozenset( + name for names in _CONDITIONAL_TOOLS.values() for name in names +) + + +def _transfers_are_configured() -> bool: + """Whether any endpoint says where its files live. + + Wrapped rather than imported at module scope so building a registry never + depends on config being readable, and so a test can decide the answer + without writing a config file. + """ + from uxarray_mcp.tools.transfer_tools import transfers_are_configured + + return transfers_are_configured() + # --------------------------------------------------------------------------- # Prompt-as-tool helpers (formerly @mcp.prompt() decorators) @@ -638,6 +667,10 @@ def _apply_output_schema(tool: object, raw_name: str) -> None: _SEARCH_HINTS: dict[str, str] = { "check_remote_yac": "yac native remap conservative interpolation worker library build smoke test hpc", + "transfer_ls": "list remote directory globus collection files hpc browse", + "transfer_put": "upload stage copy file to hpc cluster globus transfer send", + "transfer_get": "download fetch retrieve file from hpc cluster globus transfer", + "transfer_status": "transfer task progress bytes globus poll", "calculate_curl": "vorticity rotation circulation wind curl cross product compute vector field zeta", "calculate_divergence": "compression expansion source sink wind divergence", "calculate_gradient": "spatial derivative slope field gradient", @@ -760,6 +793,18 @@ def build_registry( search_hint=_SEARCH_HINTS.get(raw, ""), ) registered.add(raw) + if _transfers_are_configured(): + for ns, raw in _flatten(_CONDITIONAL_TOOLS): + func = getattr(_tools_mod, raw) + registry.register(func, namespace=ns) + qualified = f"{ns}{sep}{raw}" + _apply_tags(registry, qualified, raw, func) + registry.update_tool_metadata( + qualified, + defer=True, + search_hint=_SEARCH_HINTS.get(raw, ""), + ) + registered.add(raw) registry.enable_tool_discovery() # ``enable_tool_discovery`` registers ``discover_tools`` itself, so it @@ -795,7 +840,11 @@ def _verify_coverage(registered: set[str], profile: Profile) -> None: f"Bridge tried to register non-public tools: {sorted(bogus)}" ) return - missing = public - registered + # The conditional tools are public so they can be imported and tested, but + # absent from an unconfigured registry on purpose. Excusing them here is + # narrower than excusing whatever happens not to be registered: anything + # else missing is still the loud failure this check exists to be. + missing = public - registered - _CONDITIONAL_NAMES if missing: raise RuntimeError( f"Namespace plan out of date — {len(missing)} public tools " diff --git a/src/uxarray_mcp/remote/transfer.py b/src/uxarray_mcp/remote/transfer.py index 3afeb02..550c3fa 100644 --- a/src/uxarray_mcp/remote/transfer.py +++ b/src/uxarray_mcp/remote/transfer.py @@ -276,9 +276,22 @@ def _write_root(self) -> str: ) return root - def _read_root(self) -> str: - """Reads fall back to the write root, never to the whole filesystem.""" - return self.profile.remote_read_root or self._write_root() + def _read_roots(self) -> tuple[str, ...]: + """Where reads may come from, in the order a relative path resolves. + + The write root is always readable: somewhere you may put a file is + somewhere you may look at one, and a config that could write to + `/scratch` but not list it would fail on the first download of + something it had just uploaded. With no read root configured, reads + fall back to the write root -- never to the whole filesystem. + """ + write_root = self.profile.remote_write_root + read_root = self.profile.remote_read_root + if not read_root: + return (self._write_root(),) + if write_root and write_root != read_root: + return (read_root, write_root) + return (read_root,) def remote_write_path(self, path: str) -> str: return to_collection_path( @@ -286,9 +299,18 @@ def remote_write_path(self, path: str) -> str: ) def remote_read_path(self, path: str) -> str: - return to_collection_path( - join_under(self._read_root(), path), self.profile.collection_roots - ) + roots = self._read_roots() + last: PathOutsideRoot | None = None + for root in roots: + try: + resolved = join_under(root, path) + except PathOutsideRoot as exc: + last = exc + continue + return to_collection_path(resolved, self.profile.collection_roots) + raise PathOutsideRoot( + f"{path!r} is outside every readable root ({', '.join(roots)})." + ) from last def _local_collection_id(self) -> str: if not self.profile.local_collection_id: diff --git a/src/uxarray_mcp/tools/__init__.py b/src/uxarray_mcp/tools/__init__.py index d1c6857..ee2d124 100644 --- a/src/uxarray_mcp/tools/__init__.py +++ b/src/uxarray_mcp/tools/__init__.py @@ -68,6 +68,12 @@ resume_workflow, run_workflow, ) +from .transfer_tools import ( + transfer_get, + transfer_ls, + transfer_put, + transfer_status, +) from .vector_calc import ( calculate_azimuthal_mean, calculate_curl, @@ -135,4 +141,8 @@ "manage_session", "get_status", "get_result", + "transfer_ls", + "transfer_put", + "transfer_get", + "transfer_status", ] diff --git a/src/uxarray_mcp/tools/execution_control.py b/src/uxarray_mcp/tools/execution_control.py index b717a33..f781792 100644 --- a/src/uxarray_mcp/tools/execution_control.py +++ b/src/uxarray_mcp/tools/execution_control.py @@ -62,6 +62,110 @@ def _make_check( return result +def _transfer_check( + base_config: Any, endpoint: str | None, run_probe: bool +) -> Dict[str, Any]: + """Report whether this endpoint can move data, not just run code. + + Passing when nothing is configured is deliberate: transfers are opt-in the + way HPC itself is, and a doctor that goes red for a feature the user never + asked for teaches people to ignore it. What is worth failing on is a + configured transfer that cannot work -- SDK absent, no consent, or a write + root the collection will not show. + + The reachability probe rides on ``run_remote_probe`` because it is a real + network call, and it lists the write root rather than transferring + anything: a listing proves the collection, the consent and the path all + line up, and moves no bytes. + """ + try: + profile = base_config.resolve_endpoint(endpoint=endpoint) + except Exception: + profile = None + transfer_profile = getattr(profile, "globus_transfer", None) + if transfer_profile is None: + return _make_check( + "transfer", + True, + "No globus_transfer block configured; this endpoint moves no files.", + details={"configured": False}, + guidance=( + "Add hpc.endpoints..globus_transfer with " + "remote_collection_id and remote_write_root to enable " + "transfer_put / transfer_get." + ), + ) + + details: Dict[str, Any] = { + "configured": True, + "endpoint_name": getattr(profile, "name", None), + "remote_collection_id": transfer_profile.remote_collection_id, + "local_collection_id": transfer_profile.local_collection_id, + "remote_write_root": transfer_profile.remote_write_root, + "remote_read_root": transfer_profile.remote_read_root, + "collection_roots": list(transfer_profile.collection_roots), + } + + if not transfer_profile.remote_write_root: + return _make_check( + "transfer", + False, + "globus_transfer is configured without a remote_write_root, so " + "uploads have nowhere they are allowed to land.", + details=details, + guidance="Set remote_write_root on this endpoint's globus_transfer block.", + ) + + from uxarray_mcp.remote.transfer import TransferService + + service = TransferService(transfer_profile) + try: + service.client # noqa: B018 -- builds the client, checks login state + except Exception as exc: + return _make_check( + "transfer", + False, + "Globus Transfer is configured but no client could be built.", + details={**details, **_exception_details(exc)}, + guidance=( + "Install the transfer extra (`uv sync --extra transfer`) and " + "complete the Globus login in a terminal; an MCP server cannot " + "open a browser consent flow." + ), + ) + + if not run_probe: + return _make_check( + "transfer", + True, + "Globus Transfer client is authenticated; collection reachability " + "not probed.", + details=details, + ) + + try: + entries = service.ls(transfer_profile.remote_write_root) + except Exception as exc: + return _make_check( + "transfer", + False, + f"Write root {transfer_profile.remote_write_root!r} could not be " + f"listed on the collection.", + details={**details, **_exception_details(exc)}, + guidance=( + "Check that remote_collection_id serves this filesystem, that " + "collection_roots translate the path the way the collection " + "names it, and that the write root exists." + ), + ) + return _make_check( + "transfer", + True, + f"Collection reachable; write root lists {len(entries)} entries.", + details={**details, "entry_count": len(entries)}, + ) + + def _guidance_for_error(message: str) -> str | None: """Return targeted next-step guidance for common HPC setup failures.""" lowered = message.lower() @@ -662,6 +766,8 @@ def validate_hpc_setup( ) ) + checks.append(_transfer_check(base_config, endpoint, run_remote_probe)) + passed = all(check["passed"] for check in checks) result = { "passed": passed, diff --git a/src/uxarray_mcp/tools/transfer_tools.py b/src/uxarray_mcp/tools/transfer_tools.py new file mode 100644 index 0000000..c21bbbc --- /dev/null +++ b/src/uxarray_mcp/tools/transfer_tools.py @@ -0,0 +1,305 @@ +"""Move files between this machine and an HPC collection, as MCP tools. + +Three verbs rather than one ``transfer(op=...)`` dispatcher: a model picks +better from three schemas that each name their own arguments than from one +that takes an operation string and a bag of maybe-required fields, and an +upload and a download are not the same authorization decision. + +They are registered only when some endpoint actually carries a +``globus_transfer`` block. An install that moves no files shows no sign of +these -- an unconfigured tool that exists only to explain that it is +unconfigured is a tool the model can still call, and calling it is the wrong +thing to have learned. + +Nothing here submits without checking paths first: the service refuses out of +root before it fetches a submission id, so a rejected request costs no network +and leaves no half-made task on the Globus side. +""" + +from __future__ import annotations + +from typing import Any, Dict + +from uxarray_mcp.provenance import attach_provenance +from uxarray_mcp.remote.transfer import ( + TransferError, + TransferService, +) +from uxarray_mcp.state import OperationTracker + +__all__ = [ + "transfer_get", + "transfer_ls", + "transfer_put", + "transfer_status", + "transfers_are_configured", +] + + +def _config(): + from uxarray_mcp.tools.execution_control import _load_config_for_tools + + return _load_config_for_tools() + + +def transfers_are_configured() -> bool: + """Whether any configured endpoint declares where its files live. + + Read at registry build time to decide whether the transfer tools exist at + all, so it answers ``False`` for every failure -- an unreadable or absent + config is not a reason to raise on startup for a feature the user has not + asked for. + """ + try: + config, _ = _config() + except Exception: + return False + return any( + getattr(profile, "globus_transfer", None) is not None + for profile in getattr(config, "endpoints", {}).values() + ) + + +def _service_for(endpoint: str | None) -> tuple[TransferService, str]: + """Resolve an endpoint to a service, or say what the config is missing.""" + config, _ = _config() + profile = config.resolve_endpoint(endpoint=endpoint) + if profile is None: + configured = ", ".join(config.endpoint_names) or "none" + raise TransferError( + f"No endpoint resolved for transfers. Configured endpoints: " + f"{configured}. Pass endpoint='name'." + ) + transfer_profile = getattr(profile, "globus_transfer", None) + if transfer_profile is None: + raise TransferError( + f"Endpoint {profile.name!r} has no globus_transfer block in " + f"config.yaml, so it moves no files. Add remote_collection_id and " + f"remote_write_root to enable transfers for it." + ) + return TransferService(transfer_profile), profile.name + + +def _failure( + tool: str, inputs: Dict[str, Any], tracker: OperationTracker, exc: Exception +) -> Dict[str, Any]: + """Report a refusal as a result, not a traceback. + + A path that failed containment is an answer -- the caller asked to move a + file somewhere it may not go -- so it comes back shaped like every other + result, with the reason in it. + """ + result = attach_provenance( + { + "submitted": False, + "reason": type(exc).__name__, + "message": str(exc), + }, + tool=tool, + inputs=inputs, + venue="local", + ) + result["_provenance"]["operation_id"] = tracker.operation_id + tracker.fail(str(exc)) + return result + + +def transfer_ls( + remote_path: str = "/", + endpoint: str | None = None, + session_id: str | None = None, +) -> Dict[str, Any]: + """List a directory on the HPC collection. + + ``remote_path`` is interpreted under the endpoint's read root, so a + relative path stays inside the configured tree and an absolute one outside + it is refused rather than listed. + + Parameters + ---------- + remote_path : str + Directory to list, relative to the endpoint's read root or absolute + within it. + endpoint : str | None + Configured endpoint name. Defaults to the resolved default endpoint. + session_id : str | None + Session to track this operation under. + """ + tracker = OperationTracker("transfer_ls", session_id=session_id) + inputs = { + "remote_path": remote_path, + "endpoint": endpoint, + "session_id": session_id, + } + try: + service, name = _service_for(endpoint) + tracker.stage("listing", f"Listing {remote_path!r} on {name}.") + entries = service.ls(remote_path) + except Exception as exc: + return _failure("transfer_ls", inputs, tracker, exc) + result = attach_provenance( + { + "endpoint": name, + "remote_path": remote_path, + "entry_count": len(entries), + "entries": entries, + }, + tool="transfer_ls", + inputs=inputs, + venue="globus_transfer", + ) + result["_provenance"]["operation_id"] = tracker.operation_id + tracker.succeed(f"Listed {len(entries)} entries.") + return result + + +def transfer_put( + local_path: str, + remote_path: str, + endpoint: str | None = None, + label: str | None = None, + session_id: str | None = None, +) -> Dict[str, Any]: + """Upload a local file or directory to the HPC collection. + + The destination must resolve inside the endpoint's ``remote_write_root``; + a read root does not widen where an upload may land. Directories transfer + recursively. The call returns as soon as Globus accepts the task -- poll + ``transfer_status`` with the returned ``task_id`` for completion. + + Parameters + ---------- + local_path : str + File or directory on this machine. + remote_path : str + Destination under the endpoint's write root. + endpoint : str | None + Configured endpoint name. + label : str | None + Label shown in the Globus web app for this task. + session_id : str | None + Session to track this operation under. + """ + tracker = OperationTracker("transfer_put", session_id=session_id) + inputs = { + "local_path": local_path, + "remote_path": remote_path, + "endpoint": endpoint, + "label": label, + "session_id": session_id, + } + try: + service, name = _service_for(endpoint) + plan = service.plan_put(local_path, remote_path, label=label) + tracker.stage("submitted", f"Uploading to {plan.destination_path} on {name}.") + submitted = service.submit(plan) + except Exception as exc: + return _failure("transfer_put", inputs, tracker, exc) + result = attach_provenance( + {"submitted": True, "endpoint": name, "direction": "upload", **submitted}, + tool="transfer_put", + inputs=inputs, + venue="globus_transfer", + ) + result["_provenance"]["operation_id"] = tracker.operation_id + tracker.succeed(f"Submitted upload task {submitted.get('task_id')}.") + return result + + +def transfer_get( + remote_path: str, + local_path: str, + recursive: bool = False, + endpoint: str | None = None, + label: str | None = None, + session_id: str | None = None, +) -> Dict[str, Any]: + """Download a file or directory from the HPC collection. + + The source is read under the endpoint's read root, which falls back to the + write root when unset -- never to the whole remote filesystem. A + ``local_root`` in the endpoint's config, if set, bounds where the download + may land on this machine, checked after symlinks are resolved. + + Parameters + ---------- + remote_path : str + Source under the endpoint's read root. + local_path : str + Destination on this machine. + recursive : bool + Set for a directory. A remote directory cannot be detected from here + without a listing, so this is explicit rather than guessed. + endpoint : str | None + Configured endpoint name. + label : str | None + Label shown in the Globus web app for this task. + session_id : str | None + Session to track this operation under. + """ + tracker = OperationTracker("transfer_get", session_id=session_id) + inputs = { + "remote_path": remote_path, + "local_path": local_path, + "recursive": recursive, + "endpoint": endpoint, + "label": label, + "session_id": session_id, + } + try: + service, name = _service_for(endpoint) + plan = service.plan_get( + remote_path, local_path, recursive=recursive, label=label + ) + tracker.stage("submitted", f"Downloading {plan.source_path} from {name}.") + submitted = service.submit(plan) + except Exception as exc: + return _failure("transfer_get", inputs, tracker, exc) + result = attach_provenance( + {"submitted": True, "endpoint": name, "direction": "download", **submitted}, + tool="transfer_get", + inputs=inputs, + venue="globus_transfer", + ) + result["_provenance"]["operation_id"] = tracker.operation_id + tracker.succeed(f"Submitted download task {submitted.get('task_id')}.") + return result + + +def transfer_status( + task_id: str, + endpoint: str | None = None, + session_id: str | None = None, +) -> Dict[str, Any]: + """Report what a submitted transfer task has done so far. + + ``transfer_put`` and ``transfer_get`` return as soon as Globus accepts the + task, so without this the returned ``task_id`` names something nothing can + read back. + + Parameters + ---------- + task_id : str + Task id returned by ``transfer_put`` or ``transfer_get``. + endpoint : str | None + Configured endpoint name whose credentials own the task. + session_id : str | None + Session to track this operation under. + """ + tracker = OperationTracker("transfer_status", session_id=session_id) + inputs = {"task_id": task_id, "endpoint": endpoint, "session_id": session_id} + try: + service, name = _service_for(endpoint) + tracker.stage("polling", f"Reading task {task_id} on {name}.") + status = service.status(task_id) + except Exception as exc: + return _failure("transfer_status", inputs, tracker, exc) + result = attach_provenance( + {"endpoint": name, **status}, + tool="transfer_status", + inputs=inputs, + venue="globus_transfer", + ) + result["_provenance"]["operation_id"] = tracker.operation_id + tracker.succeed(f"Task {task_id} is {status.get('status')}.") + return result diff --git a/tests/test_globus_transfer.py b/tests/test_globus_transfer.py index e20b43c..2d87c75 100644 --- a/tests/test_globus_transfer.py +++ b/tests/test_globus_transfer.py @@ -253,6 +253,20 @@ def test_a_download_reads_under_the_read_root(self, tmp_path): plan = service.plan_get("cases/out.nc", tmp_path / "out.nc") assert plan.source_path == "/lcrc/group/e3sm/cases/out.nc" + def test_the_write_root_is_readable_too(self, tmp_path): + # Somewhere you may put a file is somewhere you may look at one: a + # download of what was just uploaded must not fail on the read root. + service = TransferService(_profile(), FakeTransferClient()) + plan = service.plan_get("/scratch/rjain/out.nc", tmp_path / "out.nc") + assert plan.source_path == "/scratch/rjain/out.nc" + + def test_a_path_under_no_readable_root_names_them_all(self, tmp_path): + service = TransferService(_profile(), FakeTransferClient()) + with pytest.raises(PathOutsideRoot) as excinfo: + service.plan_get("/home/rjain/out.nc", tmp_path / "out.nc") + assert "/scratch/rjain" in str(excinfo.value) + assert "/lcrc/group/e3sm" in str(excinfo.value) + def test_a_read_root_does_not_widen_writes(self, tmp_path): source = tmp_path / "mesh.nc" source.write_text("x") diff --git a/tests/test_server.py b/tests/test_server.py index 29713e7..60ed462 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -15,6 +15,7 @@ from uxarray_mcp.app import make_mcp_server, make_registry from uxarray_mcp.registry import ( + _CONDITIONAL_TOOLS, _CONTROL_TOOLS, _CORE_EXTRA_TOOLS, _DEFERRED_TOOLS, @@ -31,6 +32,18 @@ EXPECTED_DEFERRED = 33 # +zonal_anomaly, +remap_to_rectilinear, +check_remote_yac +@pytest.fixture(autouse=True) +def _no_transfer_tools(monkeypatch): + """Pin the conditional transfer tools off for the count assertions. + + They register only when an endpoint declares a ``globus_transfer`` block, + which is a property of whoever is running the tests, not of the code. The + counts below are about the tool plan; ``test_transfer_tools.py`` is where + the conditional registration is exercised in both states. + """ + monkeypatch.setattr("uxarray_mcp.registry._transfers_are_configured", lambda: False) + + # --------------------------------------------------------------------------- # Coverage invariants # --------------------------------------------------------------------------- @@ -57,8 +70,9 @@ def test_namespace_plan_covers_every_public_tool(): control = {n for v in _CONTROL_TOOLS.values() for n in v} core_extra = {n for v in _CORE_EXTRA_TOOLS.values() for n in v} deferred = {n for v in _DEFERRED_TOOLS.values() for n in v} + conditional = {n for v in _CONDITIONAL_TOOLS.values() for n in v} - covered = FRONTDOOR_NAMES | control | core_extra | deferred + covered = FRONTDOOR_NAMES | control | core_extra | deferred | conditional missing = set(tools_mod.__all__) - covered assert not missing, f"uncovered public tools: {sorted(missing)}" diff --git a/tests/test_transfer_tools.py b/tests/test_transfer_tools.py new file mode 100644 index 0000000..c9f10c3 --- /dev/null +++ b/tests/test_transfer_tools.py @@ -0,0 +1,264 @@ +"""The transfer tools: conditional registration, and what they return. + +The path arithmetic itself is covered in ``test_globus_transfer.py``. What is +checked here is the layer above it: that an install with no +``globus_transfer`` block shows no transfer tools at all, that a refused path +comes back as a result rather than a traceback, and that `doctor` reports on +data movement without going red for a feature nobody configured. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from uxarray_mcp.registry import _CONDITIONAL_NAMES, build_registry +from uxarray_mcp.remote.config import EndpointProfile, GlobusTransferProfile, HPCConfig +from uxarray_mcp.tools import transfer_tools +from uxarray_mcp.tools.execution_control import _transfer_check + + +def _transfer_profile(**overrides) -> GlobusTransferProfile: + base = { + "remote_collection_id": "remote-uuid", + "local_collection_id": "local-uuid", + "remote_write_root": "/scratch/rjain", + "remote_read_root": "/lcrc/group/e3sm", + } + base.update(overrides) + return GlobusTransferProfile(**base) + + +def _config(transfer: GlobusTransferProfile | None) -> HPCConfig: + return HPCConfig( + endpoints={ + "chrysalis": EndpointProfile( + name="chrysalis", + endpoint_id="11111111-2222-3333-4444-555555555555", + globus_transfer=transfer, + ) + }, + default_endpoint="chrysalis", + ) + + +class FakeService: + def __init__(self, entries=None, error=None): + self.entries = entries or [] + self.error = error + self.submitted = [] + + def _raise(self): + if self.error: + raise self.error + + def ls(self, remote_path): + self._raise() + return self.entries + + def plan_put(self, local_path, remote_path, label=None): + self._raise() + return SimpleNamespace( + kind="put", + source_path=str(local_path), + destination_path=remote_path, + ) + + def plan_get(self, remote_path, local_path, recursive=False, label=None): + self._raise() + return SimpleNamespace( + kind="get", + source_path=remote_path, + destination_path=str(local_path), + ) + + def submit(self, plan): + self.submitted.append(plan) + return {"task_id": "task-1", "status": "Accepted"} + + def status(self, task_id): + self._raise() + return {"task_id": task_id, "status": "SUCCEEDED"} + + +@pytest.fixture +def wired(monkeypatch): + """Point the tools at a fake service and a configured endpoint.""" + service = FakeService() + monkeypatch.setattr( + transfer_tools, "_service_for", lambda endpoint: (service, "chrysalis") + ) + return service + + +class TestAnUnconfiguredInstallShowsNoTransferTools: + def test_they_are_absent_when_no_endpoint_declares_a_block(self, monkeypatch): + monkeypatch.setattr( + "uxarray_mcp.registry._transfers_are_configured", lambda: False + ) + tools = build_registry(profile="deferred-full").list_tools() + assert not [name for name in tools if "transfer_" in name] + + def test_they_appear_when_one_does(self, monkeypatch): + monkeypatch.setattr( + "uxarray_mcp.registry._transfers_are_configured", lambda: True + ) + tools = build_registry(profile="deferred-full").list_tools() + for raw in _CONDITIONAL_NAMES: + assert any(name.endswith(raw) for name in tools), raw + + def test_they_are_deferred_not_visible(self, monkeypatch): + monkeypatch.setattr( + "uxarray_mcp.registry._transfers_are_configured", lambda: True + ) + status = build_registry(profile="deferred-full").get_tools_status() + transfer_rows = [s for s in status if "transfer_" in s["name"]] + assert transfer_rows + assert all(row["defer"] for row in transfer_rows) + + def test_the_coverage_check_still_fires_for_anything_else(self, monkeypatch): + # The exemption is for the conditional names only; a genuinely + # unregistered tool must still be a loud failure. + monkeypatch.setattr( + "uxarray_mcp.registry._transfers_are_configured", lambda: False + ) + monkeypatch.setattr( + "uxarray_mcp.registry._DEFERRED_TOOLS", + {"inspect": ("inspect_mesh",)}, + ) + with pytest.raises(RuntimeError, match="Namespace plan out of date"): + build_registry(profile="deferred-full") + + def test_configuration_is_read_from_the_endpoints(self, monkeypatch): + monkeypatch.setattr( + transfer_tools, "_config", lambda: (_config(_transfer_profile()), None) + ) + assert transfer_tools.transfers_are_configured() is True + monkeypatch.setattr(transfer_tools, "_config", lambda: (_config(None), None)) + assert transfer_tools.transfers_are_configured() is False + + def test_an_unreadable_config_answers_no_rather_than_raising(self, monkeypatch): + def boom(): + raise OSError("no config here") + + monkeypatch.setattr(transfer_tools, "_config", boom) + assert transfer_tools.transfers_are_configured() is False + + +class TestARefusalComesBackAsAResult: + def test_a_path_outside_the_root_is_reported_not_raised(self, monkeypatch): + from uxarray_mcp.remote.transfer import PathOutsideRoot + + service = FakeService(error=PathOutsideRoot("/etc/passwd is outside /scratch")) + monkeypatch.setattr( + transfer_tools, "_service_for", lambda endpoint: (service, "chrysalis") + ) + result = transfer_tools.transfer_put("mesh.nc", "/etc/passwd") + assert result["submitted"] is False + assert result["reason"] == "PathOutsideRoot" + assert "outside" in result["message"] + assert service.submitted == [] + + def test_an_endpoint_without_a_block_says_what_to_add(self, monkeypatch): + monkeypatch.setattr(transfer_tools, "_config", lambda: (_config(None), None)) + result = transfer_tools.transfer_ls("/scratch/rjain") + assert result["submitted"] is False + assert "globus_transfer" in result["message"] + + def test_every_result_carries_provenance_and_an_operation_id(self, wired): + result = transfer_tools.transfer_ls("/scratch/rjain") + assert result["_provenance"]["operation_id"] + assert result["_provenance"]["tool"] == "transfer_ls" + + +class TestTheToolsReportWhatTheyDid: + def test_ls_counts_its_entries(self, monkeypatch): + service = FakeService(entries=[{"name": "out.nc"}, {"name": "run"}]) + monkeypatch.setattr( + transfer_tools, "_service_for", lambda endpoint: (service, "chrysalis") + ) + result = transfer_tools.transfer_ls("cases") + assert result["entry_count"] == 2 + assert result["endpoint"] == "chrysalis" + + def test_put_reports_the_direction_and_task(self, wired): + result = transfer_tools.transfer_put("mesh.nc", "runs/mesh.nc") + assert result["submitted"] is True + assert result["direction"] == "upload" + assert result["task_id"] == "task-1" + + def test_get_reports_the_other_direction(self, wired): + result = transfer_tools.transfer_get("cases/out.nc", "out.nc") + assert result["direction"] == "download" + assert wired.submitted[0].kind == "get" + + def test_status_reads_a_submitted_task_back(self, wired): + result = transfer_tools.transfer_status("task-1") + assert result["status"] == "SUCCEEDED" + + +class TestDoctorReportsOnDataMovement: + def test_nothing_configured_passes_and_explains(self): + check = _transfer_check(_config(None), None, False) + assert check["passed"] is True + assert check["details"]["configured"] is False + assert "globus_transfer" in check["guidance"] + + def test_a_block_without_a_write_root_fails(self): + profile = _transfer_profile(remote_write_root=None) + check = _transfer_check(_config(profile), None, False) + assert check["passed"] is False + assert "remote_write_root" in check["summary"] + + def test_a_client_that_cannot_be_built_fails_with_the_install_step( + self, monkeypatch + ): + from uxarray_mcp.remote import transfer as transfer_mod + + def boom(profile): + raise transfer_mod.TransferError("globus-sdk is not installed") + + monkeypatch.setattr(transfer_mod, "default_transfer_client", boom) + check = _transfer_check(_config(_transfer_profile()), None, False) + assert check["passed"] is False + assert "transfer" in check["guidance"] + + def test_without_the_probe_it_stops_at_authentication(self, monkeypatch): + from uxarray_mcp.remote import transfer as transfer_mod + + monkeypatch.setattr( + transfer_mod, "default_transfer_client", lambda profile: object() + ) + check = _transfer_check(_config(_transfer_profile()), None, False) + assert check["passed"] is True + assert "not probed" in check["summary"] + + def test_the_probe_lists_the_write_root(self, monkeypatch): + from uxarray_mcp.remote import transfer as transfer_mod + + class Client: + def operation_ls(self, collection_id, **kwargs): + assert kwargs["path"] == "/scratch/rjain" + return {"DATA": [{"name": "runs", "type": "dir"}]} + + monkeypatch.setattr( + transfer_mod, "default_transfer_client", lambda profile: Client() + ) + check = _transfer_check(_config(_transfer_profile()), None, True) + assert check["passed"] is True + assert check["details"]["entry_count"] == 1 + + def test_a_write_root_the_collection_will_not_show_fails(self, monkeypatch): + from uxarray_mcp.remote import transfer as transfer_mod + + class Client: + def operation_ls(self, collection_id, **kwargs): + raise RuntimeError("ClientError.404.NotFound") + + monkeypatch.setattr( + transfer_mod, "default_transfer_client", lambda profile: Client() + ) + check = _transfer_check(_config(_transfer_profile()), None, True) + assert check["passed"] is False + assert "collection_roots" in check["guidance"]