# Copyright 2022-2026 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"""
from typing import FrozenSet
import pytest
import ramble.definitions.variables
import ramble.workload
import ramble.workspace
pytestmark = pytest.mark.usefixtures(
"mutable_config", "mutable_mock_workspace_path", "mutable_mock_apps_repo"
)
_FS: FrozenSet[str] = frozenset()
[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]
@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")
exp_dict = basic_exp_dict()
basic_inst.set_variables_and_variants(exp_dict, {}, None)
basic_inst.define_variable("application_name", "basic")
assert "test_wl" in basic_inst.workloads[_FS]
assert len(basic_inst.workloads[_FS]["test_wl"].executables) == 1
foo_exec = basic_inst.workloads[_FS]["test_wl"].find_executable("foo")
assert foo_exec is not None
foo_exec = basic_inst.executables[_FS][foo_exec]
assert foo_exec.template == ["bar"]
assert not foo_exec.mpi
assert len(basic_inst.workloads[_FS]["test_wl"].inputs) == 1
example_input = basic_inst.workloads[_FS]["test_wl"].find_input("input")
assert example_input is not None
assert len(basic_inst.workloads[_FS]["test_wl"].variables[_FS]) == 7
possible_vars = basic_inst.workloads[_FS]["test_wl"].find_variable("my_var")
assert len(possible_vars) == 1
assert possible_vars[0].default == "1.0"
assert possible_vars[0].description == "Example var"
assert "test_wl2" in basic_inst.workloads[_FS]
assert len(basic_inst.workloads[_FS]["test_wl2"].executables) == 1
bar_exec = basic_inst.workloads[_FS]["test_wl2"].find_executable("bar")
assert bar_exec is not None
bar_exec = basic_inst.executables[_FS][bar_exec]
assert bar_exec.template == ["baz"]
assert bar_exec.mpi
assert len(basic_inst.workloads[_FS]["test_wl2"].inputs) == 1
example_input = basic_inst.workloads[_FS]["test_wl2"].find_input("input")
assert example_input is not None
basic_inst.define_variable("workload_name", "test_wl")
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
basic_inst.define_variable("workload_name", "test_wl2")
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[_FS][_FS]
fom_conf = basic_inst.figures_of_merit[_FS][_FS]["test_fom"]
assert fom_conf["log_file"] == "{log_file}"
assert fom_conf["regex"] == r"(?P<test>[0-9]+\.[0-9]+).*seconds.*"
assert fom_conf["group_name"] == "test"
assert fom_conf["units"] == "s"
assert "input" in basic_inst.inputs[_FS]
assert basic_inst.inputs[_FS]["input"]["url"] == "file:///tmp/test_file.log"
assert basic_inst.inputs[_FS]["input"]["description"] == "Not a file"
[docs]
@pytest.mark.parametrize(
"app_name,wl_name",
[
("basic", "test_wl2"),
("zlib", "ensure_installed"),
],
)
def test_application_copy_is_deep(app_name, wl_name, mutable_mock_apps_repo):
src_inst = mutable_mock_apps_repo.get(app_name)
defined_variables = {
"n_nodes": "1",
"n_ranks": "{processes_per_node}*{n_nodes}",
"processes_per_node": "1",
"test_var1": "test_val1",
"test_var2": "test_val2",
"workload_name": wl_name,
}
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_and_variants(defined_variables, {}, None)
src_inst.set_env_variable_sets(defined_env_vars)
src_inst.set_internals(defined_internals)
clone_inst = src_inst.clone()
# Test variables
for var, val in src_inst.variables.items():
assert var in clone_inst.variables.keys()
assert clone_inst.variables[var] == val
# Test env-vars
def _compare_env_var_groups(src_group, clone_group):
for var_set in src_group.keys():
assert var_set in clone_group.keys()
# Test set sets
if var_set == "set":
for var, val in src_group[var_set].items():
assert var in clone_group[var_set]
assert clone_group[var_set][var] == val
elif var_set == "append" or var_set == "prepend":
for idx, set_group in enumerate(src_group[var_set]):
if "var-separator" in set_group:
assert "var-separator" in clone_group[var_set][idx]
assert (
clone_group[var_set][idx]["var-separator"]
== set_group["var-separator"]
)
if "vars" in set_group:
assert "vars" in clone_group[var_set][idx]
for var, val in set_group["vars"].items():
assert var in clone_group[var_set][idx]["vars"]
assert clone_group[var_set][idx]["vars"][var] == val
elif var_set == "unset":
for var in src_group[var_set]:
assert var in clone_group[var_set]
for src_group, clone_group in zip(src_inst._env_variable_sets, clone_inst._env_variable_sets):
_compare_env_var_groups(src_group, clone_group)
# Test internals:
for internal, conf in src_inst.internals.items():
assert internal in clone_inst.internals
if internal == "custom_executables":
for exec_name, exec_conf in conf.items():
assert exec_name in clone_inst.internals[internal]
for option, value in exec_conf.items():
assert option in clone_inst.internals[internal][exec_name]
assert clone_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)
exp_dict = basic_exp_dict()
app_inst.set_variables_and_variants(exp_dict, {}, None)
app_inst.define_variable("application_name", app)
required_builtins = []
for builtin, conf in app_inst.builtins[_FS].items():
if conf[app_inst._builtin_required_key]:
required_builtins.append(builtin)
for workload in app_inst.workloads[_FS].keys():
app_inst.define_variable("workload_name", workload)
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")
exp_dict = basic_exp_dict()
app_inst.set_variables_and_variants(exp_dict, {}, None)
app_inst.define_variable("application_name", "register-builtin")
required_builtins = []
excluded_builtins = []
for builtin, conf in app_inst.builtins[_FS].items():
if conf[app_inst._builtin_required_key]:
required_builtins.append(builtin)
else:
excluded_builtins.append(builtin)
for workload in app_inst.workloads[_FS].keys():
exec_graph = app_inst._get_executable_graph(workload)
app_inst.define_variable("workload_name", 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 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[_FS] = {"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[_FS] = {"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[_FS] = {"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_variables_and_variants(mutable_mock_apps_repo):
"""Test that set_variables defines workload variables"""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
del expansion_vars["n_ranks"]
experiment_variants = {
"workflow_manager": "slurm",
"foo": "bar",
}
# Set up the instance to pass the initial part of the function
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.definitions.variables.Variable("n_ranks", default="1"))
executable_application_instance.workloads[_FS] = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_application_instance.inputs[_FS] = {"input": {"target_dir": "."}}
executable_application_instance.set_variables_and_variants(
expansion_vars, experiment_variants, None
)
assert executable_application_instance.variables["n_ranks"] == "1"
variant_set = executable_application_instance.experiment_variants().as_set()
assert "workflow_manager=slurm" in variant_set
assert "package_manager=spack" not in variant_set
[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()
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.definitions.variables.Variable("n_ranks", default="1"))
executable_application_instance.workloads[_FS] = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_application_instance.inputs[_FS] = {"input": {"target_dir": "."}}
executable_application_instance.set_variables_and_variants(expansion_vars, {}, None)
exec_graph = executable_application_instance._get_executable_graph("test_wl2")
executable_application_instance.set_formatted_executables(
{"command": {"join_separator": "\n"}}
)
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_define_variables_for_template_path(mutable_mock_apps_repo, workspace_name):
"""_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
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()
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.definitions.variables.Variable("n_ranks", default="1"))
executable_application_instance.workloads[_FS] = {"test_wl": test_wl, "test_wl2": test_wl2}
executable_application_instance.internals = {}
executable_application_instance.inputs[_FS] = {"input": {"target_dir": "."}}
executable_application_instance.set_variables_and_variants(expansion_vars, {}, None)
exec_graph = executable_application_instance._get_executable_graph("test_wl2")
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.define_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_inst.variables = {
"n_nodes": "1",
"n_ranks": "{processes_per_node}*{n_nodes}",
"processes_per_node": "1",
"workload_name": "test_wl",
}
basic_clone = basic_inst.clone()
instances = [basic_inst, basic_clone]
for inst in instances:
assert hasattr(inst, "workloads")
assert "test_wl" in inst.workloads[_FS]
basic_clone.workload("added_workload", executables=["foo"])
assert "added_workload" in basic_clone.workloads[_FS]
assert "added_workload" not in basic_inst.workloads[_FS]
[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[_FS]
assert "empty" in workload_group_inst.workload_groups
assert "test_wlg" in workload_group_inst.workload_groups
possible_vars = workload_group_inst.workloads[_FS]["test_wl"].find_variable("test_var")
assert len(possible_vars) >= 1
found = False
for var in possible_vars:
if var.default == "2.0" and var.description == "Test workload vars and groups":
found = True
assert found
possible_vars = workload_group_inst.workloads[_FS]["test_wl"].find_variable("test_var_mixed")
assert len(possible_vars) >= 1
found = False
for var in possible_vars:
if var.default == "3.0" and var.description == "Test vars for workload and groups":
found = True
assert found
[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[_FS]
assert "test_wl3" in wlgi_inst.workloads[_FS]
# 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"].name
# Ensure a new workload can obtain the parent level vars via groups
possible_vars = wlgi_inst.workloads[_FS]["test_wl3"].find_variable("test_var")
assert len(possible_vars) >= 1
found = False
for var in possible_vars:
if var.default == "2.0" and var.description == "Test workload vars and groups":
found = True
assert found
for wl in ["test_wl", "test_wl3"]:
possible_vars = wlgi_inst.workloads[_FS][wl].find_variable("test_var_mixed")
assert len(possible_vars) >= 1
found = False
for var in possible_vars:
if var.default == "3.0":
found = True
assert found
[docs]
def test_undefined_executable_dies(mutable_mock_apps_repo, capsys):
"""Test that an undefined executable causes a fatal error."""
executable_application_instance = mutable_mock_apps_repo.get("basic")
expansion_vars = basic_exp_dict()
executable_application_instance.expander = ramble.expander.Expander(expansion_vars, None)
# Create a workload with an executable that is not defined
undefined_exec_wl = ramble.workload.Workload(
"wl_with_undefined_exec", executables=["undefined_exec"]
)
executable_application_instance.workloads[_FS]["wl_with_undefined_exec"] = undefined_exec_wl
with pytest.raises(SystemExit):
executable_application_instance._get_executable_graph("wl_with_undefined_exec")
captured = capsys.readouterr()
assert "Executable undefined_exec is not defined." in captured.err