feat: Make resource lookup case-insensitive to match Kubernetes API behavior

kubernetes/base/dynamic/discovery.py

@@ -195,6 +195,27 @@ def get_resources_for_api_version(self, prefix, group, version, preferred):
             resources[resource_list.kind].append(resource_list)
         return resources
 
+    def _find_resource_case_insensitive(self, kind, resources_dict):
+        """
+        Find a resource in the resources_dict where the key matches the specified kind,
+        regardless of case.
+
+        Args:
+            kind: The kind to search for (case-insensitive)
+            resources_dict: The resources dictionary to search in
+
+        Returns:
+            The actual key if found, None otherwise
+        """
+        if not kind:
+            return None
+
+        kind_lower = kind.lower()
+        for key in resources_dict.keys():
+            if key.lower() == kind_lower:
+                return key
+        return None
+
     def get(self, **kwargs):
         """ Same as search, but will throw an error if there are multiple or no
             results. If there are multiple results and only one is an exact match
@@ -241,14 +262,69 @@ def api_groups(self):
         return self.parse_api_groups(request_resources=False, update=True)['apis'].keys()
 
     def search(self, **kwargs):
+        # Save the original kind parameter for case-insensitive lookup if needed
+        original_kind = kwargs.get('kind')
+
         # In first call, ignore ResourceNotFoundError and set default value for results
         try:
             results = self.__search(self.__build_search(**kwargs), self.__resources, [])
         except ResourceNotFoundError:
             results = []
+
+        # If no results were found and a kind was specified, try case-insensitive lookup
+        if not results and original_kind and kwargs.get('kind') == original_kind:
+            # Iterate through the resource tree to find a case-insensitive match
+            for prefix, groups in self.__resources.items():
+                for group, versions in groups.items():
+                    for version, rg in versions.items():
+                        if hasattr(rg, "resources") and rg.resources:
+                            # Look for a matching kind (case-insensitive)
+                            matching_kind = self._find_resource_case_insensitive(original_kind, rg.resources)
+                            if matching_kind:
+                                # Try again with the correct case
+                                modified_kwargs = kwargs.copy()
+                                modified_kwargs['kind'] = matching_kind
+                                try:
+                                    results = self.__search(self.__build_search(**modified_kwargs), self.__resources, [])
+                                    if results:
+                                        break
+                                except ResourceNotFoundError:
+                                    continue
+
+        # If still no results, invalidate cache and retry
         if not results:
             self.invalidate_cache()
-            results = self.__search(self.__build_search(**kwargs), self.__resources, [])
+
+            # Reset kind parameter that might have been modified
+            if original_kind:
+                kwargs['kind'] = original_kind
+
+            # Try exact match first
+            try:
+                results = self.__search(self.__build_search(**kwargs), self.__resources, [])
+            except ResourceNotFoundError:
+                # If exact match fails, try case-insensitive lookup
+                if original_kind:
+                    # Same case-insensitive lookup logic as above
+                    for prefix, groups in self.__resources.items():
+                        for group, versions in groups.items():
+                            for version, rg in versions.items():
+                                if hasattr(rg, "resources") and rg.resources:
+                                    matching_kind = self._find_resource_case_insensitive(original_kind, rg.resources)
+                                    if matching_kind:
+                                        modified_kwargs = kwargs.copy()
+                                        modified_kwargs['kind'] = matching_kind
+                                        try:
+                                            results = self.__search(self.__build_search(**modified_kwargs), self.__resources, [])
+                                            if results:
+                                                break
+                                        except ResourceNotFoundError:
+                                            continue
+
+                # If still no results, set empty list
+                if not results:
+                    results = []
+
         self.__maybe_write_cache()
         return results
 
@@ -349,10 +425,54 @@ def search(self, **kwargs):
 
             The arbitrary arguments can be any valid attribute for an Resource object
         """
+        # Save original kind parameter for case-insensitive lookup if needed
+        original_kind = kwargs.get('kind')
+
+        # Try original search first
         results = self.__search(self.__build_search(**kwargs), self.__resources)
+
+        # If no results were found and a kind was specified, try case-insensitive lookup
+        if not results and original_kind and kwargs.get('kind') == original_kind:
+            # Iterate through the resource tree to find a case-insensitive match
+            for prefix, groups in self.__resources.items():
+                for group, versions in groups.items():
+                    for version, resource_dict in versions.items():
+                        if isinstance(resource_dict, ResourceGroup) and resource_dict.resources:
+                            # Look for a matching kind (case-insensitive)
+                            matching_kind = self._find_resource_case_insensitive(original_kind, resource_dict.resources)
+                            if matching_kind:
+                                # Try again with the correct case
+                                modified_kwargs = kwargs.copy()
+                                modified_kwargs['kind'] = matching_kind
+                                results = self.__search(self.__build_search(**modified_kwargs), self.__resources)
+                                if results:
+                                    break
+
+        # If still no results, invalidate cache and retry
         if not results:
             self.invalidate_cache()
+
+            # Reset kind parameter that might have been modified
+            if original_kind:
+                kwargs['kind'] = original_kind
+
+            # Try exact match first
             results = self.__search(self.__build_search(**kwargs), self.__resources)
+
+            # If exact match fails, try case-insensitive lookup
+            if not results and original_kind:
+                for prefix, groups in self.__resources.items():
+                    for group, versions in groups.items():
+                        for version, resource_dict in versions.items():
+                            if isinstance(resource_dict, ResourceGroup) and resource_dict.resources:
+                                matching_kind = self._find_resource_case_insensitive(original_kind, resource_dict.resources)
+                                if matching_kind:
+                                    modified_kwargs = kwargs.copy()
+                                    modified_kwargs['kind'] = matching_kind
+                                    results = self.__search(self.__build_search(**modified_kwargs), self.__resources)
+                                    if results:
+                                        break
+
         return results
 
     def __build_search(self, prefix=None, group=None, api_version=None, kind=None, **kwargs):

kubernetes/test/test_case_insensitive_discovery.py

@@ -0,0 +1,123 @@
+# Copyright 2023 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Test for case insensitive resource lookup in the Dynamic Client.
+This test addresses issue #2402: Resource lookup is case-sensitive while it shouldn't
+"""
+
+import unittest
+import kubernetes.config as config
+from kubernetes import client, dynamic
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
+import os
+import sys
+import logging
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class TestCaseInsensitiveDiscovery(unittest.TestCase):
+    """
+    Test case for case-insensitive resource lookup in the Dynamic Client.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        """
+        Set up test class - load kubernetes configuration
+        """
+        try:
+            config.load_kube_config()
+            cls.api_client = client.ApiClient()
+            cls.dynamic_client = dynamic.DynamicClient(cls.api_client)
+        except Exception as e:
+            logger.warning(f"Could not load kubernetes configuration: {e}")
+            cls.skipTest(cls, f"Failed to load kubernetes configuration: {e}")
+
+    def test_case_sensitivity_service(self):
+        """
+        Test that Service resource can be found regardless of case
+        """
+        # 1. Test with exact case
+        try:
+            resource = self.dynamic_client.resources.get(kind='Service')
+            self.assertEqual(resource.kind, 'Service')
+            logger.info("Successfully found resource with correct case: Service")
+        except Exception as e:
+            self.fail(f"Failed to get resource with correct case 'Service': {e}")
+
+        # 2. Test with lowercase
+        try:
+            resource = self.dynamic_client.resources.get(kind='service')
+            self.assertEqual(resource.kind, 'Service')
+            logger.info("Successfully found resource with lowercase: service")
+        except Exception as e:
+            self.fail(f"Failed to get resource with lowercase 'service': {e}")
+
+        # 3. Test with mixed case
+        try:
+            resource = self.dynamic_client.resources.get(kind='SerVicE')
+            self.assertEqual(resource.kind, 'Service')
+            logger.info("Successfully found resource with mixed case: SerVicE")
+        except Exception as e:
+            self.fail(f"Failed to get resource with mixed case 'SerVicE': {e}")
+
+    def test_case_sensitivity_deployment(self):
+        """
+        Test that Deployment resource can be found regardless of case
+        """
+        # 1. Test with exact case
+        try:
+            resource = self.dynamic_client.resources.get(kind='Deployment')
+            self.assertEqual(resource.kind, 'Deployment')
+            logger.info("Successfully found resource with correct case: Deployment")
+        except Exception as e:
+            self.fail(f"Failed to get resource with correct case 'Deployment': {e}")
+
+        # 2. Test with lowercase
+        try:
+            resource = self.dynamic_client.resources.get(kind='deployment')
+            self.assertEqual(resource.kind, 'Deployment')
+            logger.info("Successfully found resource with lowercase: deployment")
+        except Exception as e:
+            self.fail(f"Failed to get resource with lowercase 'deployment': {e}")
+
+    def test_nonexistent_resource(self):
+        """
+        Test that looking up a non-existent resource still returns the appropriate error
+        """
+        with self.assertRaises(ResourceNotFoundError):
+            self.dynamic_client.resources.get(kind='NonExistentResource')
+            logger.info("Correctly raised ResourceNotFoundError for non-existent resource")
+
+    def test_with_api_version(self):
+        """
+        Test case insensitive lookup with api_version specified
+        """
+        try:
+            resource = self.dynamic_client.resources.get(
+                api_version='apps/v1', kind='deployment')
+            self.assertEqual(resource.kind, 'Deployment')
+            self.assertEqual(resource.group, 'apps')
+            self.assertEqual(resource.api_version, 'v1')
+            logger.info("Successfully found resource with api_version and lowercase kind")
+        except Exception as e:
+            self.fail(f"Failed to get resource with api_version and lowercase kind: {e}")
+
+
+if __name__ == '__main__':
+    unittest.main()