handle duplicate keys

examples/host_vars/rtr001.yml

@@ -1,9 +1,10 @@
 ---
-asn:
+bad_asn:
   name: admin
   password: admin
 
 asn: "4294967296"
+asn: "4294967297"
 
 interfaces:
   - name: eth0

examples/schema.yaml

@@ -3,4 +3,7 @@ type: object
 properties:
   asn:
     type: string
-    asn: True
+    asn: True
+  bad_asn:
+    type: string
+    asn: True

src/helpers.py

@@ -4,17 +4,47 @@
 import yaml
 from rich import print as rprint  # noqa
 
-# Load YAML or JSON
+current_file = None  # Global variable to hold the current file being processed
+
+
+class UniqueKeyLoader(yaml.SafeLoader):
+    warnings = []
+
+    @classmethod
+    def add_warning(cls, warning_msg, key, value, filename):
+        cls.warnings.append({
+            'warning': True,
+            'msg': warning_msg,
+            'key': key,
+            'filename': str(filename)
+        })
 
+    def construct_mapping(self, node, deep=False):
+        mapping = set()
+        for key_node, value_node in node.value:
+            key = self.construct_object(key_node, deep=deep)
+            if key in mapping:
+                warning_msg = f"Duplicate keys '{key}' found in YAML file."
+                UniqueKeyLoader.add_warning(warning_msg, key, self.construct_object(value_node), current_file)
+            mapping.add(key)
+        return super().construct_mapping(node, deep)
 
+
+# Load YAML or JSON
 def load_yaml_or_json(filename):
+    global current_file
+    current_file = filename
+
     try:
         if filename.suffix in [".yaml", ".yml"]:
             with filename.open() as f:
-                return yaml.safe_load(f)
+                UniqueKeyLoader.warnings = []  # Reset warnings before loading
+                data = yaml.load(f, Loader=UniqueKeyLoader)
+                warnings = UniqueKeyLoader.warnings
+                return data, warnings
         elif filename.suffix == ".json":
             with filename.open() as f:
-                return json.load(f)
+                return json.load(f), []
     except FileNotFoundError:
         rprint(f"[ERROR] File not found: {filename}")
         sys.exit(1)

src/main.py

@@ -16,23 +16,25 @@ def __init__(self, document_path, schema, validator_engine):
         self._document_path = Path(document_path)
         self._schema = Path(schema)
         self._errors = []
+        self._warnings = []
 
     def initialize(self):
         self._load_schema()
         self._validator.initialize(self._schema)
 
     def _load_schema(self):
-        self._schema = load_yaml_or_json(self._schema)
+        self._schema, warning_msgs = load_yaml_or_json(self._schema)
 
     def _validate(self):
         for filename in self._document_path.iterdir():
             if filename.suffix in [".yaml", ".yml", ".json"]:
-                data = load_yaml_or_json(filename)
+                data, file_warnings = load_yaml_or_json(filename)
                 file_errors = self._validator.results(data)
                 for error in file_errors:
                     error["filename"] = str(filename)  # Add filename key to each error
                 self._errors.extend(file_errors)
-        return self._errors
+                self._warnings.extend(file_warnings)
+        return {"errors": self._errors, "warnings": self._warnings}
 
     @property
     def results(self):

src/plugins/json_schema/validator.py

@@ -85,15 +85,15 @@ def _validate(self, data):
         self._errors = []
 
         if self._validator.is_valid(data):
-            self._errors.append({"error": False, "msg": None, "value": None})
+            self._errors.append({"error": False, "msg": None, "key": None})
         else:
             try:
                 for error in self._validator.iter_errors(data):
                     self._errors.append(
                         {
                             "error": True,
                             "msg": error.message,
-                            "value": ", ".join([prop for prop in error.path])
+                            "key": ", ".join([prop for prop in error.path])
                             if error.path
                             else None,
                         }
@@ -103,15 +103,15 @@ def _validate(self, data):
                     {
                         "error": True,
                         "msg": f"Unresolved reference error: {str(e)}",
-                        "value": None,
+                        "key": None,
                     }
                 )
             except Exception as e:
                 self._errors.append(
                     {
                         "error": True,
                         "msg": f"Unknown error: {str(e)}",
-                        "value": None,
+                        "key": None,
                     }
                 )
         return self._errors