Check for duplicate YAML keys

cosmosis/campaign.py

@@ -9,6 +9,39 @@
 import subprocess
 import contextlib
 
+class UniqueKeyLoader(yaml.SafeLoader):
+    """
+    This is a YAML loader that raises an error if there are duplicate keys.
+
+    It is based on the discussion here:
+    https://gist.github.com/pypt/94d747fe5180851196eb
+    """
+    def construct_mapping(self, node, deep=False):
+        mapping = set()
+        for key_node, _ in node.value:
+            key = self.construct_object(key_node, deep=deep)
+            if key in mapping:
+                raise ValueError(f"Duplicate {key} key found in YAML.")
+            mapping.add(key)
+        return super().construct_mapping(node, deep)
+
+def load_yaml(stream):
+    """
+    Load YAML markup from a stream, with a check for duplicate keys.
+
+    Parameters
+    ----------
+    stream : file
+        The open file-like object to load from
+
+    Returns
+    -------
+    data : object
+        The dict or list loaded from the YAML file
+    """
+    return yaml.load(stream, Loader=UniqueKeyLoader)
+
+
 def pipeline_after(params, after_module, modules):
     """
     Insert item(s) in the pipeline after the given module.
@@ -547,7 +580,7 @@ def parse_yaml_run_file(run_config):
         info = run_config
     else:
         with open(run_config, 'r') as f:
-            info = yaml.safe_load(f)
+            info = load_yaml(f)
 
     output_dir = info.get("output_dir", ".")
     output_name = info.get("output_name", "{name}")

cosmosis/test/bad-campaign.yml

@@ -0,0 +1,7 @@
+runs:
+  - name: broken-campaign
+    base: cosmosis/test/example.ini
+    params:
+      - pipeline.debug = T
+    params:
+      - pipeline.debug = F

cosmosis/test/test_campaign.py

@@ -2,8 +2,8 @@
 from ..runtime import Inifile
 import os
 import tempfile
-import yaml
 import contextlib
+import pytest
 
 @contextlib.contextmanager
 def run_from_source_dir():
@@ -104,7 +104,7 @@ def test_campaign_functions():
 def test_campaign_functions2():
     with run_from_source_dir():
         with open("cosmosis/test/campaign.yml") as f:
-            runs_config = yaml.safe_load(f)
+            runs_config = load_yaml(f)
 
         with tempfile.TemporaryDirectory() as dirname:
             runs_config['output_dir']  = dirname
@@ -145,3 +145,8 @@ def test_campaign_env():
         runs = parse_yaml_run_file("cosmosis/test/campaign.yml")
     assert runs["env-test-1"]["params"].get("test1", "env_test_var")  == "xxx"
     assert runs["env-test-2"]["params"].get("test1", "env_test_var")  == "yyy"
+
+def test_campaign_duplicate_keys():
+    with pytest.raises(ValueError):
+        with run_from_source_dir():
+            runs = parse_yaml_run_file("cosmosis/test/bad-campaign.yml")