Use safe_eval instead of temporary rules for comparison

plugins/modules/rule.py

@@ -74,7 +74,7 @@
                 description: Properties of the rule.
                 type: dict
             value_raw:
-                description: Rule values as exported from the UI.
+                description: Rule values as exported from the web interface.
                 type: str
     ruleset:
         description: Name of the ruleset to manage.
@@ -93,15 +93,6 @@
 notes:
     - "To achieve idempotency, this module is comparing the specified rule with the already existing
       rules based on conditions, folder, value_raw and enabled/disabled."
-    - "To be able to compare the value_raw, which is internally stored in python format, the module
-      has to do a workaround: it is creating the specified rule, and then compares this rule with
-      all existing rules."
-    - "Then, in case of I(state=absent), it will delete both rules: the specified one and the duplicate
-      found."
-    - "Or, in case of I(state=present), it will delete the new rule, if a duplicate is already there,
-      and keep it if not."
-    - "This obviously leads to more rules being added and removed as one might expect. That's also
-      visible in the pending changes and audit log."
 """
 
 EXAMPLES = r"""
@@ -224,6 +215,7 @@
 import json
 
 from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.common.validation import safe_eval
 from ansible.module_utils.urls import fetch_url
 
 try:
@@ -274,26 +266,39 @@ def get_existing_rule(module, base_url, headers, ruleset, rule):
     # Get rules in ruleset
     rules = get_rules_in_ruleset(module, base_url, headers, ruleset)
 
+    (value_mod, exc) = safe_eval(rule["value_raw"], include_exceptions=True)
+    if exc is not None:
+        exit_failed(module, "value_raw in rule has invalid format")
+
     if rules is not None:
         # Loop through all rules
         for r in rules.get("value"):
+            (value_api, exc) = safe_eval(
+                r["extensions"]["value_raw"], include_exceptions=True
+            )
+            if exc is not None:
+                exit_failed("Error deserializing value_raw from API")
             if (
-                r["id"] != rule["id"]
-                and r["extensions"]["conditions"] == rule["extensions"]["conditions"]
+                r["extensions"]["folder"] == rule["folder"]
+                and r["extensions"]["conditions"] == rule["conditions"]
                 and r["extensions"]["properties"]["disabled"]
-                == rule["extensions"]["properties"]["disabled"]
-                and r["extensions"]["folder"] == rule["extensions"]["folder"]
-                and r["extensions"]["value_raw"] == rule["extensions"]["value_raw"]
+                == rule["properties"]["disabled"]
+                and value_api == value_mod
             ):
                 # If they are the same, return the ID
                 return r
 
     return None
 
 
-def get_api_repr(module, base_url, headers, ruleset, rule):
+def create_rule(module, base_url, headers, ruleset, rule):
     api_endpoint = "/domain-types/rule/collections/all"
 
+    changed = True
+    e = get_existing_rule(module, base_url, headers, ruleset, rule)
+    if e:
+        return (e["id"], not changed)
+
     params = {
         "ruleset": ruleset,
         "folder": rule["folder"],
@@ -317,46 +322,17 @@ def get_api_repr(module, base_url, headers, ruleset, rule):
 
     r = json.loads(response.read().decode("utf-8"))
 
-    return r
-
-
-def create_rule(module, base_url, headers, ruleset, rule):
-
-    created = True
-
-    # get API representation of the rule
-    r = get_api_repr(module, base_url, headers, ruleset, rule)
-
-    # compare the API output to existing rules
-    e = get_existing_rule(module, base_url, headers, ruleset, r)
-
-    # if existing rule found, delete new rule and return existing id
-    if e:
-        delete_rule_by_id(module, base_url, headers, r["id"])
-        return (e["id"], not created)
-
-    # else return new rule id
-    return (r["id"], created)
+    return (r["id"], changed)
 
 
 def delete_rule(module, base_url, headers, ruleset, rule):
-
-    deleted = True
-
-    # get API representation of the rule
-    r = get_api_repr(module, base_url, headers, ruleset, rule)
-
-    # compare the API output to existing rules
-    e = get_existing_rule(module, base_url, headers, ruleset, r)
-
-    # if existing rule found, delete both
+    changed = True
+    e = get_existing_rule(module, base_url, headers, ruleset, rule)
     if e:
-        delete_rule_by_id(module, base_url, headers, r["id"])
         delete_rule_by_id(module, base_url, headers, e["id"])
-        return deleted
+        return changed
     else:
-        delete_rule_by_id(module, base_url, headers, r["id"])
-        return not deleted
+        return not changed
 
 
 def delete_rule_by_id(module, base_url, headers, rule_id):

tests/integration/targets/rule/vars/main.yml

@@ -41,3 +41,34 @@ checkmk_rules:
         "documentation_url": "https://github.com/tribe29/ansible-collection-tribe29.checkmk/blob/main/plugins/modules/rules.py"
       }
       value_raw: "{'levels': (80.0, 90.0)}"
+  - ruleset: "periodic_discovery"
+    rule:
+      location:
+          position: "top"
+          folder: "/"
+      properties:
+        comment: "Created by Ansible"
+        description: "Perform Service Discovery every 5 minutes"
+        disabled: false
+      conditions:
+        host_tags: []
+        service_labels: []
+        host_labels:
+          - key: "robotmk"
+            operator: "is"
+            value: "yes"
+      value_raw: "{
+        'check_interval': 5.0,
+        'inventory_rediscovery': {
+          'activation': True,
+          'excluded_time': [],
+          'group_time': 900,
+          'mode': 2,
+          'service_filters':(
+            'combined', {'service_whitelist': ['^E2E.*']}
+          )
+        },
+        'severity_new_host_label': 0,
+        'severity_unmonitored': 0,
+        'severity_vanished': 0
+      }"