# Copyright 2022-2025 The Ramble Authors
#
# Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
# https://www.apache.org/licenses/LICENSE-2.0> or the MIT license
# <LICENSE-MIT or https://opensource.org/licenses/MIT>, at your
# option. This file may not be copied, modified, or distributed
# except according to those terms.
"""Perform tests of the Application class"""
import pytest
import ramble.workload
import ramble.workspace
pytestmark = pytest.mark.usefixtures(
"mutable_config", "mutable_mock_workspace_path", "mutable_mock_apps_repo"
)
[docs]
@pytest.mark.parametrize(
"app", ["basic", "basic-inherited", "input-test", "interleved-env-vars", "register-builtin"]
)
def test_app_features(mutable_mock_apps_repo, app):
app_inst = mutable_mock_apps_repo.get(app)
assert hasattr(app_inst, "workloads")
assert hasattr(app_inst, "workload_groups")
assert hasattr(app_inst, "executables")
assert hasattr(app_inst, "figures_of_merit")
assert hasattr(app_inst, "inputs")
assert hasattr(app_inst, "compilers")
assert hasattr(app_inst, "software_specs")
assert hasattr(app_inst, "required_packages")
assert hasattr(app_inst, "builtins")
[docs]
def test_basic_app(mutable_mock_apps_repo):
basic_inst = mutable_mock_apps_repo.get("basic")
assert "test_wl" in basic_inst.workloads
assert len(basic_inst.workloads["test_wl"].executables) == 1
foo_exec = basic_inst.workloads["test_wl"].find_executable("foo")
assert foo_exec is not None
foo_exec = basic_inst.executables[foo_exec]
assert foo_exec.template == ["bar"]
assert not foo_exec.mpi
assert len(basic_inst.workloads["test_wl"].inputs) == 1
example_input = basic_inst.workloads["test_wl"].find_input("input")
assert example_input is not None
assert len(basic_inst.workloads["test_wl"].variables) == 2
my_var = basic_inst.workloads["test_wl"].find_variable("my_var")
assert my_var is not None
assert my_var.default == "1.0"
assert my_var.description == "Example var"
assert "test_wl2" in basic_inst.workloads
assert len(basic_inst.workloads["test_wl2"].executables) == 1
bar_exec = basic_inst.workloads["test_wl2"].find_executable("bar")
assert bar_exec is not None
bar_exec = basic_inst.executables[bar_exec]
assert bar_exec.template == ["baz"]
assert bar_exec.mpi
assert len(basic_inst.workloads["test_wl2"].inputs) == 1
example_input = basic_inst.workloads["test_wl2"].find_input("input")
assert example_input is not None
exec_graph = basic_inst._get_executable_graph("test_wl")
assert exec_graph.get_node("foo") is not None
assert exec_graph.get_node("builtin::env_vars") is not None
exec_graph = basic_inst._get_executable_graph("test_wl2")
assert exec_graph.get_node("bar") is not None
assert exec_graph.get_node("builtin::env_vars") is not None
assert "test_fom" in basic_inst.figures_of_merit
fom_conf = basic_inst.figures_of_merit["test_fom"]
assert fom_conf["log_file"] == "{log_file}"
assert fom_conf["regex"] == r"(?P<test>[0-9]+\.[0-9]+).*seconds.*" # noqa: W605
assert fom_conf["group_name"] == "test"
assert fom_conf["units"] == "s"
assert "input" in basic_inst.inputs
assert basic_inst.inputs["input"]["url"] == "file:///tmp/test_file.log"
assert basic_inst.inputs["input"]["description"] == "Not a file"
[docs]
@pytest.mark.parametrize("app_name", ["basic", "zlib"])
def test_application_copy_is_deep(mutable_mock_apps_repo, app_name):
src_inst = mutable_mock_apps_repo.get(app_name)
defined_variables = {"test_var1": "test_val1", "test_var2": "test_val2"}
defined_env_vars = {
"set": {"SET_ENV_VAR": "TEST"},
"unset": ["UNSET_ENV_VAR"],
"append": [{"var-separator": ",", "vars": {"APPEND_VAR": "APPEND_TEST"}}],
"prepend": [{"var-separator": ",", "vars": {"PREPEND_VAR": "PREPEND_TEST"}}],
}
defined_internals = {
"custom_executables": {
"test_exec": {"templates": ["test_exec"], "use_mpi": False, "redirect": "{log_file}"}
}
}
src_inst.set_variables(defined_variables, None)
src_inst.set_env_variable_sets(defined_env_vars)
src_inst.set_internals(defined_internals)
copy_inst = src_inst.copy()
# Test variables
for var, val in src_inst.variables.items():
assert var in copy_inst.variables.keys()
assert copy_inst.variables[var] == val
# Test env-vars
for var_set in src_inst._env_variable_sets.keys():
assert var_set in copy_inst._env_variable_sets.keys()
# Test set sets
if var_set == "set":
for var, val in src_inst._env_variable_sets[var_set].items():
assert var in copy_inst._env_variable_sets[var_set]
assert copy_inst._env_variable_sets[var_set][var] == val
elif var_set == "append" or var_set == "prepend":
for idx, set_group in enumerate(src_inst._env_variable_sets[var_set]):
if "var-separator" in set_group:
assert "var-separator" in copy_inst._env_variable_sets[var_set][idx]
assert (
copy_inst._env_variable_sets[var_set][idx]["var-separator"]
== set_group["var-separator"]
)
if "vars" in set_group:
assert "vars" in copy_inst._env_variable_sets[var_set][idx]
for var, val in set_group["vars"].items():
assert var in copy_inst._env_variable_sets[var_set][idx]["vars"]
assert copy_inst._env_variable_sets[var_set][idx]["vars"][var] == val
elif var_set == "unset":
for var in src_inst._env_variable_sets[var_set]:
assert var in copy_inst._env_variable_sets[var_set]
# Test internals:
for internal, conf in src_inst.internals.items():
assert internal in copy_inst.internals
if internal == "custom_executables":
for exec_name, exec_conf in conf.items():
assert exec_name in copy_inst.internals[internal]
for option, value in exec_conf.items():
assert option in copy_inst.internals[internal][exec_name]
assert copy_inst.internals[internal][exec_name][option] == value
[docs]
@pytest.mark.parametrize(
"app", ["basic", "basic-inherited", "input-test", "interleved-env-vars", "register-builtin"]
)
def test_required_builtins(mutable_mock_apps_repo, app):
app_inst = mutable_mock_apps_repo.get(app)
required_builtins = []
for builtin, conf in app_inst.builtins.items():
if conf[app_inst._builtin_required_key]:
required_builtins.append(builtin)
for workload in app_inst.workloads.keys():
exec_graph = app_inst._get_executable_graph(workload)
for builtin in required_builtins:
assert exec_graph.get_node(builtin) is not None
[docs]
def test_register_builtin_app(mutable_mock_apps_repo):
app_inst = mutable_mock_apps_repo.get("register-builtin")
required_builtins = []
excluded_builtins = []
for builtin, conf in app_inst.builtins.items():
if conf[app_inst._builtin_required_key]:
required_builtins.append(builtin)
else:
excluded_builtins.append(builtin)
for workload in app_inst.workloads.keys():
exec_graph = app_inst._get_executable_graph(workload)
for builtin in required_builtins:
assert exec_graph.get_node(builtin) is not None
for builtin in excluded_builtins:
assert exec_graph.get_node(builtin) is None
# Test for dependency injection
found_prerequisite = False
for node in exec_graph.walk():
if node.key == "builtin::test_builtin":
break
if node.key == "builtin::test_builtin_pre":
found_prerequisite = True
assert found_prerequisite
[docs]
@pytest.mark.parametrize(
"app", ["basic", "basic-inherited", "input-test", "interleved-env-vars", "register-builtin"]
)
def test_short_print(mutable_mock_apps_repo, app):
app_inst = mutable_mock_apps_repo.get(app)
app_inst._verbosity = "short"
str_val = str(app_inst)
assert str_val == app
[docs]
def basic_exp_dict():
"""To set expander consistently with test_wl2 of builtin.mock/applications/basic"""
return {
"application_name": "bar",
"inputs": {"test_wl": "input", "test_wl2": "input"},
"workload_name": "test_wl2",
"experiment_name": "baz",
"application_input_dir": "/workspace/inputs/bar",
"workload_input_dir": "/workspace/inputs/bar/test_wl2",
"application_run_dir": "/workspace/experiments/bar",
"workload_run_dir": "/workspace/experiments/bar/test_wl2",
"experiment_run_dir": "/workspace/experiments/bar/test_wl2/baz",
"env_name": "spack_bar.test_wl2",
"n_ranks": "4",
"processes_per_node": "2",
"n_nodes": "2",
"var1": "{var2}",
"var2": "{var3}",
"var3": "3",
"mpi_command": "mpirun -n {n_ranks}",
"batch_command": "sbatch -p {partition} {execute_experiment}",
}
[docs]
def test_get_executable_graph_initial(mutable_mock_apps_repo):
"""_get_executable_graph, test1, workload executables"""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
# Set up the instance to test just the initial part of the function
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
test_wl = ramble.workload.Workload("test_wl", executables=["foo"], inputs=["input"])
test_wl2 = ramble.workload.Workload("test_wl2", executables=["bar"], inputs=["input"])
executable_application_instance.workloads = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_graph = executable_application_instance._get_executable_graph("test_wl2")
bar_node = executable_graph.get_node("bar")
assert bar_node is not None
[docs]
def test_get_executable_graph_yaml_defined(mutable_mock_apps_repo):
"""_get_executable_graph, test2, yaml-defined order"""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
# Set up the instance to pass the initial part of the function
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
test_wl = ramble.workload.Workload("test_wl", executables=["foo"], inputs=["input"])
test_wl2 = ramble.workload.Workload("test_wl2", executables=["bar"], inputs=["input"])
executable_application_instance.workloads = {"test_wl": test_wl, "test_wl2": test_wl2}
# Insert namespace.executables into the instance's internals to pass the
# second part of the function
defined_internals = {
"custom_executables": {
"test_exec": {"template": ["test_exec"], "use_mpi": False, "redirect": "{log_file}"}
},
"executables": ["bar", "test_exec"],
}
executable_application_instance.set_internals(defined_internals)
executable_graph = executable_application_instance._get_executable_graph("test_wl")
test_node = executable_graph.get_node("test_exec")
assert test_node is not None
[docs]
def test_get_executable_graph_custom_executables(mutable_mock_apps_repo):
"""_get_executable_graph, test3, custom executables"""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
# Set up the instance to pass the initial part of the function
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
test_wl = ramble.workload.Workload("test_wl", executables=["foo"], inputs=["input"])
test_wl2 = ramble.workload.Workload("test_wl2", executables=["bar"], inputs=["input"])
executable_application_instance.workloads = {"test_wl": test_wl, "test_wl2": test_wl2}
# Insert namespace.executables into the instance's internals to pass the
# second part of the function
defined_internals = {
"custom_executables": {
"test_exec2": {
"template": ["test_exec2"],
"use_mpi": False,
"redirect": "{log_file}",
}
},
"executables": ["test_exec2", "bar"],
}
executable_application_instance.set_internals(defined_internals)
executable_graph = executable_application_instance._get_executable_graph("test_wl2")
test_node = executable_graph.get_node("test_exec2")
assert test_node is not None
[docs]
def test_set_default_experiment_variables(mutable_mock_apps_repo):
"""_set_default_experiment_variables"""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
# Set up the instance to pass the initial part of the function
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
test_wl = ramble.workload.Workload("test_wl", executables=["foo"], inputs=["input"])
test_wl2 = ramble.workload.Workload("test_wl2", executables=["bar"], inputs=["input"])
test_wl2.add_variable(ramble.workload.WorkloadVariable("n_ranks", default="1"))
executable_application_instance.workloads = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_application_instance.inputs = {"input": {"target_dir": "."}}
executable_application_instance.variables = {}
executable_application_instance._set_default_experiment_variables()
assert executable_application_instance.variables["n_ranks"] == "1"
[docs]
def test_define_commands(mutable_mock_apps_repo):
"""test _define_commands"""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
# Set up the instance to pass the initial part of the function
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
test_wl = ramble.workload.Workload("test_wl", executables=["foo"], inputs=["input"])
test_wl2 = ramble.workload.Workload("test_wl2", executables=["bar"], inputs=["input"])
test_wl2.add_variable(ramble.workload.WorkloadVariable("n_ranks", default="1"))
executable_application_instance.workloads = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_application_instance.inputs = {"input": {"target_dir": "."}}
executable_application_instance.variables = {}
exec_graph = executable_application_instance._get_executable_graph("test_wl2")
executable_application_instance.set_formatted_executables(
{"command": {"join_separator": "\n"}}
)
executable_application_instance._set_default_experiment_variables()
executable_application_instance.chain_prepend = []
executable_application_instance._define_commands(exec_graph)
executable_application_instance._define_formatted_executables()
assert "mpirun" in executable_application_instance.variables["command"]
[docs]
def test_derive_variables_for_template_path(mutable_mock_apps_repo):
"""_set_default_variables_for_template_path"""
test_config = """
ramble:
variables:
mpi_command: 'mpirun -n {n_ranks} -ppn {processes_per_node}'
batch_submit: 'batch_submit {execute_experiment}'
processes_per_node: '5'
n_ranks: '{processes_per_node}*{n_nodes}'
applications:
basic:
workloads:
test_wl:
experiments:
test_experiment:
template: true
variables:
n_nodes: '2'
test_wl2:
experiments:
test_experiment:
variables:
n_nodes: '2'
software:
packages: {}
environments: {}
"""
import os.path
workspace_name = "test_derive_variables_for_template_path"
ws1 = ramble.workspace.create(workspace_name)
ws1.write()
config_path = os.path.join(ws1.config_dir, ramble.workspace.config_file_name)
with open(config_path, "w+") as f:
f.write(test_config)
ws1._re_read()
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
# Set up the instance to pass the initial part of the function
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
test_wl = ramble.workload.Workload("test_wl", executables=["foo"], inputs=["input"])
test_wl2 = ramble.workload.Workload("test_wl2", executables=["bar"], inputs=["input"])
test_wl2.add_variable(ramble.workload.WorkloadVariable("n_ranks", default="1"))
executable_application_instance.workloads = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_application_instance.inputs = {"input": {"target_dir": "."}}
executable_application_instance.variables = {}
exec_graph = executable_application_instance._get_executable_graph("test_wl2")
executable_application_instance._set_default_experiment_variables()
executable_application_instance.chain_prepend = []
executable_application_instance._define_commands(exec_graph)
executable_application_instance._define_formatted_executables()
test_answer = "/workspace/experiments/bar/test_wl2/baz/execute_experiment"
executable_application_instance._derive_variables_for_template_path(ws1)
assert executable_application_instance.variables["execute_experiment"] == test_answer
[docs]
def test_class_attributes(mutable_mock_apps_repo):
basic_inst = mutable_mock_apps_repo.get("basic")
basic_copy = basic_inst.copy()
instances = [basic_inst, basic_copy]
for inst in instances:
assert hasattr(inst, "workloads")
assert "test_wl" in inst.workloads
basic_copy.workload("added_workload", executables=["foo"])
assert "added_workload" in basic_copy.workloads
assert "added_workload" not in basic_inst.workloads
[docs]
def test_workload_groups(mutable_mock_apps_repo):
workload_group_inst = mutable_mock_apps_repo.get("workload-groups")
assert "test_wl" in workload_group_inst.workloads
assert "empty" in workload_group_inst.workload_groups
assert "test_wlg" in workload_group_inst.workload_groups
my_var = workload_group_inst.workloads["test_wl"].find_variable("test_var")
assert my_var is not None
assert my_var.default == "2.0"
assert my_var.description == "Test workload vars and groups"
my_mixed_var_wl = workload_group_inst.workloads["test_wl"].find_variable("test_var_mixed")
assert my_mixed_var_wl is not None
assert my_mixed_var_wl.default == "3.0"
assert my_mixed_var_wl.description == "Test vars for workload and groups"
[docs]
def test_workload_groups_inherited(mutable_mock_apps_repo):
wlgi_inst = mutable_mock_apps_repo.get("workload-groups-inherited")
assert "test_wl" in wlgi_inst.workloads
assert "test_wl3" in wlgi_inst.workloads
# check we inherit groups we don't touch
assert "empty" in wlgi_inst.workload_groups
assert "test_wlg" in wlgi_inst.workload_groups
assert "test_wl" in wlgi_inst.workload_groups["test_wlg"]
# Ensure a new workload can obtain the parent level vars via groups
my_var = wlgi_inst.workloads["test_wl3"].find_variable("test_var")
assert my_var is not None
assert my_var.default == "2.0"
assert my_var.description == "Test workload vars and groups"
for wl in ["test_wl", "test_wl3"]:
my_mixed_var_wl = wlgi_inst.workloads[wl].find_variable("test_var_mixed")
assert my_mixed_var_wl is not None
assert my_mixed_var_wl.default == "3.0"