Add ComputeClientV2 version and AMQP info methods

changelog.d/20241125_154944_30907815+rjmello_compute_client_misc_methods.rst

@@ -0,0 +1,5 @@
+Added
+~~~~~
+
+- Added the ``ComputeClientV2.get_version()`` and ``ComputeClientV2.get_result_amqp_url()``
+  methods. (:pr:`NUMBER`)

src/globus_sdk/_testing/data/compute/v2/get_result_amqp_url.py

@@ -0,0 +1,15 @@
+from globus_sdk._testing.models import RegisteredResponse, ResponseSet
+
+DEFAULT_RESPONSE_DOC = {
+    "queue_prefix": "some_prefix",
+    "connection_url": "amqps://user:password@amqp.fqdn",
+}
+
+RESPONSES = ResponseSet(
+    default=RegisteredResponse(
+        service="compute",
+        path="/v2/get_amqp_result_connection_url",
+        method="GET",
+        json=DEFAULT_RESPONSE_DOC,
+    ),
+)

src/globus_sdk/_testing/data/compute/v2/get_version.py

@@ -0,0 +1,28 @@
+from responses.matchers import query_param_matcher
+
+from globus_sdk._testing.models import RegisteredResponse, ResponseSet
+
+API_VERSION = "1.23.0"
+ALL_RESPONSE_DOC = {
+    "api": API_VERSION,
+    "min_sdk_version": "1.0.0a6",
+    "min_endpoint_version": "1.0.0a0",
+    "git_sha": "80b2ef87bc546b3b386cf2e1d372f4be50f10bc4",
+}
+
+RESPONSES = ResponseSet(
+    metadata={"api_version": API_VERSION},
+    default=RegisteredResponse(
+        service="compute",
+        path="/v2/version",
+        method="GET",
+        json=API_VERSION,  # type: ignore[arg-type]
+    ),
+    all=RegisteredResponse(
+        service="compute",
+        path="/v2/version",
+        method="GET",
+        json=ALL_RESPONSE_DOC,
+        match=[query_param_matcher(params={"service": "all"})],
+    ),
+)

src/globus_sdk/services/compute/client.py

@@ -24,6 +24,36 @@ class ComputeClientV2(client.BaseClient):
     scopes = ComputeScopes
     default_scope_requirements = [Scope(ComputeScopes.all)]
 
+    def get_version(self, service: str | None = None) -> GlobusHTTPResponse:
+        """Get the current version of the API and other services.
+
+        :param service: Service for which to get version information.
+
+        .. tab-set::
+
+            .. tab-item:: API Info
+
+                .. extdoclink:: Get Version
+                    :service: compute
+                    :ref: Root/operation/get_version_v2_version_get
+        """
+        query_params = {"service": service} if service else None
+        return self.get("/v2/version", query_params=query_params)
+
+    def get_result_amqp_url(self) -> GlobusHTTPResponse:
+        """Generate new credentials (in the form of a connection URL) for
+        connecting to the AMQP service.
+
+        .. tab-set::
+
+            .. tab-item:: API Info
+
+                .. extdoclink:: Get Result AMQP URL
+                    :service: compute
+                    :ref: Root/operation/get_user_specific_result_amqp_url_v2_get_amqp_result_connection_url_get
+        """  # noqa: E501
+        return self.get("/v2/get_amqp_result_connection_url")
+
     def register_endpoint(self, data: dict[str, t.Any]) -> GlobusHTTPResponse:
         """Register a new endpoint.
 

tests/functional/services/compute/v2/test_get_result_amqp_url.py

@@ -0,0 +1,9 @@
+import globus_sdk
+from globus_sdk._testing import load_response
+
+
+def test_get_result_amqp_url(compute_client_v2: globus_sdk.ComputeClientV2):
+    load_response(compute_client_v2.get_result_amqp_url)
+    res = compute_client_v2.get_result_amqp_url()
+    assert res.http_status == 200
+    assert "connection_url" in res.data

tests/functional/services/compute/v2/test_get_version.py

@@ -0,0 +1,16 @@
+import globus_sdk
+from globus_sdk._testing import load_response
+
+
+def test_get_version(compute_client_v2: globus_sdk.ComputeClientV2):
+    meta = load_response(compute_client_v2.get_version).metadata
+    res = compute_client_v2.get_version()
+    assert res.http_status == 200
+    assert res.data == meta["api_version"]
+
+
+def test_get_version_all(compute_client_v2: globus_sdk.ComputeClientV2):
+    meta = load_response(compute_client_v2.get_version, case="all").metadata
+    res = compute_client_v2.get_version(service="all")
+    assert res.http_status == 200
+    assert res.data["api"] == meta["api_version"]