proxmox_firewall & proxmox module_utils

- Move check_rules() to proxmox module_utils and rename to compare_list_of_dicts()
- Generalize the implemnetation as this is usefull in multiple places.
- e.g. filtering out which fw rules, aliases, etc needs to be created/updated

Author: JanaHoch

plugins/module_utils/proxmox.py

@@ -73,6 +73,53 @@ def ansible_to_proxmox_bool(value):
     return 1 if value else 0
 
 
+def compare_list_of_dicts(existing_list, new_list, uid, params_to_ignore=None):
+    """ Compare 2 list of dicts
+    Use case - for firewall rules we will be getting a list of rules from user.
+    We want to filter out which rules needs to be updated and which rules are completely new and needs to be created
+
+    @param existing_list: Existing values example - list of existing rules
+    @param new_list: New values example - list of rules passed to module
+    @param uid: unique identifier in dict. It should always be present in both lists - in case of firewall rules it's pos
+    @param params_to_ignore:  list of params we want to ignore which are present in existing_list's dict.
+                            In case of firewall rules we want to ignore ['digest', 'ipversion']
+
+    @return: returns 2 list items 1st is the list of items which are completely new and needs to be created
+            2nd is a list of items which needs to be updated
+    """
+    if params_to_ignore is None:
+        params_to_ignore = list()
+    items_to_update = []
+    new_list = [{k: v for k, v in item.items() if v is not None} for item in new_list]
+
+    if existing_list is None:
+        items_to_create = new_list
+        items_to_update = list()
+        return items_to_create, items_to_update
+
+    existing_list = {x[uid]: x for x in existing_list}
+    new_list = {x[uid]: x for x in new_list}
+
+    common_uids = set(existing_list.keys()).intersection(set(new_list.keys()))
+    missing_uids = set(new_list.keys()) - set(existing_list.keys())
+    items_to_create = [new_list[id] for id in missing_uids]
+
+    for id in common_uids:
+        # If new rule has a parameter that is not present in existing rule we need to update
+        if set(new_list[id].keys()) - set(existing_list[id].keys()) != set():
+            items_to_update.append(new_list[id])
+            continue
+
+        # If existing rule param value doesn't match new rule param OR
+        # If existing rule has a param that is not present in new rule except for params in params_to_ignore
+        for existing_rule_param, existing_parm_value in existing_list[id].items():
+            if (existing_rule_param not in params_to_ignore and
+                    new_list[id].get(existing_rule_param) != existing_parm_value):
+                items_to_update.append(new_list[id])
+
+    return items_to_create, items_to_update
+
+
 class ProxmoxAnsible(object):
     """Base class for Proxmox modules"""
     TASK_TIMED_OUT = 'timeout expired'

plugins/modules/proxmox_firewall.py

@@ -399,6 +399,7 @@
 from ansible_collections.community.proxmox.plugins.module_utils.proxmox import (
     proxmox_auth_argument_spec,
     ansible_to_proxmox_bool,
+    compare_list_of_dicts,
     ProxmoxAnsible
 )
 
@@ -491,10 +492,11 @@ def run(self):
         group = self.params.get("group")
         group_conf = self.params.get("group_conf")
 
-        for rule in rules:
-            rule['icmp-type'] = rule.get('icmp_type')
-            rule['enable'] = ansible_to_proxmox_bool(rule.get('enable'))
-            del rule['icmp_type']
+        if rules is not None:
+            for rule in rules:
+                rule['icmp-type'] = rule.get('icmp_type')
+                rule['enable'] = ansible_to_proxmox_bool(rule.get('enable'))
+                del rule['icmp_type']
 
         if level == "vm":
             vm = self.get_vm(vmid=self.params.get('vmid'))
@@ -616,54 +618,25 @@ def delete_fw_rule(self, rules_obj, pos):
                 msg=f'Failed to delete firewall rule at pos {pos}: {e}'
             )
 
-    def check_rules(self, existing_rules, new_rules):
-        rules_to_update = []
-        new_rules = [{k: v for k, v in item.items() if v is not None} for item in new_rules]
-
-        if existing_rules is None:
-            rules_to_create = new_rules
-            rules_to_update = list()
-            return rules_to_create, rules_to_update
-
-        existing_rules = {x['pos']: x for x in existing_rules}
-        new_rules = {x['pos']: x for x in new_rules}
-
-        common_pos = set(existing_rules.keys()).intersection(set(new_rules.keys()))
-        pos_to_create = set(new_rules.keys()) - set(existing_rules.keys())
-        rules_to_create = [new_rules[pos] for pos in pos_to_create]
-
-        params_to_ignore = ['digest', 'ipversion']
-
-        for pos in common_pos:
-            # If new rule has a parameter that is not present in existing rule we need to update
-            if set(new_rules[pos].keys()) - set(existing_rules[pos].keys()) != set():
-                rules_to_update.append(new_rules[pos])
-                continue
-
-            # If existing rule param value doesn't match new rule param OR
-            # If existing rule has a param that is not present in new rule except for params in params_to_ignore
-            for existing_rule_param, existing_parm_value in existing_rules[pos].items():
-                if (existing_rule_param not in params_to_ignore and
-                        new_rules[pos].get(existing_rule_param) != existing_parm_value):
-                    rules_to_update.append(new_rules[pos])
-
-        return rules_to_create, rules_to_update
-
     def update_fw_rules(self, rules_obj, rules, force):
         existing_rules = self.get_fw_rules(rules_obj)
-        rules_to_create, rules_to_update = self.check_rules(existing_rules=existing_rules, new_rules=rules)
+        rules_to_create, rules_to_update = compare_list_of_dicts(
+            existing_list=existing_rules,
+            new_list=rules,
+            uid='pos',
+            params_to_ignore=['digest', 'ipversion']
+        )
 
-        if len(rules_to_update) == 0:
-            if len(rules_to_create) == 0:
-                self.module.exit_json(
-                    changed=False,
-                    msg='No need to update any FW rules.'
+        if len(rules_to_update) == 0 and len(rules_to_create) == 0:
+            self.module.exit_json(
+                changed=False,
+                msg='No need to update any FW rules.'
 
-                )
-            elif len(rules_to_create) > 0 and not force:
-                self.module.fail_json(
-                    msg=f"Need to create new rules for pos - {[x['pos'] for x in rules_to_create]} But force is false"
-                )
+            )
+        elif len(rules_to_create) > 0 and not force:
+            self.module.fail_json(
+                msg=f"Need to create new rules for pos - {[x['pos'] for x in rules_to_create]} But force is false"
+            )
 
         for rule in rules_to_update:
             try:
@@ -684,7 +657,12 @@ def update_fw_rules(self, rules_obj, rules, force):
 
     def create_fw_rules(self, rules_obj, rules, force):
         existing_rules = self.get_fw_rules(rules_obj=rules_obj)
-        rules_to_create, rules_to_update = self.check_rules(existing_rules=existing_rules, new_rules=rules)
+        rules_to_create, rules_to_update = compare_list_of_dicts(
+            existing_list=existing_rules,
+            new_list=rules,
+            uid='pos',
+            params_to_ignore=['digest', 'ipversion']
+        )
 
         if len(rules_to_create) == 0 and len(rules_to_update) == 0:
             self.module.exit_json(