Adding Get Tunnel Event support

changelog.d/20260212_121439_john_tunnel_events.rst

@@ -0,0 +1,24 @@
+..
+.. A new scriv changelog fragment
+..
+.. Uncomment the header that is right (remove the leading dots).
+..
+.. Leave the "(:pr:`...`)" text in your change description.
+.. GitHub Actions will automatically replace it when the PR is merged.
+..
+.. Python Support
+.. --------------
+..
+.. - A bullet item for the Python Support category. (:pr:`NUMBER`)
+..
+.. Breaking Changes
+.. ----------------
+..
+.. - A bullet item for the Breaking Changes category. (:pr:`NUMBER`)
+..
+.. Added
+.. -----
+..
+.. - A bullet item for the Added category. (:pr:`NUMBER`)
+
+ - ``TransferClient.get_tunnel_events()`` fetches the events associated with a tunnel

src/globus_sdk/services/transfer/client.py

@@ -2891,3 +2891,32 @@ def get_stream_access_point(
             f"/v2/stream_access_points/{stream_ap_id}", query_params=query_params
         )
         return r
+
+    def get_tunnel_events(
+        self,
+        tunnel_id: str | uuid.UUID,
+        *,
+        query_params: dict[str, t.Any] | None = None,
+    ) -> response.GlobusHTTPResponse:
+        """
+        :param tunnel_id: The ID of the Tunnel which we are fetching events about.
+        :param query_params: Any additional parameters will be passed through
+            as query params.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    result = tc.get_tunnel_events(tunnel_id)
+                    print(result["data"])
+
+            .. tab-item:: API Info
+
+                ``GET /v2/tunnels/<tunnel_id>/events``
+        """
+        log.debug("TransferClient.get_tunnel_events(...)")
+        r = self.get(f"/v2/tunnels/{tunnel_id}/events", query_params=query_params)
+        return r

src/globus_sdk/testing/data/transfer/get_tunnel_events.py

@@ -0,0 +1,58 @@
+import uuid
+
+from globus_sdk.testing.models import RegisteredResponse, ResponseSet
+
+TUNNEL_ID = str(uuid.uuid4())
+
+GET_TUNNEL_EVENTS_DOC = {
+    "data": [
+        {
+            "type": "TunnelEvent",
+            "id": 2768,
+            "attributes": {
+                "code": "STARTED",
+                "is_error": False,
+                "description": "started",
+                "details": "Attempting tunnel establishment",
+                "time": "2026-02-12T21:59:01.857473",
+            },
+        },
+        {
+            "type": "TunnelEvent",
+            "id": 2769,
+            "attributes": {
+                "code": "TUNNEL_ACTIVE",
+                "is_error": False,
+                "description": "tunnel is active",
+                "details": "Tunnel has been established",
+                "time": "2026-02-12T21:59:02.876253",
+            },
+        },
+        {
+            "type": "TunnelEvent",
+            "id": 2777,
+            "attributes": {
+                "code": "TUNNEL_STOPPED",
+                "is_error": False,
+                "description": "tunnel has been stopped",
+                "details": "Tunnel stopped as requested.",
+                "time": "2026-02-12T22:12:03.655877",
+            },
+        },
+    ],
+    "links": None,
+    "meta": {"request_id": "655TZe5vm"},
+}
+
+
+RESPONSES = ResponseSet(
+    metadata={
+        "tunnel_id": TUNNEL_ID,
+    },
+    default=RegisteredResponse(
+        service="transfer",
+        path=f"/v2/tunnels/{TUNNEL_ID}/events",
+        json=GET_TUNNEL_EVENTS_DOC,
+        method="GET",
+    ),
+)

tests/functional/services/transfer/test_get_tunnel_events.py

@@ -0,0 +1,12 @@
+from globus_sdk.testing import get_last_request, load_response
+
+
+def test_get_tunnel(client):
+    meta = load_response(client.get_tunnel_events).metadata
+    res = client.get_tunnel_events(meta["tunnel_id"])
+    assert res.http_status == 200
+    assert res["data"][0]["type"] == "TunnelEvent"
+    assert res["data"][1]["type"] == "TunnelEvent"
+
+    req = get_last_request()
+    assert req.body is None