import os
import shlex
import shutil
from buildtest.builders.base import BuilderBase
from buildtest.tools.modules import get_module_commands
from buildtest.utils.file import write_file
from buildtest.utils.tools import deep_get
[docs]class ScriptBuilder(BuilderBase):
"""This is a subclass of BuilderBase used for building test that uses ``type: script`` in the buildspec."""
type = "script"
def __init__(
self,
name,
recipe,
buildspec,
executor,
buildexecutor,
configuration,
testdir=None,
numprocs=None,
numnodes=None,
compiler=None,
):
super().__init__(
name=name,
recipe=recipe,
buildspec=buildspec,
executor=executor,
buildexecutor=buildexecutor,
testdir=testdir,
numprocs=numprocs,
numnodes=numnodes,
compiler=compiler,
)
self.compiler_settings = {"vars": None, "env": None, "modules": None}
self.configuration = configuration
self.compiler_section = self.recipe.get("compilers")
# if 'compilers' property defined resolve compiler logic
if self.compiler_section:
self._process_compiler_config()
self.resolve_compilers()
self.status = deep_get(
self.recipe, "executors", self.executor, "status"
) or self.recipe.get("status")
self.metrics = deep_get(
self.recipe, "executors", self.executor, "metrics"
) or self.recipe.get("metrics")
[docs] def resolve_compilers(self):
# get environment variables
self.compiler_settings["env"] = deep_get(
self.compiler_section, "config", self.compiler, "env"
) or deep_get(self.compiler_section, "default", self.compiler_group, "env")
# get environment variables
self.compiler_settings["vars"] = deep_get(
self.compiler_section, "config", self.compiler, "vars"
) or deep_get(self.compiler_section, "default", self.compiler_group, "vars")
# compiler set in compilers 'config' section, we try to get module lines using self._get_modules
self.compiler_settings["modules"] = get_module_commands(
deep_get(self.compiler_section, "config", self.compiler, "module")
)
if not self.compiler_settings["modules"]:
self.compiler_settings["modules"] = get_module_commands(
self.bc_compiler.get("module")
)
[docs] def write_python_script(self):
"""This method is used for writing python script when ``shell: python``
is set. The content from ``run`` section is added into a python
script. The file is written to run directory and we simply invoke
python script by running ``python script.py``
"""
python_content = self.recipe.get("run")
script_path = "%s.py" % os.path.join(self.stage_dir, self.name)
write_file(script_path, python_content)
self.logger.debug(f"[{self.name}]: Writing python script to: {script_path}")
shutil.copy2(
script_path, os.path.join(self.test_root, os.path.basename(script_path))
)
self.logger.debug(
f"[{self.name}]: Copying file: {script_path} to: {os.path.join(self.test_root, os.path.basename(script_path))}"
)
# lines = [f"python {script_path}"]
# return lines
[docs] def generate_script(self):
"""This method builds the content of the test script which will return a list
of shell commands that will be written to file.
A typical test will contain: shebang line, job directives, environment variables and variable declaration,
and content of ``run`` property. For ``shell: python`` we write a python script and
return immediately. The variables, environment section are not applicable
for python scripts
Returns:
List of shell commands that will be written to file
"""
# start of each test should have the shebang
lines = [self.shebang]
# if shell is python the generated testscript will be run via bash, we invoke
# python script in bash script.
if self.shell.name == "python":
lines = ["#!/bin/bash"]
sched_lines = self.get_job_directives()
if sched_lines:
lines += sched_lines
if self.burstbuffer:
burst_buffer_lines = self._get_burst_buffer(self.burstbuffer)
if burst_buffer_lines:
lines += burst_buffer_lines
if self.datawarp:
data_warp_lines = self._get_data_warp(self.datawarp)
if data_warp_lines:
lines += data_warp_lines
lines.append(self._emit_set_command())
# for python scripts we generate python script and return lines
if self.shell.name == "python":
self.logger.debug(f"[{self.name}]: Detected python shell")
self.write_python_script()
py_script = "%s.py" % format(os.path.join(self.stage_dir, self.name))
python_wrapper = self.buildexecutor.executors[self.executor]._settings[
"shell"
]
python_wrapper_buildspec = shlex.split(self.recipe.get("shell"))[0]
# if 'shell' property in buildspec specifies 'shell: python' or 'shell: python3' then we use this instead
if python_wrapper_buildspec.endswith(
"python"
) or python_wrapper_buildspec.endswith("python3"):
python_wrapper = python_wrapper_buildspec
lines.append(f"{python_wrapper} {py_script}")
return lines
# section below is for shell-scripts (bash, sh, csh, zsh, tcsh, zsh)
if self.compiler:
compiler_variables = {
"BUILDTEST_CC": self.cc,
"BUILDTEST_CXX": self.cxx,
"BUILDTEST_FC": self.fc,
"BUILDTEST_CFLAGS": self.cflags,
"BUILDTEST_CXXFLAGS": self.cxxflags,
"BUILDTEST_FFLAGS": self.fflags,
"BUILDTEST_CPPFLAGS": self.cppflags,
"BUILDTEST_LDFLAGS": self.ldflags,
}
lines += self._get_variables(compiler_variables)
if self.compiler_settings["env"]:
lines += self._get_environment(self.compiler_settings["env"])
if self.compiler_settings["vars"]:
lines += self._get_variables(self.compiler_settings["vars"])
# Add environment variables
env_section = deep_get(
self.recipe, "executors", self.executor, "env"
) or self.recipe.get("env")
var_section = deep_get(
self.recipe, "executors", self.executor, "vars"
) or self.recipe.get("vars")
env_lines = self._get_environment(env_section)
if env_lines:
lines += env_lines
var_lines = self._get_variables(var_section)
if var_lines:
lines += var_lines
if self.compiler_settings["modules"]:
lines += self.compiler_settings["modules"]
lines.append("# Content of run section")
if "container" in self.recipe:
container = self.recipe["container"]
container_platform = container["platform"]
container_command = []
if container_platform in ["docker", "podman"]:
container_command.extend(
[container_platform, "run", "-v", f"{self.stage_dir}:/buildtest"]
)
elif container_platform in ["singularity"]:
container_command.extend(
[f"{container_platform}", "exec", f"-B {self.stage_dir}:/buildtest"]
)
if container.get("mounts"):
if container_platform in ["singularity"]:
container_command.extend(["-B", container["mounts"]])
else:
container_command.extend(["-v", container["mounts"]])
if container.get("options"):
container_command.append(container["options"])
container_command.append(container["image"])
if container.get("command"):
container_command.append(container["command"])
lines.append(" ".join(container_command))
# Add run section
lines += [self.recipe["run"]]
return lines