add stat command (#963)

* add stat command

* add additional error handling tests

* fix stat type hints

Co-authored-by: Stephen Rosen <[email protected]>

* require sdk v3.39.0

---------

Co-authored-by: Stephen Rosen <[email protected]>

changelog.d/20240304_132149_aaschaer.md

@@ -0,0 +1,3 @@
+### Enhancements
+
+* Add `globus stat` command for getting the status of a path on a collection

src/globus_cli/commands/__init__.py

@@ -23,6 +23,7 @@
         "rm": ("rm", "rm_command"),
         "search": ("search", "search_command"),
         "session": ("session", "session_command"),
+        "stat": ("stat", "stat_command"),
         "task": ("task", "task_command"),
         "timer": ("timer", "timer_command"),
         "transfer": ("transfer", "transfer_command"),

src/globus_cli/commands/stat.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import uuid
+
+import click
+import globus_sdk
+
+from globus_cli.login_manager import LoginManager
+from globus_cli.parsing import ENDPOINT_PLUS_REQPATH, command, local_user_option
+from globus_cli.termio import Field, TextMode, display
+
+STAT_FIELDS = [
+    Field("Name", "name"),
+    Field("Type", "type"),
+    Field("Last Modified", "last_modified"),
+    Field("Size", "size"),
+    Field("Permissions", "permissions"),
+    Field("User", "user"),
+    Field("Group", "group"),
+]
+
+
+@command(
+    "stat",
+    short_help="Get the status of a path",
+    adoc_examples=r"""Get the status of a path on a collection.
+
+[source,bash]
+----
+$ col_id=6c54cade-bde5-45c1-bdea-f4bd71dba2cc
+$ globus stat $col_id:/home/share/godata/file1.txt
+----
+
+""",
+)
+@click.argument("endpoint_plus_path", type=ENDPOINT_PLUS_REQPATH)
+@local_user_option
+@LoginManager.requires_login("transfer")
+def stat_command(
+    login_manager: LoginManager,
+    *,
+    endpoint_plus_path: tuple[uuid.UUID, str],
+    local_user: str | None,
+) -> None:
+    """
+    Get the status of a path on a collection.
+    """
+    transfer_client = login_manager.get_transfer_client()
+    endpoint_id, path = endpoint_plus_path
+
+    try:
+        res = transfer_client.operation_stat(endpoint_id, path, local_user=local_user)
+        display(
+            res,
+            text_mode=TextMode.text_record,
+            fields=STAT_FIELDS,
+        )
+
+    except globus_sdk.TransferAPIError as error:
+        if error.code == "NotFound":
+            display(error.raw_json, simple_text=f"Nothing found at {path}")
+        else:
+            raise

tests/functional/test_stat.py

@@ -0,0 +1,57 @@
+import urllib.parse
+
+import globus_sdk
+from globus_sdk._testing import get_last_request, load_response
+
+
+def test_stat(run_line):
+    """
+    Make a stat with the --local-user option, confirm output is rendered and query
+    parameters are passed as expected
+    """
+    meta = load_response(globus_sdk.TransferClient.operation_stat).metadata
+    endpoint_id = meta["endpoint_id"]
+
+    result = run_line(f"globus stat {endpoint_id}:foo/ --local-user bar")
+    expected = """Name:          file1.txt
+Type:          file
+Last Modified: 2023-12-18 16:52:50+00:00
+Size:          4
+Permissions:   0644
+User:          tutorial
+Group:         tutorial
+"""
+    assert result.output == expected
+
+    req = get_last_request()
+    parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
+    assert parsed_qs == {"path": ["foo/"], "local_user": ["bar"]}
+
+
+def test_stat_not_found(run_line):
+    """
+    operation_stat returns a NotFound error, confirm non-error output
+    """
+    meta = load_response(
+        globus_sdk.TransferClient.operation_stat, case="not_found"
+    ).metadata
+    endpoint_id = meta["endpoint_id"]
+
+    result = run_line(f"globus stat {endpoint_id}:foo/")
+
+    assert result.output == "Nothing found at foo/\n"
+
+
+def test_stat_permission_denied(run_line):
+    """
+    operation_stat hits a permission denied error, confirm error output
+    """
+    meta = load_response(
+        globus_sdk.TransferClient.operation_stat, case="permission_denied"
+    ).metadata
+    endpoint_id = meta["endpoint_id"]
+
+    result = run_line(f"globus stat {endpoint_id}:foo/", assert_exit_code=1)
+
+    assert "A Transfer API Error Occurred." in result.stderr
+    assert "EndpointPermissionDenied" in result.stderr