fix(python): support Kubernetes call_api compatibility

.github/workflows/ci.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        kubernetes: ["28.1.0", "36.0.0"]
+        kubernetes: ["28.1.0", "36.0.3"]
 
     steps:
       - name: Checkout code

shifu-sdk-python/README.md

@@ -143,10 +143,12 @@ Use DeviceShifu class for managing multiple devices with isolated instances.
 
 ### Kubernetes Python Client Compatibility
 
-The SDK supports Kubernetes Python client versions 28.1.0 through 36.x. It
-automatically uses `response_type` with clients before v36 and
-`response_types_map` with v36. The dependency is capped below v37 so a future
-breaking client release cannot be installed silently before it is tested.
+The SDK supports Kubernetes Python client versions 28.1.0 through 36.x. CI
+tests the exact compatibility endpoints `kubernetes==28.1.0` (legacy) and
+`kubernetes==36.0.3` (latest stable). The SDK automatically uses
+`response_type` with clients before v36 and `response_types_map` with v36. The
+dependency is capped below v37 so a future breaking client release cannot be
+installed silently before it is tested.
 
 ### Environment Variables
 | Variable | Required | Default | Description |

shifu-sdk-python/src/shifu_sdk/core.py

@@ -68,10 +68,26 @@ class EdgeDevicePhase(Enum):
 
 def _response_type_kwargs(api_client: client.ApiClient) -> Dict[str, Any]:
     """Return response deserialization arguments for the installed client API."""
-    parameters = inspect.signature(api_client.call_api).parameters
+    try:
+        parameters = inspect.signature(api_client.call_api).parameters
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError(
+            "Unsupported Kubernetes ApiClient.call_api signature: expected "
+            "'response_type' (kubernetes 28.1.0) or 'response_types_map' "
+            "(kubernetes 36.0.3), but the signature could not be inspected"
+        ) from exc
+
     if "response_types_map" in parameters:
         return {"response_types_map": {200: "object", 401: None}}
-    return {"response_type": "object"}
+    if "response_type" in parameters:
+        return {"response_type": "object"}
+
+    raise RuntimeError(
+        "Unsupported Kubernetes ApiClient.call_api signature: expected "
+        "'response_type' (kubernetes 28.1.0) or 'response_types_map' "
+        "(kubernetes 36.0.3); found parameters: "
+        f"{', '.join(parameters)}"
+    )
 
 
 def init():

shifu-sdk-python/tests/test_kubernetes_client_compat.py

@@ -1,3 +1,4 @@
+import json
 import unittest
 from pathlib import Path
 
@@ -6,6 +7,12 @@
 from shifu_sdk import core
 
 
+EDGEDEVICE_PATH = (
+    "/apis/shifu.edgenesis.io/v1alpha1/namespaces/devices/"
+    "edgedevices/test-device"
+)
+
+
 class LegacyApiClient:
     def __init__(self):
         self.calls = []
@@ -27,6 +34,8 @@ def call_api(
                 "body": body,
                 "auth_settings": auth_settings,
                 "response_type": response_type,
+                "_return_http_data_only": _return_http_data_only,
+                "_preload_content": _preload_content,
             }
         )
         return {"metadata": {"name": "test-device"}}
@@ -46,16 +55,28 @@ def call_api(
         _return_http_data_only=None,
         _preload_content=None,
     ):
+        response_type = response_types_map.get(FakeResponse.status, None)
         self.calls.append(
             {
                 "resource_path": resource_path,
                 "method": method,
                 "body": body,
                 "auth_settings": auth_settings,
                 "response_types_map": response_types_map,
+                "selected_response_type": response_type,
+                "_return_http_data_only": _return_http_data_only,
+                "_preload_content": _preload_content,
             }
         )
-        return {"metadata": {"name": "test-device"}}
+        if response_type != "object":
+            raise AssertionError("HTTP 200 must deserialize as object")
+        return json.loads(FakeResponse.data)
+
+
+class UnsupportedApiClient:
+    @staticmethod
+    def call_api(resource_path, method):
+        raise AssertionError("unsupported call_api must not be called")
 
 
 class FakeResponse:
@@ -72,12 +93,15 @@ def getheaders():
 
 
 class FakeRestClient:
-    @staticmethod
-    def GET(*args, **kwargs):
+    def __init__(self):
+        self.calls = []
+
+    def GET(self, url, **kwargs):
+        self.calls.append(("GET", url, kwargs))
         return FakeResponse()
 
-    @staticmethod
-    def PUT(*args, **kwargs):
+    def PUT(self, url, **kwargs):
+        self.calls.append(("PUT", url, kwargs))
         return FakeResponse()
 
 
@@ -94,6 +118,13 @@ def tearDown(self):
         core.edgedevice_namespace = self.original_namespace
         core.edgedevice_name = self.original_name
 
+    def assert_common_call(self, call, method):
+        self.assertEqual(call["resource_path"], EDGEDEVICE_PATH)
+        self.assertEqual(call["method"], method)
+        self.assertEqual(call["auth_settings"], ["BearerToken"])
+        self.assertIs(call["_return_http_data_only"], True)
+        self.assertIs(call["_preload_content"], True)
+
     def test_get_uses_response_argument_supported_by_installed_client(self):
         for api_client, expected_key, expected_value in (
             (LegacyApiClient(), "response_type", "object"),
@@ -105,7 +136,13 @@ def test_get_uses_response_argument_supported_by_installed_client(self):
                 response = core._rest_get_edgedevice()
 
                 self.assertEqual(response["metadata"]["name"], "test-device")
-                self.assertEqual(api_client.calls[0][expected_key], expected_value)
+                self.assertEqual(len(api_client.calls), 1)
+                call = api_client.calls[0]
+                self.assert_common_call(call, "GET")
+                self.assertIsNone(call["body"])
+                self.assertEqual(call[expected_key], expected_value)
+                if isinstance(api_client, ModernApiClient):
+                    self.assertEqual(call["selected_response_type"], "object")
 
     def test_put_uses_response_argument_supported_by_installed_client(self):
         edge_device = {"metadata": {"name": "test-device"}}
@@ -118,8 +155,22 @@ def test_put_uses_response_argument_supported_by_installed_client(self):
 
                 core._rest_put_edgedevice(edge_device)
 
-                self.assertEqual(api_client.calls[0]["body"], edge_device)
-                self.assertEqual(api_client.calls[0][expected_key], expected_value)
+                self.assertEqual(len(api_client.calls), 1)
+                call = api_client.calls[0]
+                self.assert_common_call(call, "PUT")
+                self.assertIs(call["body"], edge_device)
+                self.assertEqual(call[expected_key], expected_value)
+                if isinstance(api_client, ModernApiClient):
+                    self.assertEqual(call["selected_response_type"], "object")
+
+    def test_unsupported_call_api_signature_raises_clear_error(self):
+        with self.assertRaisesRegex(
+            RuntimeError,
+            r"Unsupported Kubernetes ApiClient\.call_api signature: expected "
+            r"'response_type'.*'response_types_map'.*found parameters: "
+            r"resource_path, method",
+        ):
+            core._response_type_kwargs(UnsupportedApiClient())
 
     def test_dependency_range_covers_only_tested_client_majors(self):
         pyproject = Path(__file__).parents[1] / "pyproject.toml"
@@ -131,13 +182,20 @@ def test_dependency_range_covers_only_tested_client_majors(self):
 
     def test_installed_client_completes_get_and_put_calls(self):
         api_client = client.ApiClient()
-        api_client.rest_client = FakeRestClient()
+        rest_client = FakeRestClient()
+        api_client.rest_client = rest_client
         core.k8s_api_client = api_client
 
         response = core._rest_get_edgedevice()
         core._rest_put_edgedevice(response)
 
         self.assertEqual(response["metadata"]["name"], "test-device")
+        self.assertEqual(
+            [method for method, _, _ in rest_client.calls], ["GET", "PUT"]
+        )
+        self.assertTrue(
+            all(url.endswith(EDGEDEVICE_PATH) for _, url, _ in rest_client.calls)
+        )
 
 
 if __name__ == "__main__":