Changed level of kwargs merger method.

snappy_pipeline/base.py

@@ -4,6 +4,7 @@
 
 from collections import OrderedDict
 from collections.abc import MutableMapping
+from copy import deepcopy
 import os
 import sys
 import warnings
@@ -89,6 +90,33 @@ def print_sample_sheets(step, file=sys.stderr):
         _ordered_dump(info.sheet.json_data, stream=file, default_flow_style=False)
 
 
+def merge_kwargs(first_kwargs, second_kwargs):
+    """Merge two keyword arguments.
+
+    :param first_kwargs: First keyword arguments dictionary.
+    :type first_kwargs: dict
+
+    :param second_kwargs: Second keyword arguments dictionary.
+    :type second_kwargs: dict
+
+    :return: Returns merged dictionary with inputted keyword arguments.
+    """
+    # Global if no individual dict
+    if first_kwargs and (not second_kwargs):
+        return first_kwargs
+    # Individual if no global dict
+    elif (not first_kwargs) and second_kwargs:
+        return second_kwargs
+    # Merge dicts if both defined
+    elif first_kwargs and second_kwargs:
+        global_copy_kwargs = deepcopy(first_kwargs)
+        global_copy_kwargs.update(second_kwargs)
+        return global_copy_kwargs
+    # None if both None
+    else:
+        return None
+
+
 def merge_dicts(dict1, dict2, dict_class=OrderedDict):
     """Merge dictionary ``dict2`` into ``dict1``"""
 

snappy_pipeline/workflows/abstract/__init__.py

@@ -28,6 +28,7 @@
 from snappy_pipeline.base import (
     MissingConfiguration,
     merge_dicts,
+    merge_kwargs,
     print_config,
     print_sample_sheets,
     snakefile_path,
@@ -508,8 +509,8 @@ def __init__(
         self.sheets = [info.sheet for info in self.data_set_infos]
         #: Shortcut BioMed SampleSheet keyword arguments
         sheet_kwargs_list = [
-            self._shortcut_sheets_kwargs(global_sheet_kwargs=self.sheet_shortcut_kwargs,
-                                         single_sheet_kwargs=info.pedigree_field_kwargs)
+            merge_kwargs(first_kwargs=self.sheet_shortcut_kwargs,
+                         second_kwargs=info.pedigree_field_kwargs)
             for info in self.data_set_infos
         ]
         #: Shortcut sheets
@@ -526,35 +527,6 @@ def __init__(
         #: Functions from sub workflows, can be used to generate output paths into these workflows
         self.sub_workflows = {}
 
-    @staticmethod
-    def _shortcut_sheets_kwargs(global_sheet_kwargs, single_sheet_kwargs):
-        """
-        :param global_sheet_kwargs: Globally defined keyword arguments used to construct
-        shortcut sheets.
-        :type global_sheet_kwargs: dict
-
-        :param single_sheet_kwargs: Individually defined keyword arguments used to
-        construct specific shortcut sheets. For instance, it might contains specifications on which
-        field to base the pedigree definition in the cohort.
-        :type single_sheet_kwargs: dict
-
-        :return: Returns merged dictionary with global and individual keyword arguments.
-        """
-        # Global if no individual dict
-        if global_sheet_kwargs and (not single_sheet_kwargs):
-            return global_sheet_kwargs
-        # Individual if no global dict
-        elif (not global_sheet_kwargs) and single_sheet_kwargs:
-            return single_sheet_kwargs
-        # Merge dicts if both defined
-        elif global_sheet_kwargs and single_sheet_kwargs:
-            global_copy_kwargs = deepcopy(global_sheet_kwargs)
-            global_copy_kwargs.update(single_sheet_kwargs)
-            return global_copy_kwargs
-        # None if both None
-        if not (global_sheet_kwargs and single_sheet_kwargs):
-            return None
-
     def _setup_hooks(self):
         """Setup Snakemake workflow hooks for start/end/error"""
         # In the following, the "log" parameter to the handler functions is set to "_" as we

tests/snappy_pipeline/test_base.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+"""Tests for base module code"""
+
+from snappy_pipeline.base import merge_kwargs
+
+
+def test_merge_kwargs():
+    """Tests dictionary merger for shortcut sheet keyword arguments."""
+    # Initialise variables
+    global_kwargs = {1: "one", 2: "two"}
+    sheet_kwargs = {3: "three"}
+    merged_kwargs = {1: "one", 2: "two", 3: "three"}
+
+    # None if both none
+    expected = None
+    actual = merge_kwargs(first_kwargs=None, second_kwargs=None)
+    assert actual == expected
+
+    # Only sheet if global is none
+    actual = merge_kwargs(first_kwargs=None, second_kwargs=sheet_kwargs)
+    assert actual == sheet_kwargs
+
+    # Only global if sheet is none
+    actual = merge_kwargs(first_kwargs=global_kwargs, second_kwargs=None)
+    assert actual == global_kwargs
+
+    # Merged if both presents
+    actual = merge_kwargs(first_kwargs=global_kwargs, second_kwargs=sheet_kwargs)
+    assert actual == merged_kwargs

tests/snappy_pipeline/workflows/test_workflows_abstract.py

@@ -284,33 +284,3 @@ def test_base_step_ensure_w_config(dummy_generic_step):
     dummy_generic_step.ensure_w_config(("step_config", "dummy", "key"), "should be OK")
     with pytest.raises(MissingConfiguration):
         dummy_generic_step.ensure_w_config(("step_config", "dummy", "foo"), "should fail")
-
-
-def test_base_step__shortcut_sheets_kwargs(dummy_generic_step):
-    """Tests dictionary merger for shortcut sheet keyword arguments."""
-    # Initialise variables
-    global_kwargs = {1: "one", 2: "two"}
-    sheet_kwargs = {3: "three"}
-    merged_kwargs = {1: "one", 2: "two", 3: "three"}
-
-    # None if both none
-    expected = None
-    actual = dummy_generic_step._shortcut_sheets_kwargs(global_sheet_kwargs=None,
-                                                        single_sheet_kwargs=None)
-    assert actual == expected
-
-    # Only sheet if global is none
-    actual = dummy_generic_step._shortcut_sheets_kwargs(global_sheet_kwargs=None,
-                                                        single_sheet_kwargs=sheet_kwargs)
-    assert actual == sheet_kwargs
-
-    # Only global if sheet is none
-    actual = dummy_generic_step._shortcut_sheets_kwargs(global_sheet_kwargs=global_kwargs,
-                                                        single_sheet_kwargs=None)
-    assert actual == global_kwargs
-
-    # Merged if both presents
-    actual = dummy_generic_step._shortcut_sheets_kwargs(global_sheet_kwargs=global_kwargs,
-                                                        single_sheet_kwargs=sheet_kwargs)
-    assert actual == merged_kwargs
-