Skip to content

Commit

Permalink
Move and rename function test_pipeline_folders
Browse files Browse the repository at this point in the history
  • Loading branch information
@Stefanie08
Stefanie08 committed Jun 5, 2024
1 parent 1541b0c commit f166b25
Show file tree
Hide file tree
Showing 2 changed files with 75 additions and 85 deletions.
62 changes: 62 additions & 0 deletions oemof_b3/tools/testing_pipeline.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -358,3 +358,65 @@ def pipeline_file_output_test(delete_switch, output_rule_list):

# Remove raw data if it has been created. It is needed as input data for the tests
remove_raw_data_created(raw_data_exists)


def pipeline_folder_output_test(delete_switch, output_rule_list):
# Raw data is needed for some rules and therefore is created if missing
raw_data_exists = check_raw_data_exists()

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_file_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}"
)

# Remove raw data if it has been created. It is needed as input data for the tests
remove_raw_data_created(raw_data_exists)
98 changes: 13 additions & 85 deletions tests/test_pipeline_results.py
View file
Original file line number Diff line number Diff line change
@@ -1,22 +1,16 @@
"""
This script checks the snakemake pipeline for target rules creating the scenario results.
"""
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,
remove_raw_data_created,
check_raw_data_exists,
pipeline_folder_output_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)
Expand Down Expand Up @@ -44,16 +38,14 @@

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 + "/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",
]

Expand All @@ -70,72 +62,8 @@ def output_rule_set(scenario):
# snakemake -j1 prepare_re_potential
output_rule_list = ["results/_resources/RE_potential"]


def test_pipeline_folders(delete_switch, scenarios):
# Raw data is needed for some rules and therefore is created if missing
raw_data_exists = check_raw_data_exists()

# 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_file_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}"
)

# Remove raw data if it has been created. It is needed as input data for the tests
remove_raw_data_created(raw_data_exists)
scenario_output_rule_list = [output_rule_set(scenario) for scenario in scenarios]


def test_pipeline_results():
test_pipeline_folders(delete_switch, scenarios)
pipeline_folder_output_test(delete_switch, scenario_output_rule_list)

0 comments on commit f166b25

Please sign in to comment.