Minimize update parser parity diff.

Inline legacy update parsing in _update_rbac for source-of-truth behavior and keep old/new parity assertions behind KUBEPLUS_UPDATE_EQ_CHECK, while removing extra helper methods.

Made-with: Cursor

Author: anniegracehu

provider-kubeconfig.py

@@ -85,34 +85,6 @@ def _parse_permission_rules(self, perms):
                                 rule_list.append(rule_group)
                 return rule_list, resources
 
-        def _parse_permission_rules_old(self, perms):
-                """Legacy update parser path kept for parity verification."""
-                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"
@@ -259,12 +231,6 @@ def _assert_all_resources_parity(self, label, old_resources, new_resources):
                         f"Only in new: {new_only}"
                     )
 
-        def _assert_update_parser_parity(self, perms):
-                old_rule_list, old_resources = self._parse_permission_rules_old(perms)
-                new_rule_list, new_resources = self._parse_permission_rules(perms)
-                self._assert_rule_parity("update-parser", old_rule_list, new_rule_list)
-                self._assert_all_resources_parity("update-parser", old_resources, new_resources)
-
         def _build_consumer_rules_old(self):
                 # Read all resources
                 ruleGroup1 = {}
@@ -685,10 +651,35 @@ 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)
-                old_rule_list, old_resources = self._parse_permission_rules_old(perms)
+                # Keep old update parser path inline as source of truth.
+                old_rule_list = []
+                old_resources = []
+                for api_group, res_actions in perms.items():
+                    for res in res_actions:
+                        for resource, verbs in res.items():
+                            if resource not in old_resources:
+                                old_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]
+                            old_rule_list.append(rule_group)
+                new_rule_list, new_resources = self._parse_permission_rules(perms)
                 if os.getenv("KUBEPLUS_UPDATE_EQ_CHECK", "0") == "1":
-                    self._assert_update_parser_parity(perms)
-                # Keep old update parser path as source of truth.
+                    self._assert_rule_parity("update-parser", old_rule_list, new_rule_list)
+                    self._assert_all_resources_parity("update-parser", old_resources, new_resources)
                 rule_list = old_rule_list
                 new_resources = old_resources
 

tests/test_provider_kubeconfig.py

@@ -143,7 +143,30 @@ def test_update_old_and_new_parsers_match(self):
                 {"nonResourceURL::/metrics": ["get"]},
             ],
         }
-        old_rules, old_resources = self.generator._parse_permission_rules_old(perms)
+        old_rules = []
+        old_resources = []
+        for api_group, res_actions in perms.items():
+            for res in res_actions:
+                for resource, verbs in res.items():
+                    if resource not in old_resources:
+                        old_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]
+                    old_rules.append(rule_group)
         new_rules, new_resources = self.generator._parse_permission_rules(perms)
         self.generator._assert_rule_parity("test-update-parser", old_rules, new_rules)
         self.generator._assert_all_resources_parity("test-update-parser", old_resources, new_resources)