Restore update parser parity checks.

Add _parse_update_rules_legacy for a single legacy implementation, optional KUBEPLUS_UPDATE_EQ_CHECK assertion against _parse_permission_rules in _update_rbac, and a unit test including non-apigroup edge cases.

Made-with: Cursor

Author: anniegracehu

provider-kubeconfig.py

@@ -84,6 +84,34 @@ 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"
@@ -650,30 +678,11 @@ 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 = []
-                for api_group, res_actions in perms.items():
-                    for res in res_actions:
-                        for resource, verbs in res.items():
-                            if resource not in new_resources:
-                                new_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)
+                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)
 
                 role = {
                     "apiVersion": "rbac.authorization.k8s.io/v1",

tests/test_provider_kubeconfig.py

@@ -133,6 +133,25 @@ 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
+        )
+
 
 class TestKubeconfigIntegration(unittest.TestCase):
     """