Add function test_pipeline_folders to test_pipeline_results.py and ad…

…just code in testing_pipeline.py

oemof_b3/tools/testing_pipeline.py

@@ -69,6 +69,7 @@ def rename_path(file_path, before, after):
     """
     This function renames existing files in directories by appending
     the suffix "_original" to their filenames.
+    TODO: Update docstring
 
     Inputs
     -------
@@ -88,8 +89,13 @@ def rename_path(file_path, before, after):
     # Split the path and file name
     directory, filename = os.path.split(file_path)
 
-    # Add suffix "_original" before the file extension
-    new_filename = filename.replace(before, after)
+    # Determine new name based on whether it's a file or directory
+    if os.path.isfile(file_path):
+        # Add suffix "_original" before the file extension
+        new_filename = filename.replace(before, after)
+    else:
+        # Add suffix "_original" to the directory name
+        new_filename = filename + "_original"
 
     # Join the directory and new filename to get the new path
     new_file_path = os.path.join(directory, new_filename)
@@ -154,6 +160,7 @@ def remove_raw_data_created(exists):
 def remove_test_data(path):
     """
     This function removes test data.
+    TODO: Update docstring
 
     Inputs
     -------
@@ -167,6 +174,8 @@ def remove_test_data(path):
     """
     if os.path.isfile(path):
         os.remove(path)
+    elif os.path.isdir(path):
+        shutil.rmtree(path)
 
 
 def get_abs_path_list(output_rule_list):
@@ -237,15 +246,21 @@ def rule_test(sublist):
     )
 
     # Check if snakemake rule exited without error (true)
-    assert output, f"Snakemake rule failed for targets: {sublist}"
+    assert output
 
     # Log the success
     logging.info(f"Snakemake rule executed successfully for targets: {sublist}")
 
 
+def remove_extension(before, after):
+    # TODO: Update docstring
+    shutil.move(before, after)
+
+
 def clean_file(sublist, delete_switch, renamed_path):
     """
     This function removes test data files and reverts renamed files.
+    TODO: Update docstring
 
     Inputs
     -------
@@ -277,6 +292,9 @@ def clean_file(sublist, delete_switch, renamed_path):
                 "_original." + file_extension,
                 "." + file_extension,
             )
+        else:
+            original_path = renamed_file.partition("_original")[0]
+            remove_extension(renamed_file, original_path)
 
 
 def pipeline_file_output_test(delete_switch, output_rule_list):
@@ -308,27 +326,11 @@ def pipeline_file_output_test(delete_switch, output_rule_list):
 
         renamed_path = []
         for raw_file_path in absolute_path_list:
-            try:
-                # Check if file already exists in directory
-                if os.path.isfile(raw_file_path):
-                    # Rename file with extension original
-                    renamed_file = file_name_extension(raw_file_path)
-                    renamed_path.append(renamed_file)
-                else:
-                    # Check for the file with the _original suffix
-                    original_file_path = raw_file_path.replace(
-                        os.path.splitext(raw_file_path)[1],
-                        "_original" + os.path.splitext(raw_file_path)[1],
-                    )
-                    if os.path.exists(original_file_path):
-                        raise FileExistsError(
-                            f"File {original_file_path} already exists."
-                            f"Please rename the file {raw_file_path} first."
-                        )
-
-            except FileNotFoundError as e:
-                print(e)
-                continue
+            # Check if file already exists in directory
+            if os.path.isfile(raw_file_path):
+                # Rename file with extension original
+                renamed_file = file_name_extension(raw_file_path)
+                renamed_path.append(renamed_file)
 
         try:
             # Run the snakemake rule

tests/test_pipeline_results.py

@@ -1,6 +1,28 @@
+"""
+
+"""
+import os
+from oemof_b3.tools.testing_pipeline import (
+    get_repo_path,
+    get_abs_path_list,
+    rename_path,
+    file_name_extension,
+    clean_file,
+    rule_test,
+)
+
+
 # Delete data from test run of pipeline if True otherwise False
 delete_switch = True
 
+# Get current path
+current_path = os.path.abspath(os.getcwd())
+target_path = get_repo_path(current_path)
+
+# Set the current path to the target path
+os.chdir(target_path)
+
+
 scenarios = [
     "2050-100-el_eff",
     "2050-95-el_eff",
@@ -17,20 +39,25 @@
     "all-optimized",
 ]
 
-for scenario in scenarios:
 
+def output_rule_set(scenario):
     output_rule_list = [
-        "results/" + scenario + "/preprocessed",
-        "results/" + scenario + "/optimized",
-        "results/" + scenario + "/postprocessed",
-        "results/" + scenario + "/b3_results/data",
-        "results/" + scenario + "/tables",
-        "results/" + scenario + "/plotted/dispatch",
-        "results/" + scenario + "/plotted/storage_level",
-        "results/" + scenario + "/plotted/scalars",
-        "results/" + scenario + "/report",
+        [
+            "results/" + scenario + "/preprocessed",
+            "results/" + scenario + "/optimized",
+            "results/" + scenario + "/postprocessed",
+            "results/" + scenario + "/b3_results/data",
+            "results/" + scenario + "/tables",
+            "results/" + scenario + "/plotted/dispatch",
+            "results/" + scenario + "/plotted/storage_level",
+            "results/" + scenario + "/plotted/scalars",
+        ]
+        # "results/" + scenario + "/report",
     ]
 
+    return output_rule_list
+
+
 for scenario_group in scenario_groups:
     output_rule_list = [
         "results/joined_scenarios/" + scenario_group + "/joined",
@@ -40,3 +67,67 @@
 
 # snakemake -j1 prepare_re_potential
 output_rule_list = ["results/_resources/RE_potential"]
+
+
+def test_pipeline_folders(delete_switch, scenarios):
+    # Get output rule set from scenario
+    for scenario in scenarios:
+        output_rule_list = output_rule_set(scenario)
+
+        for sublist in output_rule_list:
+            absolute_path_list = get_abs_path_list(sublist)
+
+        renamed_file_path = []
+        for raw_dir_path in absolute_path_list:
+            try:
+                # Check if file already exists in directory
+                if os.path.isfile(raw_dir_path):
+                    # Rename file with extension original
+                    renamed_file = file_name_extension(raw_dir_path)
+                    renamed_path.append(renamed_file)
+                # Check if file already exists in directory
+                if os.path.isdir(raw_dir_path):
+                    # Rename file with extension original
+                    renamed_file = rename_path(raw_dir_path, "", "")
+                    renamed_file_path.append(renamed_file)
+                else:
+                    # Check for the file with the _original suffix
+                    dir_file = raw_dir_path + "_original"
+
+                    if os.path.exists(dir_file):
+                        raise FileExistsError(
+                            f"File {dir_file} already exists."
+                            f"Please rename the file {raw_dir_path} first."
+                        )
+
+            except FileNotFoundError as e:
+                print(e)
+                continue
+
+        try:
+            # Run the snakemake rule
+            rule_test(sublist)
+
+            # Check if the output file was created
+            for raw_dir_path in absolute_path_list:
+                assert os.path.exists(raw_dir_path)
+
+            # Revert file changes
+            clean_file(sublist, delete_switch, renamed_file_path)
+
+        except BaseException:
+            # Revert file changes
+            clean_file(sublist, delete_switch, renamed_file_path)
+
+            raise AssertionError(
+                f"The workflow {raw_dir_path} could not be executed correctly. "
+                f"Changes were reverted."
+                "\n"
+                f"{absolute_path_list}"
+                "\n"
+                f"{sublist}"
+            )
+
+
+def test_pipeline_results():
+    test_pipeline_folders(delete_switch, scenarios)