diff --git a/oemof_b3/tools/testing_pipeline.py b/oemof_b3/tools/testing_pipeline.py index 53b0bce4..364523b5 100644 --- a/oemof_b3/tools/testing_pipeline.py +++ b/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 diff --git a/tests/test_pipeline_results.py b/tests/test_pipeline_results.py index 6c7c33d4..f7e64b12 100644 --- a/tests/test_pipeline_results.py +++ b/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)