Merge branch 'fix/paramater-aggregation' of https://github.com/IBM/co…

…mpliance-trestle into fix/paramater-aggregation

tests/trestle/tasks/csv_to_oscal_cd_test.py

@@ -661,6 +661,32 @@ def test_execute_add_rule(tmp_path: pathlib.Path) -> None:
     assert len(component.control_implementations[1].set_parameters) == 1
 
 
+def test_execute_param_duplicate_value(tmp_path: pathlib.Path) -> None:
+    """Test execute param duplicate default value."""
+    _, section = _get_config_section_init(tmp_path, 'test-csv-to-oscal-cd-bp.config')
+    # duplicate default param default value
+    rows = _get_rows('tests/data/csv/bp.sample.v2.csv')
+    row = rows[3]
+    assert row[13] == 'allowed_admins_per_account'
+    assert row[15] == '10'
+    row = rows[2]
+    row[13] = 'allowed_admins_per_account'
+    row[15] = '10'
+    rows[2] = row
+    with mock.patch('trestle.tasks.csv_to_oscal_cd.csv.reader') as mock_csv_reader:
+        mock_csv_reader.return_value = rows
+        tgt = csv_to_oscal_cd.CsvToOscalComponentDefinition(section)
+        retval = tgt.execute()
+        assert retval == TaskOutcome.SUCCESS
+    row[15] = '11'
+    rows[2] = row
+    with mock.patch('trestle.tasks.csv_to_oscal_cd.csv.reader') as mock_csv_reader:
+        mock_csv_reader.return_value = rows
+        tgt = csv_to_oscal_cd.CsvToOscalComponentDefinition(section)
+        retval = tgt.execute()
+        assert retval == TaskOutcome.FAILURE
+
+
 def test_execute_missing_param_default_value(tmp_path: pathlib.Path) -> None:
     """Test execute missing param default_value."""
     _, section = _get_config_section_init(tmp_path, 'test-csv-to-oscal-cd-bp.config')

trestle/tasks/csv_to_oscal_cd.py

@@ -331,7 +331,7 @@ def _calculate_set_params(self, mod_rules: List) -> tuple:
                 logger.debug(f'params mod: {key}')
             else:
                 add_set_params.append(key)
-                logger.debug(f'prams add: {key}')
+                logger.debug(f'params add: {key}')
         return (del_set_params, add_set_params, mod_set_params)
 
     def _calculate_control_mappings(self, mod_rules: List) -> tuple:
@@ -824,7 +824,17 @@ class _OscalHelper():
     @staticmethod
     def add_set_parameter(set_parameter_list: List[SetParameter], set_parameter: SetParameter) -> None:
         """Add set parameter."""
-        set_parameter_list.append(set_parameter)
+        add = True
+        # don't add duplicate
+        for sp in set_parameter_list:
+            if sp.param_id == set_parameter.param_id:
+                add = False
+                if sp.values != set_parameter.values:
+                    text = f'set-parameter id={sp.param_id} conflicting values'
+                    raise RuntimeError(text)
+                break
+        if add:
+            set_parameter_list.append(set_parameter)
 
     @staticmethod
     def remove_rule_statement(statements: List[Statement], rule_id: str, smt_id: str) -> List[Statement]: