Repository: fkie-cad/Logprep (language: python)

Issue:
--dry-run feature is broken for rules with defined outputs like the predetector
<!--- A clear and concise description of the issue. --->
Currently the `--dry-run` feature is broken if you use processors or rules with extra_outputs.
**Expected behavior**
<!--- A clear and concise description of what you expected to happen. --->

**Current behavior**
<!--- A clear and concise description of what happens instead of the expected behavior. --->

**Steps to reproduce**
<!--- A link to a live example, or an unambiguous set of steps to reproduce this bug. -->

```powershell
pwsh> $env:PYTHONPATH="."; python  ./logprep/run_logprep.py ./quickstart/exampledata/config/pipeline.yml --dry-run ./quickstart/exampledata/input_logdata/test_input.jsonl                                                           2023-03-17 09:51:47,937 Logprep INFO    : Log level set to 'INFO'                                                                                                                                                                    
2023-03-17 09:51:47,979 Logprep INFO    : python version:          3.9.16
2023-03-17 09:51:47,979 Logprep INFO    : logprep version:         5.0.1+26.g0ab127a
2023-03-17 09:51:47,979 Logprep INFO    : configuration version:   1, file://./quickstart/exampledata/config/pipeline.yml
Invalid processor configuration: pre_detector: output 'kafka' does not exist in logprep outputs
Invalid processor configuration: pseudonymizer: output 'kafka' does not exist in logprep outputs
Traceback (most recent call last):
  File "/home/vagrant/external_work/Logprep/./logprep/run_logprep.py", line 242, in <module>
    main()
  File "/home/vagrant/external_work/Logprep/./logprep/run_logprep.py", line 220, in main
    dry_runner.run()
  File "/home/vagrant/external_work/Logprep/logprep/util/rule_dry_runner.py", line 145, in run
    patched_runner = get_patched_runner(self.patched_config_path, self._logger)
  File "/home/vagrant/external_work/Logprep/logprep/util/rule_dry_runner.py", line 118, in get_patched_runner
    runner.load_configuration(config_path)
  File "/home/vagrant/external_work/Logprep/logprep/runner.py", line 153, in load_configuration
    configuration.verify(self._logger)
  File "/home/vagrant/external_work/Logprep/logprep/util/configuration.py", line 214, in verify
    raise error
  File "/home/vagrant/external_work/Logprep/logprep/util/configuration.py", line 256, in _perform_verfification_and_get_errors
    self._verify_pipeline(logger)
  File "/home/vagrant/external_work/Logprep/logprep/util/configuration.py", line 381, in _verify_pipeline
    raise InvalidConfigurationErrors(errors)
logprep.util.configuration.InvalidConfigurationErrors: Invalid processor configuration: pre_detector: output 'kafka' does not exist in logprep outputs
Invalid processor configuration: pseudonymizer: output 'kafka' does not exist in logprep outputs
(.venv)  .venv python3.9:vagrant@dev-machine:~/external_work/Logprep:  [main ≡]                                                                                                                                           819ms  ✓ 
PS> 
```

**Environment**

Logprep version: main branch commit 0ab127a
Python version: all



Discussion / hints:

Fix patch:
diff --git a/logprep/framework/pipeline.py b/logprep/framework/pipeline.py
index de40275ca..424d6539b 100644
--- a/logprep/framework/pipeline.py
+++ b/logprep/framework/pipeline.py
@@ -436,9 +436,10 @@ def process_event(self, event: dict):
                     self._handle_processing_warning(processor, warning)
             except BaseException as error:  # pylint: disable=broad-except
                 msg = self._handle_fatal_processing_error(processor, error)
-                for _, output in self._output.items():
-                    if output.default:
-                        output.store_failed(msg, json.loads(event_received), event)
+                if self._output:
+                    for _, output in self._output.items():
+                        if output.default:
+                            output.store_failed(msg, json.loads(event_received), event)
                 processor.metrics.number_of_errors += 1
                 event.clear()  # 'delete' the event, i.e. no regular output
             if not event:
diff --git a/logprep/util/aggregating_logger.py b/logprep/util/aggregating_logger.py
index 05e17c67b..4d550c248 100644
--- a/logprep/util/aggregating_logger.py
+++ b/logprep/util/aggregating_logger.py
@@ -49,10 +49,11 @@ def setup(cls, config: dict, logger_disabled: bool = False):
 
         """
         cls.logger_disabled = logger_disabled
-        cls.logger_config = config.get("logger", dict())
+        cls.logger_config = config.get("logger", {})
 
         cls.level_str = cls.logger_config.get("level", "INFO")
-
+        if cls.level_str is None:
+            cls.level_str = "INFO"
         cls.log_level = name_to_level.get(cls.level_str.upper(), INFO)
         basicConfig(
             level=cls.log_level, format="%(asctime)-15s %(name)-5s %(levelname)-8s: %(message)s"
@@ -84,7 +85,7 @@ def create(cls, name: str) -> Logger:
             logger.handlers = []
             logger.addHandler(SysLogHandler(address="/dev/log"))
 
-        if cls.level_str.upper() not in name_to_level.keys():
+        if cls.level_str.upper() not in name_to_level:
             logger.info(f"Invalid log level '{cls.level_str.upper()}', defaulting to 'INFO'")
         else:
             logger.setLevel(cls.log_level)
diff --git a/logprep/util/configuration.py b/logprep/util/configuration.py
index a8ab4abb9..bfe89ee6c 100644
--- a/logprep/util/configuration.py
+++ b/logprep/util/configuration.py
@@ -1,16 +1,17 @@
 """This module is used to create the configuration for the runner."""
 
-from copy import deepcopy
 import re
 import sys
+from copy import deepcopy
 from logging import Logger
 from pathlib import Path
 from typing import List
 
-from ruamel.yaml.scanner import ScannerError
 from colorama import Fore
-from logprep.abc.processor import Processor
+from ruamel.yaml.scanner import ScannerError
 
+from logprep.abc.getter import Getter
+from logprep.abc.processor import Processor
 from logprep.factory import Factory
 from logprep.factory_error import FactoryError
 from logprep.factory_error import (
@@ -19,7 +20,6 @@
 )
 from logprep.processor.base.exceptions import InvalidRuleDefinitionError
 from logprep.util.getter import GetterFactory
-from logprep.abc.getter import Getter
 from logprep.util.helper import print_fcolor
 from logprep.util.json_handling import dump_config_as_file
 
@@ -207,9 +207,8 @@ def patch_yaml_with_json_connectors(
 
     def verify(self, logger: Logger):
         """Verify the configuration."""
-        errors = self._perform_verfification_and_get_errors(logger)
-        self._print_errors(errors)
-
+        errors = self._check_for_errors(logger)
+        self._print_and_raise_errors(errors)
         for error in errors:
             raise error
 
@@ -220,17 +219,24 @@ def verify_pipeline_only(self, logger: Logger):
         """
         errors = []
         try:
-            self._verify_pipeline(logger, ignore_rule_errors=False)
+            self._verify_pipeline(logger)
         except InvalidConfigurationError as error:
             errors.append(error)
-        self._print_errors(errors)
+        self._print_and_raise_errors(errors)
 
-        for error in errors:
-            raise error
+    def verify_pipeline_without_processor_outputs(self, logger: Logger):
+        """Verify the configuration only for the pipeline, but ignore processor output errors.
+        This is used to check if the configuration is valid inside the auto rule tester and the
+        rule corpus tester, as the configuration does not have an output there.
+        """
+        errors = []
+        try:
+            self._verify_pipeline_without_processor_outputs(logger)
+        except InvalidConfigurationError as error:
+            errors.append(error)
+        self._print_and_raise_errors(errors)
 
-    def _perform_verfification_and_get_errors(
-        self, logger: Logger
-    ) -> List[InvalidConfigurationError]:
+    def _check_for_errors(self, logger: Logger) -> List[InvalidConfigurationError]:
         errors = []
         try:
             self._verify_environment()
@@ -332,45 +338,13 @@ def _format_type_error(error: TypeError) -> str:
             msg = f"Unknown option: {parameter}."
         return msg
 
-    def _verify_pipeline(self, logger: Logger, ignore_rule_errors=False):
-        if not self.get("pipeline"):
-            raise RequiredConfigurationKeyMissingError("pipeline")
-
-        if not isinstance(self["pipeline"], list):
-            error = InvalidConfigurationError(
-                '"pipeline" must be a list of processor dictionary configurations!'
-            )
-            raise InvalidConfigurationErrors([error])
-
+    def _verify_pipeline(self, logger: Logger):
+        self._verify_pipeline_key()
         errors = []
         for processor_config in self["pipeline"]:
-            processor = None
-            try:
-                processor = Factory.create(processor_config, logger)
-            except (FactoryInvalidConfigurationError, UnknownComponentTypeError) as error:
-                errors.append(
-                    InvalidProcessorConfigurationError(
-                        f"{list(processor_config.keys())[0]} - {error}"
-                    )
-                )
-            except TypeError as error:
-                msg = self._format_type_error(error)
-                errors.append(
-                    InvalidProcessorConfigurationError(
-                        f"{list(processor_config.keys())[0]} - {msg}"
-                    )
-                )
-            except InvalidRuleDefinitionError:
-                if not ignore_rule_errors:
-                    errors.append(
-                        InvalidConfigurationError(
-                            "Could not verify configuration for processor instance "
-                            f"'{list(processor_config.keys())[0]}', because it has invalid rules."
-                        )
-                    )
+            processor = self._verify_processor(errors, logger, processor_config)
             try:
-                if processor:
-                    self._verify_rules_outputs(processor)
+                self._verify_rules_outputs(processor)
             except InvalidRuleDefinitionError as error:
                 errors.append(error)
             try:
@@ -380,8 +354,47 @@ def _verify_pipeline(self, logger: Logger, ignore_rule_errors=False):
         if errors:
             raise InvalidConfigurationErrors(errors)
 
+    def _verify_pipeline_without_processor_outputs(self, logger: Logger):
+        self._verify_pipeline_key()
+        errors = []
+        for processor_config in self["pipeline"]:
+            self._verify_processor(errors, logger, processor_config)
+        if errors:
+            raise InvalidConfigurationErrors(errors)
+
+    def _verify_pipeline_key(self):
+        if not self.get("pipeline"):
+            raise RequiredConfigurationKeyMissingError("pipeline")
+        if not isinstance(self["pipeline"], list):
+            error = InvalidConfigurationError(
+                '"pipeline" must be a list of processor dictionary configurations!'
+            )
+            raise InvalidConfigurationErrors([error])
+
+    def _verify_processor(self, errors, logger, processor_config):
+        processor = None
+        try:
+            processor = Factory.create(processor_config, logger)
+        except (FactoryInvalidConfigurationError, UnknownComponentTypeError) as error:
+            errors.append(
+                InvalidProcessorConfigurationError(f"{list(processor_config.keys())[0]} - {error}")
+            )
+        except TypeError as error:
+            msg = self._format_type_error(error)
+            errors.append(
+                InvalidProcessorConfigurationError(f"{list(processor_config.keys())[0]} - {msg}")
+            )
+        except InvalidRuleDefinitionError:
+            errors.append(
+                InvalidConfigurationError(
+                    "Could not verify configuration for processor instance "
+                    f"'{list(processor_config.keys())[0]}', because it has invalid rules."
+                )
+            )
+        return processor
+
     def _verify_rules_outputs(self, processor: Processor):
-        if not hasattr(processor.rule_class, "outputs"):
+        if not processor or not hasattr(processor.rule_class, "outputs"):
             return
         for rule in processor.rules:
             for output in rule.outputs:
@@ -482,6 +495,8 @@ def _verify_measure_time_config(measure_time_config):
             )
 
     @staticmethod
-    def _print_errors(errors: List[BaseException]):
+    def _print_and_raise_errors(errors: List[BaseException]):
         for error in errors:
             print_fcolor(Fore.RED, str(error))
+        for error in errors:
+            raise error
diff --git a/logprep/util/rule_dry_runner.py b/logprep/util/rule_dry_runner.py
index 4f27aad13..1cac1b608 100644
--- a/logprep/util/rule_dry_runner.py
+++ b/logprep/util/rule_dry_runner.py
@@ -44,146 +44,82 @@
 import tempfile
 from copy import deepcopy
 from difflib import ndiff
+from functools import cached_property
 
 from colorama import Fore, Back
 from ruamel.yaml import YAML
 
-from logprep.runner import Runner
+from logprep.framework.pipeline import Pipeline
 from logprep.util.configuration import Configuration
 from logprep.util.getter import GetterFactory
 from logprep.util.helper import (
     color_print_line,
     recursive_compare,
-    remove_file_if_exists,
     color_print_title,
 )
-from logprep.util.json_handling import parse_jsonl, parse_json
 
 yaml = YAML(typ="safe", pure=True)
 
 
-def get_runner_outputs(patched_runner) -> list:
-    # pylint: disable=protected-access
-    """
-    Extracts the outputs of a patched logprep runner.
-
-    Parameters
-    ----------
-    patched_runner : Runner
-        The patched logprep runner
-
-    Returns
-    -------
-    parsed_outputs : list
-        A list of logprep outputs containing events, extra outputs like pre-detections or pseudonyms
-        and errors
-    """
-    parsed_outputs = [None, None, None]
-    output_config = list(patched_runner._configuration.get("output").values())[0]
-    output_paths = [
-        output_path for key, output_path in output_config.items() if "output_file" in key
-    ]
-
-    for output_path in output_paths:
-        remove_file_if_exists(output_path)
-
-    patched_runner.start()
-
-    for index, output_path in enumerate(output_paths):
-        parsed_outputs[index] = parse_jsonl(output_path)
-        remove_file_if_exists(output_path)
-
-    return parsed_outputs
-
-
-def get_patched_runner(config_path, logger):
-    """
-    Creates a patched runner that bypasses check to obtain non singleton instance and the runner
-    won't continue iterating on an empty pipeline.
-
-    Parameters
-    ----------
-    config_path : str
-        The logprep configuration that should be used for the patched runner
-    logger : Logger
-        The application logger the runner should use
-
-    Returns
-    -------
-    runner : Runner
-        The patched logprep runner
-    """
-    runner = Runner(bypass_check_to_obtain_non_singleton_instance=True)
-    runner.set_logger(logger)
-    runner.load_configuration(config_path)
-
-    # patch runner to stop on empty pipeline
-    def keep_iterating():
-        """generator that stops on first iteration"""
-        return  # nosemgrep
-        yield
-
-    runner._keep_iterating = keep_iterating  # pylint: disable=protected-access
-
-    return runner
-
-
 class DryRunner:
     """Used to run pipeline with given events and show changes made by processing."""
 
-    def __init__(self, dry_run: str, config_path: str, full_output: bool, use_json: bool, logger):
-        self._tmp_path = tempfile.mkdtemp()
-        self.patched_config_path = Configuration.patch_yaml_with_json_connectors(
-            original_config_path=config_path, output_dir=self._tmp_path, input_file_path=dry_run
+    @cached_property
+    def _tmp_path(self):
+        return tempfile.mkdtemp()
+
+    @cached_property
+    def _pipeline(self):
+        patched_config_path = Configuration.patch_yaml_with_json_connectors(
+            original_config_path=self._config_path,
+            output_dir=self._tmp_path,
+            input_file_path=self._input_file_path,
         )
+        config = Configuration.create_from_yaml(patched_config_path)
+        config.verify_pipeline_without_processor_outputs(self._logger)
+        del config["output"]
+        return Pipeline(config=config)
+
+    @cached_property
+    def _input_documents(self):
+        return GetterFactory.from_string(self._input_file_path).get_yaml()
+
+    def __init__(
+        self, input_file_path: str, config_path: str, full_output: bool, use_json: bool, logger
+    ):
+        self._input_file_path = input_file_path
+        self._config_path = config_path
         self._full_output = full_output
         self._use_json = use_json
         self._logger = logger
 
     def run(self):
         """Run the dry runner."""
-        patched_runner = get_patched_runner(self.patched_config_path, self._logger)
-        test_output, test_output_custom, test_output_error = get_runner_outputs(patched_runner)
-        config_yaml = GetterFactory.from_string(self.patched_config_path).get_yaml()
-        input_path = config_yaml.get("input", {}).get("patched_input", {}).get("documents_path")
-        input_data = parse_json(input_path) if self._use_json else parse_jsonl(input_path)
-        self._print_output_results(input_data, test_output, test_output_custom, test_output_error)
-        shutil.rmtree(self._tmp_path)
-
-    def _print_output_results(self, input_data, test_output, test_output_custom, test_output_error):
-        """Call the print methods that correspond to the output type"""
-        if not test_output_error:
-            self._print_transformed_events(input_data, test_output, test_output_custom)
-        if self._full_output and test_output_custom and not test_output_error:
-            self._print_pseudonyms(test_output_custom)
-            self._print_predetections(test_output_custom)
-        if test_output_error:
-            self._print_errors(test_output_error)
-
-    def _print_transformed_events(self, input_data, test_output, test_output_custom):
-        """
-        Print the differences between input and output event as well as corresponding pre-detections
-        """
         transformed_cnt = 0
-        for idx, test_item in enumerate(test_output):
-            test_copy = deepcopy(test_item)
-            input_copy = deepcopy(input_data[idx])
-            difference = recursive_compare(test_copy, input_copy)
-            if difference:
-                test_json = json.dumps(test_item, sort_keys=True, indent=4)
-                input_path_json = json.dumps(input_data[idx], sort_keys=True, indent=4)
-                diff = ndiff(input_path_json.splitlines(), test_json.splitlines())
-                color_print_title(Back.CYAN, "PROCESSED EVENT")
-                self._print_ndiff_items(diff)
+        output_count = 0
+        for input_document in self._input_documents:
+            test_output, test_output_custom = self._pipeline.process_pipeline()
+            if test_output:
+                output_count += 1
+            diff = self._print_output_results(input_document, test_output, test_output_custom)
+            if diff:
                 transformed_cnt += 1
+        color_print_title(Back.WHITE, f"TRANSFORMED EVENTS: {transformed_cnt}/{output_count}")
+        shutil.rmtree(self._tmp_path)
 
-            for test_item_custom in test_output_custom:
-                detector_id = test_item_custom.get("pre_detection_id")
-                if detector_id and detector_id == test_item.get("pre_detection_id"):
-                    color_print_title(Back.YELLOW, "PRE-DETECTION FOR PRECEDING EVENT")
-                    test_json_custom = json.dumps(test_item_custom, sort_keys=True, indent=4)
-                    color_print_line(Back.BLACK, Fore.YELLOW, test_json_custom)
-        color_print_title(Back.WHITE, f"TRANSFORMED EVENTS: {transformed_cnt}/{len(test_output)}")
+    def _print_output_results(self, input_document, test_output, test_output_custom):
+        test_copy = deepcopy(test_output)
+        input_copy = deepcopy(input_document)
+        difference = recursive_compare(test_copy, input_copy)
+        if difference:
+            test_json = json.dumps(test_output, sort_keys=True, indent=4)
+            input_path_json = json.dumps(input_document, sort_keys=True, indent=4)
+            diff = ndiff(input_path_json.splitlines(), test_json.splitlines())
+            color_print_title(Back.CYAN, "PROCESSED EVENT")
+            self._print_ndiff_items(diff)
+        if self._full_output and test_output_custom:
+            self._print_custom_outputs(test_output_custom)
+        return difference
 
     def _print_ndiff_items(self, diff):
         """
@@ -199,35 +135,10 @@ def _print_ndiff_items(self, diff):
             else:
                 color_print_line(Back.BLACK, Fore.CYAN, item)
 
-    def _print_pseudonyms(self, test_output_custom):
-        """Print only the pseudonyms from all custom outputs"""
-        color_print_title(Back.MAGENTA, "ALL PSEUDONYMS")
-        for test_item in test_output_custom:
-            if "pseudonym" in test_item.keys() and "origin" in test_item.keys():
-                test_json = json.dumps(test_item, sort_keys=True, indent=4)
-                color_print_line(Back.BLACK, Fore.MAGENTA, test_json)
-
-    def _print_predetections(self, test_output_custom):
-        """Print only the pre-detections from all custom outputs"""
-        color_print_title(Back.YELLOW, "ALL PRE-DETECTIONS")
-        for test_item in test_output_custom:
-            if "pre_detection_id" in test_item.keys():
-                test_json = json.dumps(test_item, sort_keys=True, indent=4)
+    def _print_custom_outputs(self, test_output_custom):
+        color_print_title(Back.MAGENTA, "CUSTOM OUTPUTS")
+        for output, output_target in test_output_custom:
+            color_print_title(Back.YELLOW, f"Output Target: {str(output_target[0])}")
+            for out in output:
+                test_json = json.dumps(out, sort_keys=True, indent=4)
                 color_print_line(Back.BLACK, Fore.YELLOW, test_json)
-
-    def _print_errors(self, test_output_error):
-        """Print all errors"""
-        for test_items in test_output_error:
-            color_print_title(Back.RED, "ERROR")
-
-            json_message = test_items.get("error_message")
-            color_print_line(Back.BLACK, Fore.RED, json_message)
-
-            json_original = test_items.get("document_received")
-            json_processed = test_items.get("document_processed")
-
-            diff = ndiff(str(json_original), str(json_processed))
-            color_print_title(Back.YELLOW, "PARTIALLY PROCESSED EVENT")
-            self._print_ndiff_items(diff)
-        log_message = "^^^ RESULTS CAN NOT BE SHOWN UNTIL ALL ERRORS HAVE BEEN FIXED ^^^"
-        color_print_line(Back.RED, Fore.WHITE, log_message)


Test patch:
diff --git a/logprep/util/auto_rule_tester/auto_rule_corpus_tester.py b/logprep/util/auto_rule_tester/auto_rule_corpus_tester.py
index d81d18d33..e8a6ac1f5 100644
--- a/logprep/util/auto_rule_tester/auto_rule_corpus_tester.py
+++ b/logprep/util/auto_rule_tester/auto_rule_corpus_tester.py
@@ -88,6 +88,7 @@
 import tempfile
 from functools import cached_property
 from json import JSONDecodeError
+from logging import getLogger
 from pathlib import Path
 from pprint import pprint
 from typing import List
@@ -159,6 +160,7 @@ def _pipeline(self):
             self._original_config_path, self._tmp_dir, str(merged_input_file_path)
         )
         config = Configuration.create_from_yaml(path_to_patched_config)
+        config.verify_pipeline_without_processor_outputs(getLogger("logprep"))
         del config["output"]
         pipeline = Pipeline(config=config)
         return pipeline
diff --git a/tests/acceptance/util.py b/tests/acceptance/util.py
index b2d3113a8..5e2287c6f 100644
--- a/tests/acceptance/util.py
+++ b/tests/acceptance/util.py
@@ -21,9 +21,10 @@
 
 from logprep.abc.processor import Processor
 from logprep.registry import Registry
+from logprep.runner import Runner
 from logprep.util.decorators import timeout
-from logprep.util.helper import recursive_compare
-from logprep.util.rule_dry_runner import get_patched_runner, get_runner_outputs
+from logprep.util.helper import recursive_compare, remove_file_if_exists
+from logprep.util.json_handling import parse_jsonl
 from tests.unit.processor.base import BaseProcessorTestCase
 
 basicConfig(level=DEBUG, format="%(asctime)-15s %(name)-5s %(levelname)-8s: %(message)s")
@@ -112,6 +113,72 @@ def store_latest_test_output(target_output_identifier, output_of_test):
             latest_output.write(json.dumps(test_output_line) + "\n")
 
 
+def get_runner_outputs(patched_runner) -> list:
+    # pylint: disable=protected-access
+    """
+    Extracts the outputs of a patched logprep runner.
+
+    Parameters
+    ----------
+    patched_runner : Runner
+        The patched logprep runner
+
+    Returns
+    -------
+    parsed_outputs : list
+        A list of logprep outputs containing events, extra outputs like pre-detections or pseudonyms
+        and errors
+    """
+    parsed_outputs = [None, None, None]
+    output_config = list(patched_runner._configuration.get("output").values())[0]
+    output_paths = [
+        output_path for key, output_path in output_config.items() if "output_file" in key
+    ]
+
+    for output_path in output_paths:
+        remove_file_if_exists(output_path)
+
+    patched_runner.start()
+
+    for index, output_path in enumerate(output_paths):
+        parsed_outputs[index] = parse_jsonl(output_path)
+        remove_file_if_exists(output_path)
+
+    return parsed_outputs
+
+
+def get_patched_runner(config_path, logger):
+    """
+    Creates a patched runner that bypasses check to obtain non singleton instance and the runner
+    won't continue iterating on an empty pipeline.
+
+    Parameters
+    ----------
+    config_path : str
+        The logprep configuration that should be used for the patched runner
+    logger : Logger
+        The application logger the runner should use
+
+    Returns
+    -------
+    runner : Runner
+        The patched logprep runner
+    """
+    runner = Runner(bypass_check_to_obtain_non_singleton_instance=True)
+    runner.set_logger(logger)
+    runner.load_configuration(config_path)
+
+    # patch runner to stop on empty pipeline
+    def keep_iterating():
+        """generator that stops on first iteration"""
+        return  # nosemgrep
+        yield
+
+    runner._keep_iterating = keep_iterating  # pylint: disable=protected-access
+
+    return runner
+
+
 def get_test_output(config_path):
     patched_runner = get_patched_runner(config_path, logger)
     return get_runner_outputs(patched_runner=patched_runner)
diff --git a/tests/unit/util/test_configuration.py b/tests/unit/util/test_configuration.py
index 02eeb192e..68b0d44a3 100644
--- a/tests/unit/util/test_configuration.py
+++ b/tests/unit/util/test_configuration.py
@@ -659,7 +659,7 @@ def test_verify_errors_get_collected(self, config_dict, raised_errors, test_case
         config = deepcopy(self.config)
         config.update(config_dict)
         if raised_errors is not None:
-            errors = config._perform_verfification_and_get_errors(logger)
+            errors = config._check_for_errors(logger)
             collected_errors = []
             for error in errors:
                 collected_errors += error.errors
@@ -947,3 +947,22 @@ def test_verifies_processor_configs_against_defined_outputs(self):
         assert len(raised.value.errors) == 3
         for error in raised.value.errors:
             assert "output 'kafka' does not exist in logprep outputs" in error.args[0]
+
+    def test_verify_pipeline_without_processor_outputs_ignores_processor_output_errors(self):
+        config = Configuration()
+        pipeline = [
+            {
+                "pd": {
+                    "type": "pre_detector",
+                    "generic_rules": ["tests/testdata/unit/pre_detector/rules/generic"],
+                    "specific_rules": ["tests/testdata/unit/pre_detector/rules/specific"],
+                    "outputs": [{"kafka": "pre_detector_alerts"}],
+                    "alert_ip_list_path": "tests/testdata/unit/pre_detector/alert_ips.yml",
+                }
+            },
+        ]
+        config.update({"pipeline": pipeline, "output": {}})
+        try:
+            config.verify_pipeline_without_processor_outputs(logger=logger)
+        except InvalidConfigurationErrors as error:
+            assert False, f"Shouldn't raise output does not exist error: '{error}'"
diff --git a/tests/unit/util/test_rule_dry_runner.py b/tests/unit/util/test_rule_dry_runner.py
index 36d4a8e4c..362165e55 100644
--- a/tests/unit/util/test_rule_dry_runner.py
+++ b/tests/unit/util/test_rule_dry_runner.py
@@ -1,9 +1,9 @@
 # pylint: disable=missing-docstring
-# pylint: disable=no-self-use
 # pylint: disable=attribute-defined-outside-init
 import json
 import logging
 import os
+import re
 import tempfile
 from unittest import mock
 
@@ -68,7 +68,7 @@ def test_dry_run_accepts_json_as_input(self, tmp_path, capsys):
             json.dump(test_json, input_file)
 
         dry_runner = DryRunner(
-            dry_run=input_json_file,
+            input_file_path=input_json_file,
             config_path=self.config_path,
             full_output=True,
             use_json=True,
@@ -87,7 +87,7 @@ def test_dry_run_accepts_json_in_list_as_input(self, tmp_path, capsys):
             json.dump(test_json, input_file)
 
         dry_runner = DryRunner(
-            dry_run=input_json_file,
+            input_file_path=input_json_file,
             config_path=self.config_path,
             full_output=True,
             use_json=True,
@@ -109,7 +109,7 @@ def test_dry_run_accepts_jsonl_as_input(self, tmp_path, capsys):
             input_file.writelines(test_jsonl)
 
         dry_runner = DryRunner(
-            dry_run=input_jsonl_file,
+            input_file_path=input_jsonl_file,
             config_path=self.config_path,
             full_output=True,
             use_json=False,
@@ -135,7 +135,7 @@ def test_dry_run_print_custom_output(self, tmp_path, capsys):
             json.dump(test_json, input_file)
 
         dry_runner = DryRunner(
-            dry_run=input_json_file,
+            input_file_path=input_json_file,
             config_path=self.config_path,
             full_output=True,
             use_json=True,
@@ -146,8 +146,7 @@ def test_dry_run_print_custom_output(self, tmp_path, capsys):
         captured = capsys.readouterr()
         assert "------ PROCESSED EVENT ------" in captured.out
         assert "------ TRANSFORMED EVENTS: 1/1 ------" in captured.out
-        assert "------ ALL PSEUDONYMS ------" in captured.out
-        assert "------ ALL PRE-DETECTIONS ------" in captured.out
+        assert "------ CUSTOM OUTPUTS ------" in captured.out
 
     def test_dry_run_prints_predetection(self, tmp_path, capsys):
         test_json = {
@@ -161,7 +160,7 @@ def test_dry_run_prints_predetection(self, tmp_path, capsys):
             json.dump(test_json, input_file)
 
         dry_runner = DryRunner(
-            dry_run=input_json_file,
+            input_file_path=input_json_file,
             config_path=self.config_path,
             full_output=True,
             use_json=True,
@@ -172,8 +171,7 @@ def test_dry_run_prints_predetection(self, tmp_path, capsys):
         captured = capsys.readouterr()
         assert "------ PROCESSED EVENT ------" in captured.out
         assert "------ TRANSFORMED EVENTS: 1/1 ------" in captured.out
-        assert "------ ALL PSEUDONYMS ------" in captured.out
-        assert "------ ALL PRE-DETECTIONS ------" in captured.out
+        assert "------ CUSTOM OUTPUTS ------" in captured.out
 
     @mock.patch("logprep.processor.labeler.processor.Labeler.process", side_effect=BaseException)
     def test_dry_run_prints_errors(self, _, tmp_path, capsys):
@@ -188,7 +186,7 @@ def test_dry_run_prints_errors(self, _, tmp_path, capsys):
             json.dump(test_json, input_file)
 
         dry_runner = DryRunner(
-            dry_run=input_json_file,
+            input_file_path=input_json_file,
             config_path=self.config_path,
             full_output=True,
             use_json=True,
@@ -197,4 +195,4 @@ def test_dry_run_prints_errors(self, _, tmp_path, capsys):
         dry_runner.run()
 
         captured = capsys.readouterr()
-        assert "------ ERROR ------" in captured.out
+        assert not re.match(r".*A critical error occured for processor Labeler", captured.err)