Use single _parse_permission_rules for update and revoke.

Remove duplicate _parse_update_rules_legacy and env parity self-check; add a focused test for non-apigroup edge case.

Made-with: Cursor

Author: anniegracehu

provider-kubeconfig.py

@@ -84,34 +84,6 @@ def _parse_permission_rules(self, perms):
                             rule_list.append(rule_group)
                 return rule_list, resources
 
-        def _parse_update_rules_legacy(self, perms):
-                """Legacy update permission parse; runtime source of truth for update."""
-                rule_list = []
-                resources = []
-                for api_group, res_actions in perms.items():
-                    for res in res_actions:
-                        for resource, verbs in res.items():
-                            if resource not in resources:
-                                resources.append(resource.strip())
-                            rule_group = {}
-                            if api_group == "non-apigroup":
-                                if "nonResourceURL" in resource:
-                                    parts = resource.split("nonResourceURL::")
-                                    non_res = parts[1].strip() if len(parts) > 1 else parts[0].strip()
-                                    rule_group["nonResourceURLs"] = [non_res]
-                                    rule_group["verbs"] = verbs
-                            else:
-                                rule_group["apiGroups"] = [api_group]
-                                rule_group["verbs"] = verbs
-                                if "resourceName" in resource:
-                                    parts = resource.split("/resourceName::")
-                                    rule_group["resources"] = [parts[0].strip()]
-                                    rule_group["resourceNames"] = [parts[1].strip()]
-                                else:
-                                    rule_group["resources"] = [resource]
-                            rule_list.append(rule_group)
-                return rule_list, resources
-
         def _read_perm_configmap_resources(self, sa, namespace, kubeconfig):
                 cfg_map_name = sa + "-perms"
                 cfg_map_filename = sa + "-perms.txt"
@@ -678,11 +650,7 @@ def _apply_provider_rbac(self, sa, namespace, kubeconfig):
         def _update_rbac(self, permissionfile, sa, namespace, kubeconfig):
                 """Add permissions from JSON/YAML file to provider (update command)."""
                 perms = self._load_permission_data(permissionfile)
-                rule_list, new_resources = self._parse_update_rules_legacy(perms)
-                if os.getenv("KUBEPLUS_UPDATE_EQ_CHECK", "0") == "1":
-                    pq_rules, pq_resources = self._parse_permission_rules(perms)
-                    self._assert_rule_parity("update-parser", rule_list, pq_rules)
-                    self._assert_all_resources_parity("update-parser", new_resources, pq_resources)
+                rule_list, new_resources = self._parse_permission_rules(perms)
 
                 role = {
                     "apiVersion": "rbac.authorization.k8s.io/v1",

tests/test_provider_kubeconfig.py

@@ -133,24 +133,13 @@ def test_load_permission_data_accepts_yaml(self):
         finally:
             os.remove(path)
 
-    def test_update_legacy_parser_matches_shared_parser(self):
-        """_parse_permission_rules must match legacy update parse (used by revoke vs update)."""
-        perms = {
-            "apps": [
-                {"deployments": ["get", "create"]},
-                {"deployments/resourceName::sample": ["get"]},
-            ],
-            "non-apigroup": [
-                {"nonResourceURL::/metrics": ["get"]},
-                {"invalid-without-nonResourceURL-marker": ["get"]},
-            ],
-        }
-        legacy_rules, legacy_resources = self.generator._parse_update_rules_legacy(perms)
-        shared_rules, shared_resources = self.generator._parse_permission_rules(perms)
-        self.generator._assert_rule_parity("update-parser", legacy_rules, shared_rules)
-        self.generator._assert_all_resources_parity(
-            "update-parser", legacy_resources, shared_resources
-        )
+    def test_parse_permission_rules_non_apigroup_without_marker_appends_empty_rule(self):
+        """non-apigroup entries without nonResourceURL still append a rule (legacy behavior)."""
+        perms = {"non-apigroup": [{"not-a-url": ["get"]}]}
+        rules, resources = self.generator._parse_permission_rules(perms)
+        self.assertEqual(len(rules), 1)
+        self.assertEqual(rules[0], {})
+        self.assertEqual(resources, ["not-a-url"])
 
 
 class TestKubeconfigIntegration(unittest.TestCase):