Repository: matchms/matchms
Pull request #479 (2023-08-14 08:55:30)

Move yaml file creation and loading outside Pipeline.py
This should solve #475 
The PR's #482 #481 #480 build further onto this PR. 

Pipeline became quite a complicated class. I wanted to implement some changes to the way the yaml file was stored, but this was a bit challenging due to the fact that yaml file/workflow is used although Pipeline, making it harder to make changes.

As a solution the yaml file creation is moved outside of Pipeline. 

These changes will remove the option to modify an already existing Pipeline object. So you cannot run Pipeline and after add some steps and than run again. I think this was unwanted behaviour anyway, since this can result in having a difference between your Yaml file and the actual pipeline that was run, which is unwanted. 


Patch:
diff --git a/.github/workflows/CI_build.yml b/.github/workflows/CI_build.yml
index f8e3c7cd..09a25c89 100644
--- a/.github/workflows/CI_build.yml
+++ b/.github/workflows/CI_build.yml
@@ -1,8 +1,9 @@
 name: CI Build
 
 on:
+  push:
   pull_request:
-    types: [opened, ready_for_review, closed]
+    branches: [ master ]
   workflow_dispatch:
     inputs:
       tag:
@@ -11,6 +12,7 @@ on:
         default: ''
 
 jobs:
+
   first_check:
     name: first code check / python-3.9 / ubuntu-latest
     runs-on: ubuntu-latest
diff --git a/.github/workflows/CI_first_code_check.yml b/.github/workflows/CI_first_code_check.yml
deleted file mode 100644
index d667c4bb..00000000
--- a/.github/workflows/CI_first_code_check.yml
+++ /dev/null
@@ -1,60 +0,0 @@
-name: CI first code check
-
-on:
-  push:
-  workflow_dispatch:
-    inputs:
-      tag:
-        description: Tag for manually running CI first code check workflow
-        required: False
-        default: ''
-
-jobs:
-
-  first_check:
-    name: first code check / python-3.9 / ubuntu-latest
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - name: Set up Python
-        uses: actions/setup-python@v2
-        with:
-          python-version: 3.9
-      - name: Python info
-        run: |
-          which python
-          python --version
-      - name: Build package and create dev environment
-        run: |
-          python -m pip install --upgrade pip poetry
-          poetry install
-      - name: Show pip list
-        run: |
-          pip list
-      - name: Test with coverage
-        run: |
-          poetry run pytest --cov --cov-report term --cov-report xml --junitxml=xunit-result.xml
-      - name: Correct coverage paths
-        run: sed -i "s+$PWD/++g" coverage.xml
-      - name: Check style against standards using prospector
-        shell: bash -l {0}
-        run: prospector -o grouped -o pylint:pylint-report.txt
-      - name: Store PR metadata in json file
-        env:
-          PR_NUMBER: ${{ github.event.pull_request.number }}
-          HEAD_REF: ${{ github.event.pull_request.head.ref }}
-          BASE_REF: ${{ github.event.pull_request.base.ref }}
-        run: |
-          jq -n --arg pr_number "$PR_NUMBER"\
-                --arg head_ref "$HEAD_REF"\
-                --arg base_ref "$BASE_REF"\
-            '{"pr_number": $pr_number, "head_ref": $head_ref, "base_ref": $base_ref}' > pr_metadata.json
-      - uses: actions/upload-artifact@v3
-        with:
-          name: sonarcloud-data-pip
-          path: |
-            coverage.xml
-            xunit-result.xml
-            pr_metadata.json
-      - name: Check whether import statements are used consistently
-        run: isort . --check-only --diff
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 76da7629..32b7b238 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,16 +5,6 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [unreleased]
-
-### Added
-
-- New `SpectrumProcessing` class to be the central hub for all filter functions [#455](https://github.com/matchms/matchms/pull/455). Also takes care that filters are executed in a useful order. This is also integrated into the `Pipeline` class.
-
-### Changed
-
-- Extensive code refactoring and cleaning.
-
 ## [0.21.2] - 2023-08-01
 
 ### Added
diff --git a/NOTICE b/NOTICE
new file mode 100644
index 00000000..e69de29b
diff --git a/README.rst b/README.rst
index 3ebdd02e..0250c3bb 100644
--- a/README.rst
+++ b/README.rst
@@ -137,8 +137,6 @@ To date we are aware of:
 
 + `matchmsextras <https://github.com/matchms/matchmsextras>`_ which contains additional functions to create networks based on spectral similarities, to run spectrum searchers against `PubChem`, or additional plotting methods.
 
-+ `MS2Query <https://github.com/iomega/ms2query>`_ Reliable and fast MS/MS spectral-based analogue search, running on top op matchms.
-
 + `memo <https://github.com/mandelbrot-project/memo>`_ a method allowing a Retention Time (RT) agnostic alignment of metabolomics samples using the fragmentation spectra (MS2) of their consituents.
 
 *(if you know of any other packages that are fully compatible with matchms, let us know!)*
diff --git a/matchms/Pipeline.py b/matchms/Pipeline.py
index a1c70624..01cd774a 100644
--- a/matchms/Pipeline.py
+++ b/matchms/Pipeline.py
@@ -17,6 +17,66 @@ _score_functions = {key.lower(): f for key, f in mssimilarity.__dict__.items() i
 logger = logging.getLogger("matchms")
 
 
+def create_workflow(yaml_file_name=None,
+                    predefined_processing_queries="default",
+                    predefined_processing_reference="default",
+                    additional_filters_queries=(),
+                    additional_filters_references=(),
+                    score_computations=(),
+                    query_file_name=None,
+                    reference_file_name=None,
+                    ):
+    # pylint: disable=too-many-arguments
+    workflow = OrderedDict()
+    workflow["importing"] = {"queries": query_file_name,
+                             "references": reference_file_name}
+    workflow["predefined_processing_queries"] = predefined_processing_queries
+    workflow["predefined_processing_references"] = predefined_processing_reference
+    workflow["additional_processing_queries"] = additional_filters_queries
+    workflow["additional_processing_references"] = additional_filters_references
+    workflow["score_computations"] = score_computations
+    if yaml_file_name is not None:
+        with open(yaml_file_name, 'w', encoding="utf-8") as file:
+            file.write("# Matchms pipeline config file \n")
+            file.write("# Change and adapt fields where necessary \n")
+            file.write("# " + 20 * "=" + " \n")
+            ordered_dump(workflow, file)
+    return workflow
+
+
+def load_in_workflow_from_yaml_file(yaml_file):
+    with open(yaml_file, 'r', encoding="utf-8") as file:
+        workflow = ordered_load(file, yaml.SafeLoader)
+    return workflow
+
+
+def ordered_load(stream, loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
+    """ Code from https://stackoverflow.com/questions/5121931/in-python-how-can-you-load-yaml-mappings-as-ordereddicts
+    """
+    class OrderedLoader(loader):
+        pass
+
+    def construct_mapping(loader, node):
+        loader.flatten_mapping(node)
+        return object_pairs_hook(loader.construct_pairs(node))
+    OrderedLoader.add_constructor(
+        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
+        construct_mapping)
+    return yaml.load(stream, OrderedLoader)
+
+
+def ordered_dump(data, stream=None, dumper=yaml.SafeDumper, **kwds):
+    class OrderedDumper(dumper):
+        pass
+
+    def _dict_representer(dumper, data):
+        return dumper.represent_mapping(
+            yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
+            data.items())
+    OrderedDumper.add_representer(OrderedDict, _dict_representer)
+    return yaml.dump(data, stream, OrderedDumper, **kwds)
+
+
 class Pipeline:
     """Central pipeline class.
 
@@ -71,64 +131,53 @@ class Pipeline:
                                        ["filter_by_range", {"name": "Spec2Vec", "low": 0.3}]]
 
     """
-    def __init__(self, config_file=None, progress_bar=True):
+    def __init__(self, workflow: OrderedDict, progress_bar=True):
         """
         Parameters
         ----------
-        config_file
-            Filename of config file (yaml file) to define pipeline. Default is None
-            in which case the pipeline should be defined via a Python script.
-        progress_bar
+        workflow:
+            Contains an orderedDict containing the workflow settings. Can be created using create_workflow.
+        progress_bar:
             Default is True. Set to False if no progress bar should be displayed.
         """
         self.spectrums_queries = []
         self.spectrums_references = []
         self.is_symmetric = False
-        self.import_workflow_from_yaml(config_file)
+        self.workflow = workflow
+        self.complete_workflow()
+        self.write_to_logfile("--- Processing pipeline: ---")
+        self._initialize_spectrum_processor_queries()
+        if self.is_symmetric is False:
+            self._initialize_spectrum_processor_references()
         self.scores = None
         self.progress_bar = progress_bar
 
     def _initialize_spectrum_processor_queries(self):
         """Initialize spectrum processing workflow for the query spectra."""
         self.processing_queries = initialize_spectrum_processor(
-            self.predefined_processing_queries,
-            self.additional_processing_queries
+            self.workflow["predefined_processing_queries"],
+            self.workflow["additional_processing_queries"]
             )
         self.write_to_logfile(str(self.processing_queries))
 
     def _initialize_spectrum_processor_references(self):
         """Initialize spectrum processing workflow for the reference spectra."""
         self.processing_references = initialize_spectrum_processor(
-            self.predefined_processing_references,
-            self.additional_processing_references
+            self.workflow["predefined_processing_references"],
+            self.workflow["additional_processing_references"]
             )
         self.write_to_logfile(str(self.processing_references))
 
-    def import_workflow_from_yaml(self, config_file):
+    def complete_workflow(self):
         """Define Pipeline workflow based on a yaml file (config_file).
         """
-        if config_file is None:
-            self.workflow = OrderedDict()
-            self.workflow["importing"] = {"queries": None,
-                                          "references": None}
-            self.workflow["predefined_processing_queries"] = "default"
-            self.workflow["predefined_processing_references"] = "default"
-            self.workflow["additional_processing_queries"] = []
-            self.workflow["additional_processing_references"] = []
-            self.workflow["score_computations"] = []
-        else:
-            with open(config_file, 'r', encoding="utf-8") as file:
-                self.workflow = ordered_load(file, yaml.SafeLoader)
-            if self.workflow["additional_processing_references"] == "processing_queries":
-                self.workflow["additional_processing_references"] = self.workflow["additional_processing_queries"]
+        #todo add a check if all the fields are given that are needed.
+        if self.workflow["additional_processing_references"] == "processing_queries":
+            self.workflow["additional_processing_references"] = self.workflow["additional_processing_queries"]
         if "logging" not in self.workflow:
             self.workflow["logging"] = {}
         if self.logging_level is None:
             self.logging_level = "WARNING"
-        self.write_to_logfile("--- Processing pipeline: ---")
-        self._initialize_spectrum_processor_queries()
-        if self.is_symmetric is False:
-            self._initialize_spectrum_processor_references()
 
     def run(self):
         """Execute the defined Pipeline workflow.
@@ -300,15 +349,6 @@ class Pipeline:
                 spectrums_references += list(load_spectra(reference_file))
             self.spectrums_references += spectrums_references
 
-    def create_workflow_config_file(self, filename):
-        """Save the current pipeline workflow as a yaml file.
-        This file allows to reconstruct the current workflow or can be adapted as desired.
-        """
-        with open(filename, 'w', encoding="utf-8") as file:
-            file.write("# Matchms pipeline config file \n")
-            file.write("# Change and adapt fields where necessary \n")
-            file.write("# " + 20 * "=" + " \n")
-            ordered_dump(self.workflow, file)
 
     # Getter & Setters
     @property
@@ -343,42 +383,6 @@ class Pipeline:
     def logging_level(self, log_level):
         self.workflow["logging"]["logging_level"] = log_level
 
-    @property
-    def predefined_processing_queries(self):
-        return self.workflow.get("predefined_processing_queries")
-
-    @predefined_processing_queries.setter
-    def predefined_processing_queries(self, files):
-        self.workflow["predefined_processing_queries"] = files
-        self._initialize_spectrum_processor_queries()
-
-    @property
-    def predefined_processing_references(self):
-        return self.workflow.get("predefined_processing_references")
-
-    @predefined_processing_references.setter
-    def predefined_processing_references(self, files):
-        self.workflow["predefined_processing_references"] = files
-        self._initialize_spectrum_processor_references()
-
-    @property
-    def additional_processing_queries(self):
-        return self.workflow.get("additional_processing_queries")
-
-    @additional_processing_queries.setter
-    def additional_processing_queries(self, files):
-        self.workflow["additional_processing_queries"] = files
-        self._initialize_spectrum_processor_queries()
-
-    @property
-    def additional_processing_references(self):
-        return self.workflow.get("additional_processing_references")
-
-    @additional_processing_references.setter
-    def additional_processing_references(self, files):
-        self.workflow["additional_processing_references"] = files
-        self._initialize_spectrum_processor_references()
-
     @property
     def score_computations(self):
         return self.workflow.get("score_computations")
@@ -388,33 +392,6 @@ class Pipeline:
         self.workflow["score_computations"] = computations_list
 
 
-def ordered_load(stream, loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
-    """ Code from https://stackoverflow.com/questions/5121931/in-python-how-can-you-load-yaml-mappings-as-ordereddicts
-    """
-    class OrderedLoader(loader):
-        pass
-
-    def construct_mapping(loader, node):
-        loader.flatten_mapping(node)
-        return object_pairs_hook(loader.construct_pairs(node))
-    OrderedLoader.add_constructor(
-        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
-        construct_mapping)
-    return yaml.load(stream, OrderedLoader)
-
-
-def ordered_dump(data, stream=None, dumper=yaml.SafeDumper, **kwds):
-    class OrderedDumper(dumper):
-        pass
-
-    def _dict_representer(dumper, data):
-        return dumper.represent_mapping(
-            yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
-            data.items())
-    OrderedDumper.add_representer(OrderedDict, _dict_representer)
-    return yaml.dump(data, stream, OrderedDumper, **kwds)
-
-
 def initialize_spectrum_processor(predefined_workflow, additional_filters):
     """Initialize spectrum processing workflow."""
     processor = SpectrumProcessor(predefined_workflow)


Test patch:
diff --git a/integration-tests/test_user_workflow.py b/integration-tests/test_user_workflow.py
index 8a2bec0e..35c7210f 100644
--- a/integration-tests/test_user_workflow.py
+++ b/integration-tests/test_user_workflow.py
@@ -2,24 +2,22 @@ import os
 import numpy as np
 import pytest
 from matchms import Pipeline
+from matchms.Pipeline import create_workflow
 from matchms.similarity import CosineGreedy
 
 
 def test_user_workflow():
     module_root = os.path.join(os.path.dirname(__file__), "..")
     spectrums_file = os.path.join(module_root, "tests", "testdata", "pesticides.mgf")
-
-    pipeline = Pipeline()
-    pipeline.query_files = spectrums_file
-    pipeline.predefined_processing_queries = "basic"
-    pipeline.additional_processing_queries = [
-        ["add_parent_mass"],
-        ["normalize_intensities"],
-        ["select_by_relative_intensity", {"intensity_from": 0.0, "intensity_to": 1.0}],
-        ["select_by_mz", {"mz_from": 0, "mz_to": 1000}],
-        ["require_minimum_number_of_peaks", {"n_required": 5}]
-    ]
-    pipeline.score_computations = [["cosinegreedy",  {"tolerance": 0.3}]]
+    workflow = create_workflow(query_file_name=spectrums_file,
+                               predefined_processing_queries="basic",
+                               additional_filters_queries=[["add_parent_mass"],
+                                                           ["normalize_intensities"],
+                                                           ["select_by_relative_intensity", {"intensity_from": 0.0, "intensity_to": 1.0}],
+                                                           ["select_by_mz", {"mz_from": 0, "mz_to": 1000}],
+                                                           ["require_minimum_number_of_peaks", {"n_required": 5}]],
+                               score_computations=[["cosinegreedy",  {"tolerance": 0.3}]])
+    pipeline = Pipeline(workflow)
     pipeline.run()
 
     scores = pipeline.scores
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 4c7e8bf1..8fe47c5f 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -3,6 +3,7 @@ import numpy as np
 import pytest
 from matchms import Pipeline
 from matchms.filtering import select_by_mz
+from matchms.Pipeline import create_workflow, load_in_workflow_from_yaml_file
 from matchms.similarity import ModifiedCosine
 
 
@@ -11,7 +12,8 @@ spectrums_file_msp = os.path.join(module_root, "tests", "testdata", "massbank_fi
 
 
 def test_pipeline_initial_check_missing_file():
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = "non_existing_file.msp"
     pipeline.score_computations = [["precursormzmatch",  {"tolerance": 120.0}]]
     with pytest.raises(AssertionError) as msg:
@@ -20,7 +22,8 @@ def test_pipeline_initial_check_missing_file():
 
 
 def test_pipeline_initial_check_unknown_step():
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.score_computations = [["precursormzOOPSmatch",  {"tolerance": 120.0}]]
     with pytest.raises(ValueError) as msg:
@@ -29,7 +32,8 @@ def test_pipeline_initial_check_unknown_step():
 
 
 def test_pipeline_symmetric():
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.predefined_processing_queries = "basic"
     pipeline.score_computations = [["precursormzmatch",  {"tolerance": 120.0}],
@@ -49,7 +53,8 @@ def test_pipeline_symmetric():
 
 
 def test_pipeline_symmetric_filters():
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.predefined_processing_queries = "basic"
     pipeline.additional_processing_queries = [[select_by_mz, {"mz_from": 0, "mz_to": 1000}]]
@@ -71,7 +76,8 @@ def test_pipeline_symmetric_filters():
 
 
 def test_pipeline_symmetric_masking():
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.predefined_processing_queries = "basic"
     pipeline.score_computations = [["precursormzmatch",  {"tolerance": 120.0}],
@@ -92,7 +98,8 @@ def test_pipeline_symmetric_masking():
 
 
 def test_pipeline_symmetric_custom_score():
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.predefined_processing_queries = "basic"
     pipeline.score_computations = [["precursormzmatch",  {"tolerance": 120.0}],
@@ -113,7 +120,8 @@ def test_pipeline_symmetric_custom_score():
 
 def test_pipeline_non_symmetric():
     """Test importing from multiple files and different inputs for query and references."""
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.predefined_processing_queries = "basic"
     pipeline.predefined_processing_references = "basic"
@@ -138,8 +146,8 @@ def test_pipeline_non_symmetric():
 def test_pipeline_from_yaml():
     pytest.importorskip("rdkit")
     config_file = os.path.join(module_root, "tests", "test_pipeline.yaml")
-    pipeline = Pipeline(config_file)
-    assert pipeline.predefined_processing_queries == "default"
+    workflow = load_in_workflow_from_yaml_file(config_file)
+    pipeline = Pipeline(workflow)
     pipeline.run()
     assert len(pipeline.spectrums_queries) == 5
     assert pipeline.spectrums_queries[0] == pipeline.spectrums_references[0]
@@ -155,24 +163,27 @@ def test_pipeline_from_yaml():
 def test_pipeline_to_and_from_yaml(tmp_path):
     pytest.importorskip("rdkit")
     config_file = os.path.join(tmp_path, "test_pipeline.yaml")
-    pipeline = Pipeline()
-    pipeline.query_files = spectrums_file_msp
-    pipeline.score_computations = [["precursormzmatch",  {"tolerance": 120.0}],
-                                   ["modifiedcosine", {"tolerance": 10.0}]]
-    pipeline.create_workflow_config_file(config_file)
+
+    workflow = create_workflow(config_file, score_computations=[["precursormzmatch",  {"tolerance": 120.0}],
+                                                                ["modifiedcosine", {"tolerance": 10.0}]],
+                               query_file_name=spectrums_file_msp)
+    assert os.path.exists(config_file)
+
+    pipeline = Pipeline(workflow)
     pipeline.run()
     scores_run1 = pipeline.scores
-    assert os.path.exists(config_file)
 
     # Load again
-    pipeline = Pipeline(config_file)
+    workflow = load_in_workflow_from_yaml_file(config_file)
+    pipeline = Pipeline(workflow)
     pipeline.run()
     assert pipeline.scores.scores == scores_run1.scores
 
 
 def test_pipeline_logging(tmp_path):
     pytest.importorskip("rdkit")
-    pipeline = Pipeline()
+    workflow = create_workflow()
+    pipeline = Pipeline(workflow)
     pipeline.query_files = spectrums_file_msp
     pipeline.score_computations = [["precursormzmatch",  {"tolerance": 120.0}],
                                    [ModifiedCosine, {"tolerance": 10.0}]]
@@ -184,21 +195,20 @@ def test_pipeline_logging(tmp_path):
     with open(logfile, "r", encoding="utf-8") as f:
         firstline = f.readline().rstrip()
     assert "Start running matchms pipeline" in firstline
-    
+
 
 def test_FingerprintSimilarity_pipeline():
     pytest.importorskip("rdkit")
-    pipeline = Pipeline()
-    pipeline.query_files = spectrums_file_msp
-    pipeline.reference_files = spectrums_file_msp
-    pipeline.predefined_processing_queries = "basic"
-    pipeline.additional_processing_queries = ["add_fingerprint"]
-    pipeline.predefined_processing_references = "basic"
-    pipeline.additional_processing_references = ["add_fingerprint"]
-    pipeline.score_computations = [
-        ["metadatamatch", {"field": "precursor_mz", "matching_type": "difference", "tolerance": 50}],
-        ["fingerprintsimilarity", {"similarity_measure": "jaccard"}]
-    ]    
+    workflow = create_workflow(predefined_processing_queries="basic",
+                               additional_filters_queries=["add_fingerprint"],
+                               predefined_processing_reference="basic",
+                               additional_filters_references=["add_fingerprint"],
+                               query_file_name=spectrums_file_msp,
+                               reference_file_name=spectrums_file_msp,
+                               score_computations=[["metadatamatch", {"field": "precursor_mz", "matching_type": "difference", "tolerance": 50}],
+                                                   ["fingerprintsimilarity", {"similarity_measure": "jaccard"}]],
+                               )
+    pipeline = Pipeline(workflow)
     pipeline.run()
     assert len(pipeline.spectrums_queries[0].get("fingerprint")) == 2048
     assert pipeline.scores.scores.shape == (5, 5, 2)
diff --git a/tests/test_scoring_method_consistency.py b/tests/test_scoring_method_consistency.py
index c8fa8b2e..77f4ccc5 100644
--- a/tests/test_scoring_method_consistency.py
+++ b/tests/test_scoring_method_consistency.py
@@ -13,6 +13,7 @@ import matchms.filtering as msfilter
 import matchms.similarity as mssimilarity
 from matchms import Pipeline
 from matchms.importing import load_from_json
+from matchms.Pipeline import create_workflow
 
 
 module_root = os.path.dirname(__file__)
@@ -83,14 +84,12 @@ def test_consistency_scoring_and_pipeline(spectrums, similarity_measure):
     computed_scores_matrix = scoring_method.matrix(spectrums, spectrums)
 
     # Run pipeline
-    pipeline = Pipeline()
-    pipeline.query_files = json_file
-    pipeline.predefined_processing_queries = "basic"
-    pipeline.additional_processing_queries = [
-        ["add_parent_mass"],
-        ["normalize_intensities"]
-        ]
-    pipeline.score_computations = [similarity_measure]
+    workflow = create_workflow(query_file_name=json_file,
+                               predefined_processing_queries="basic",
+                               additional_filters_queries=[["add_parent_mass"], ["normalize_intensities"]],
+                               score_computations=[similarity_measure]
+                               )
+    pipeline = Pipeline(workflow)
     pipeline.run()
 
     if computed_scores_matrix.dtype.names is None: