testsuite/python/tests/test_116_12.py - SchedMD/slurm - Git at Google

 ############################################################################
 # Copyright (C) SchedMD LLC.
 ############################################################################
 import logging
 import re

 import pytest

 import atf

 ERROR_TYPE = "error"
 OUTPUT_TYPE = "output"
 node_count = 4


 # Setup
 @pytest.fixture(scope="module", autouse=True)
 def setup():
     atf.require_nodes(node_count)
     atf.require_slurm_running()


 # creates either an error or an output file with file name formatting
 def create_file_path(fnp, file_type=OUTPUT_TYPE):
     if file_type == ERROR_TYPE:
         return f"file_err.%{fnp}.error"
     return f"file_out.%{fnp}.output"


 # creates either an error or an output file with file name formatting
 def create_file(val, file_type=OUTPUT_TYPE):
     if file_type == ERROR_TYPE:
         return f"file_err.{val}.error"
     return f"file_out.{val}.output"


 def remove_file(file_path):
     atf.run_command(f"rm {file_path}")


 # returns the directory listing, waiting until all expected files appear.
 # If expected is None, waits for at least one file.
 def wait_and_assert_files(expected=None):
     for t in atf.timer(quiet=True):
         files = atf.run_command_output("ls").splitlines()
         if expected is None:
             if files:
                 return files
         elif all(f in files for f in expected):
             return files

         logging.debug(
             f"Found files: {files}. Expected files: {expected}. Remaining time: {t}"
         )
     assert False, f"Expected files not found: expected={expected}, got={files}"


 def test_output_error_formatting():
     """Verify srun stdout/err file name formatting (--output and --error options)."""

     # Test %t puts the task identifier in the file names
     task_count = 5
     file_out = create_file_path("t")
     atf.run_job(f"--output={file_out} -N1 -n{task_count} -O id")
     tmp_dir_list = wait_and_assert_files(
         [create_file(task) for task in range(task_count)]
     )
     for task in range(task_count):
         task_file = create_file(task)
         assert task_file in tmp_dir_list, f"%t: task file ({task_file}) was not created"
         remove_file(task_file)

     file_err = create_file_path("t", ERROR_TYPE)
     atf.run_job(f"--error={file_err} -N1 -n{task_count} -O uid")
     tmp_dir_list = wait_and_assert_files(
         [create_file(task, ERROR_TYPE) for task in range(task_count)]
     )
     for task in range(task_count):
         task_file = create_file(task, ERROR_TYPE)
         assert task_file in tmp_dir_list, f"%t: task file ({task_file}) was not created"
         remove_file(task_file)

     # Test %j puts the job id in the file names
     file_out = create_file_path("j")
     job_id = atf.submit_job_srun(f"--output={file_out} -N1 -O id")
     file_out = wait_and_assert_files()[0]
     assert (
         re.search(str(job_id), file_out) is not None
     ), f"%j: Job id ({job_id}) was not in file name ({file_out})"
     remove_file(file_out)

     file_err = create_file_path("j", ERROR_TYPE)
     job_id = atf.submit_job_srun(f"--error={file_err} -N1 -O uid")
     file_err = wait_and_assert_files()[0]
     assert (
         re.search(str(job_id), file_err) is not None
     ), f"%j: Job id ({job_id}) was not in file name ({file_err})"
     remove_file(file_err)

     # Test %% creates a file with one % in the file name
     output_file = create_file("%")
     file_out = create_file_path("%")
     atf.run_job(f"--output={file_out} -N1 -O id")
     wait_and_assert_files([output_file])
     remove_file(output_file)

     error_file = create_file("%", ERROR_TYPE)
     file_err = create_file_path("%", ERROR_TYPE)
     atf.run_job(f"--error={file_err} -N1 -O uid")
     wait_and_assert_files([error_file])
     remove_file(error_file)

     # Test %J puts the job.step id in the file name
     file_out = create_file_path("J")
     error_out = atf.run_job_error(f"--output={file_out} -v -N1 -O id")
     match = re.search(r"(?<=StepId=)\d+\.\d+", error_out)
     step_id = match.group(0)
     file_out = wait_and_assert_files()[0]
     assert (
         re.search(step_id, file_out) is not None
     ), f"%J: Step id ({step_id}) was not in file name ({file_out})"
     remove_file(file_out)

     file_err = create_file_path("J", ERROR_TYPE)
     error_out = atf.run_job_error(f"--error={file_err} -v -N1 -O uid")
     match = re.search(r"(?<=StepId=)\d+\.\d+", error_out)
     step_id = match.group(0)
     file_err = wait_and_assert_files()[0]
     assert (
         re.search(step_id, file_err) is not None
     ), f"%J: Step id ({step_id}) was not in file name ({file_err})"
     remove_file(file_err)

     # Test %u puts the user name in the file name
     user_name = atf.properties["test-user"]
     file_out = create_file_path("u")
     atf.run_job(f"--output={file_out} -N1 -O id")
     file_out = wait_and_assert_files()[0]
     assert re.search(
         user_name, file_out
     ), f"%u: User name ({user_name}) was not in file name ({file_out})"
     remove_file(file_out)

     file_err = create_file_path("u", ERROR_TYPE)
     atf.run_job(f"--error={file_err} -N1 -O uid")
     file_err = wait_and_assert_files()[0]
     assert re.search(
         user_name, file_err
     ), f"%u: User name ({user_name}) was not in file name ({file_err})"
     remove_file(file_err)

     # Test %n puts the node identifier relative to current job in the file name
     node_id = 0
     file_out = create_file_path("n")
     result_out = create_file(node_id)
     atf.run_job(f"--output={file_out} -N1 -O id")
     file_out = wait_and_assert_files()[0]
     assert (
         result_out in file_out
     ), f"%n: Node id ({node_id}) was not in file name ({file_out})"
     remove_file(file_out)

     file_err = create_file_path("n", ERROR_TYPE)
     result_err = create_file(node_id, ERROR_TYPE)
     atf.run_job(f"--error={file_err} -N1 -O id")
     file_err = wait_and_assert_files()[0]
     assert (
         result_err in file_err
     ), f"%n: Node id ({node_id}) was not in file name ({file_err})"
     remove_file(file_err)

     # Test %s puts the step identifier in the file name
     step_count = 4
     node_count = "1"
     file_out = create_file_path("s")
     file_in = "file_in.s.input"
     atf.make_bash_script(
         file_in,
         f"""for i in {{1..{step_count}}}
 do
     srun -O --output={file_out} true
 done""",
     )
     job_id = atf.submit_job_sbatch(f"-N{node_count} --output /dev/null {str(file_in)}")
     atf.wait_for_job_state(job_id, "DONE")
     tmp_dir_list = wait_and_assert_files(
         [create_file(step) for step in range(step_count)]
     )
     for step in range(0, step_count):
         step_file = create_file(step)
         assert step_file in tmp_dir_list, f"%s: Step file ({step_file}) was not created"
         remove_file(step_file)

     file_err = create_file_path("s", ERROR_TYPE)
     atf.make_bash_script(
         file_in,
         f"""for i in {{1..{step_count}}}
 do
     srun -O --error={file_err} true
 done""",
     )
     job_id = atf.submit_job_sbatch(f"-N{node_count} --output /dev/null {str(file_in)}")
     atf.wait_for_job_state(job_id, "DONE")
     tmp_dir_list = wait_and_assert_files(
         [create_file(step, ERROR_TYPE) for step in range(step_count)]
     )
     for step in range(0, step_count):
         step_file = create_file(step, ERROR_TYPE)
         assert step_file in tmp_dir_list, f"%s: Step file ({step_file}) was not created"
         remove_file(step_file)

     # Test %x puts the Job name in the file name
     job_command = "uid"
     file_out = create_file_path("x")
     file_err = create_file_path("x", ERROR_TYPE)
     job_id = atf.submit_job_srun(f"--output={file_out} {job_command}")
     job_name = atf.get_job_parameter(job_id, "JobName")
     assert (
         job_command == job_name
     ), f"%x: Job command ({job_command}) is not the same as the JobName ({job_name})"
     result_out = create_file(job_command)
     wait_and_assert_files([result_out])
     remove_file(result_out)

     job_id = atf.submit_job_srun(f"--error={file_err} {job_command}")
     job_name = atf.get_job_parameter(job_id, "JobName")
     assert (
         job_command == job_name
     ), f"%x: Job command ({job_command}) is not the same as the JobName ({job_name})"
     result_err = create_file(job_command, ERROR_TYPE)
     wait_and_assert_files([result_err])
     remove_file(result_err)

     # Test %N puts the short hostname in the file name
     file_out = create_file_path("N")
     file_err = create_file_path("N", ERROR_TYPE)

     job_id = atf.submit_job_srun(f"--output={file_out} -N1 -O id")
     node_name = atf.get_job_parameter(job_id, "NodeList")
     if atf.get_version("sbin/slurmd") >= (25, 11):
         # Ticket 23357: Use the node_name instead of hostname with %N
         expected_name = node_name
     else:
         expected_name = atf.get_node_parameter(node_name, "hostname")
         node_addr = atf.get_node_parameter(node_name, "address")
         if node_addr != expected_name and not atf.is_integer(node_addr[0]):
             expected_name = node_addr
     result_out = create_file(expected_name)
     wait_and_assert_files([result_out])
     remove_file(result_out)

     atf.submit_job_srun(f"--error={file_err} -N1 -O -w {node_name} true")
     result_err = create_file(expected_name, ERROR_TYPE)
     wait_and_assert_files([result_err])
     remove_file(result_err)

     # Test %A puts the Job array's master job allocation number in the file name
     array_size = 2
     file_out = create_file_path("A")
     file_err = create_file_path("A", ERROR_TYPE)
     file_in = "file_in.A.input"
     atf.make_bash_script(file_in, f"""srun -O --output={file_out} hostname""")
     job_id = atf.submit_job_sbatch(
         f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
     )
     atf.wait_for_job_state(job_id, "DONE")
     remove_file(file_in)
     result_out = create_file(str(job_id))
     wait_and_assert_files([result_out])
     remove_file(result_out)

     atf.make_bash_script(file_in, f"""srun -O --error={file_err} uid""")
     job_id = atf.submit_job_sbatch(
         f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
     )
     atf.wait_for_job_state(job_id, "DONE")
     remove_file(file_in)
     result_err = create_file(str(job_id), ERROR_TYPE)
     wait_and_assert_files([result_err])
     remove_file(result_err)

     # Test %a puts the Job array ID in the file name
     array_size = 2
     file_out = create_file_path("A.%a")
     file_err = create_file_path("A.%a", ERROR_TYPE)
     file_in = "file_in.A.a.input"
     atf.make_bash_script(file_in, f"""srun -O --output={file_out} hostname""")
     job_id = atf.submit_job_sbatch(
         f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
     )
     atf.wait_for_job_state(job_id, "DONE")
     tmp_dir_list = wait_and_assert_files(
         [create_file(f"{job_id}.{array_id}") for array_id in range(1, array_size + 1)]
     )
     for array_id in range(1, array_size + 1):
         id_file = create_file(str(job_id) + "." + str(array_id))
         assert (
             id_file in tmp_dir_list
         ), f"%a: Job array file ({id_file}) was not created"
         remove_file(id_file)

     atf.make_bash_script(file_in, f"""srun -O --error={file_err} uid""")
     job_id = atf.submit_job_sbatch(
         f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
     )
     atf.wait_for_job_state(job_id, "DONE")
     tmp_dir_list = wait_and_assert_files(
         [
             create_file(f"{job_id}.{array_id}", ERROR_TYPE)
             for array_id in range(1, array_size + 1)
         ]
     )
     for array_id in range(1, array_size + 1):
         id_file = create_file(str(job_id) + "." + str(array_id), ERROR_TYPE)
         assert (
             id_file in tmp_dir_list
         ), f"%a: Job array file ({id_file}) was not created"
         remove_file(id_file)
	############################################################################
	# Copyright (C) SchedMD LLC.
	############################################################################
	import logging
	import re

	import pytest

	import atf

	ERROR_TYPE = "error"
	OUTPUT_TYPE = "output"
	node_count = 4


	# Setup
	@pytest.fixture(scope="module", autouse=True)
	def setup():
	atf.require_nodes(node_count)
	atf.require_slurm_running()


	# creates either an error or an output file with file name formatting
	def create_file_path(fnp, file_type=OUTPUT_TYPE):
	if file_type == ERROR_TYPE:
	return f"file_err.%{fnp}.error"
	return f"file_out.%{fnp}.output"


	# creates either an error or an output file with file name formatting
	def create_file(val, file_type=OUTPUT_TYPE):
	if file_type == ERROR_TYPE:
	return f"file_err.{val}.error"
	return f"file_out.{val}.output"


	def remove_file(file_path):
	atf.run_command(f"rm {file_path}")


	# returns the directory listing, waiting until all expected files appear.
	# If expected is None, waits for at least one file.
	def wait_and_assert_files(expected=None):
	for t in atf.timer(quiet=True):
	files = atf.run_command_output("ls").splitlines()
	if expected is None:
	if files:
	return files
	elif all(f in files for f in expected):
	return files

	logging.debug(
	f"Found files: {files}. Expected files: {expected}. Remaining time: {t}"
	)
	assert False, f"Expected files not found: expected={expected}, got={files}"


	def test_output_error_formatting():
	"""Verify srun stdout/err file name formatting (--output and --error options)."""

	# Test %t puts the task identifier in the file names
	task_count = 5
	file_out = create_file_path("t")
	atf.run_job(f"--output={file_out} -N1 -n{task_count} -O id")
	tmp_dir_list = wait_and_assert_files(
	[create_file(task) for task in range(task_count)]
	)
	for task in range(task_count):
	task_file = create_file(task)
	assert task_file in tmp_dir_list, f"%t: task file ({task_file}) was not created"
	remove_file(task_file)

	file_err = create_file_path("t", ERROR_TYPE)
	atf.run_job(f"--error={file_err} -N1 -n{task_count} -O uid")
	tmp_dir_list = wait_and_assert_files(
	[create_file(task, ERROR_TYPE) for task in range(task_count)]
	)
	for task in range(task_count):
	task_file = create_file(task, ERROR_TYPE)
	assert task_file in tmp_dir_list, f"%t: task file ({task_file}) was not created"
	remove_file(task_file)

	# Test %j puts the job id in the file names
	file_out = create_file_path("j")
	job_id = atf.submit_job_srun(f"--output={file_out} -N1 -O id")
	file_out = wait_and_assert_files()[0]
	assert (
	re.search(str(job_id), file_out) is not None
	), f"%j: Job id ({job_id}) was not in file name ({file_out})"
	remove_file(file_out)

	file_err = create_file_path("j", ERROR_TYPE)
	job_id = atf.submit_job_srun(f"--error={file_err} -N1 -O uid")
	file_err = wait_and_assert_files()[0]
	assert (
	re.search(str(job_id), file_err) is not None
	), f"%j: Job id ({job_id}) was not in file name ({file_err})"
	remove_file(file_err)

	# Test %% creates a file with one % in the file name
	output_file = create_file("%")
	file_out = create_file_path("%")
	atf.run_job(f"--output={file_out} -N1 -O id")
	wait_and_assert_files([output_file])
	remove_file(output_file)

	error_file = create_file("%", ERROR_TYPE)
	file_err = create_file_path("%", ERROR_TYPE)
	atf.run_job(f"--error={file_err} -N1 -O uid")
	wait_and_assert_files([error_file])
	remove_file(error_file)

	# Test %J puts the job.step id in the file name
	file_out = create_file_path("J")
	error_out = atf.run_job_error(f"--output={file_out} -v -N1 -O id")
	match = re.search(r"(?<=StepId=)\d+\.\d+", error_out)
	step_id = match.group(0)
	file_out = wait_and_assert_files()[0]
	assert (
	re.search(step_id, file_out) is not None
	), f"%J: Step id ({step_id}) was not in file name ({file_out})"
	remove_file(file_out)

	file_err = create_file_path("J", ERROR_TYPE)
	error_out = atf.run_job_error(f"--error={file_err} -v -N1 -O uid")
	match = re.search(r"(?<=StepId=)\d+\.\d+", error_out)
	step_id = match.group(0)
	file_err = wait_and_assert_files()[0]
	assert (
	re.search(step_id, file_err) is not None
	), f"%J: Step id ({step_id}) was not in file name ({file_err})"
	remove_file(file_err)

	# Test %u puts the user name in the file name
	user_name = atf.properties["test-user"]
	file_out = create_file_path("u")
	atf.run_job(f"--output={file_out} -N1 -O id")
	file_out = wait_and_assert_files()[0]
	assert re.search(
	user_name, file_out
	), f"%u: User name ({user_name}) was not in file name ({file_out})"
	remove_file(file_out)

	file_err = create_file_path("u", ERROR_TYPE)
	atf.run_job(f"--error={file_err} -N1 -O uid")
	file_err = wait_and_assert_files()[0]
	assert re.search(
	user_name, file_err
	), f"%u: User name ({user_name}) was not in file name ({file_err})"
	remove_file(file_err)

	# Test %n puts the node identifier relative to current job in the file name
	node_id = 0
	file_out = create_file_path("n")
	result_out = create_file(node_id)
	atf.run_job(f"--output={file_out} -N1 -O id")
	file_out = wait_and_assert_files()[0]
	assert (
	result_out in file_out
	), f"%n: Node id ({node_id}) was not in file name ({file_out})"
	remove_file(file_out)

	file_err = create_file_path("n", ERROR_TYPE)
	result_err = create_file(node_id, ERROR_TYPE)
	atf.run_job(f"--error={file_err} -N1 -O id")
	file_err = wait_and_assert_files()[0]
	assert (
	result_err in file_err
	), f"%n: Node id ({node_id}) was not in file name ({file_err})"
	remove_file(file_err)

	# Test %s puts the step identifier in the file name
	step_count = 4
	node_count = "1"
	file_out = create_file_path("s")
	file_in = "file_in.s.input"
	atf.make_bash_script(
	file_in,
	f"""for i in {{1..{step_count}}}
	do
	srun -O --output={file_out} true
	done""",
	)
	job_id = atf.submit_job_sbatch(f"-N{node_count} --output /dev/null {str(file_in)}")
	atf.wait_for_job_state(job_id, "DONE")
	tmp_dir_list = wait_and_assert_files(
	[create_file(step) for step in range(step_count)]
	)
	for step in range(0, step_count):
	step_file = create_file(step)
	assert step_file in tmp_dir_list, f"%s: Step file ({step_file}) was not created"
	remove_file(step_file)

	file_err = create_file_path("s", ERROR_TYPE)
	atf.make_bash_script(
	file_in,
	f"""for i in {{1..{step_count}}}
	do
	srun -O --error={file_err} true
	done""",
	)
	job_id = atf.submit_job_sbatch(f"-N{node_count} --output /dev/null {str(file_in)}")
	atf.wait_for_job_state(job_id, "DONE")
	tmp_dir_list = wait_and_assert_files(
	[create_file(step, ERROR_TYPE) for step in range(step_count)]
	)
	for step in range(0, step_count):
	step_file = create_file(step, ERROR_TYPE)
	assert step_file in tmp_dir_list, f"%s: Step file ({step_file}) was not created"
	remove_file(step_file)

	# Test %x puts the Job name in the file name
	job_command = "uid"
	file_out = create_file_path("x")
	file_err = create_file_path("x", ERROR_TYPE)
	job_id = atf.submit_job_srun(f"--output={file_out} {job_command}")
	job_name = atf.get_job_parameter(job_id, "JobName")
	assert (
	job_command == job_name
	), f"%x: Job command ({job_command}) is not the same as the JobName ({job_name})"
	result_out = create_file(job_command)
	wait_and_assert_files([result_out])
	remove_file(result_out)

	job_id = atf.submit_job_srun(f"--error={file_err} {job_command}")
	job_name = atf.get_job_parameter(job_id, "JobName")
	assert (
	job_command == job_name
	), f"%x: Job command ({job_command}) is not the same as the JobName ({job_name})"
	result_err = create_file(job_command, ERROR_TYPE)
	wait_and_assert_files([result_err])
	remove_file(result_err)

	# Test %N puts the short hostname in the file name
	file_out = create_file_path("N")
	file_err = create_file_path("N", ERROR_TYPE)

	job_id = atf.submit_job_srun(f"--output={file_out} -N1 -O id")
	node_name = atf.get_job_parameter(job_id, "NodeList")
	if atf.get_version("sbin/slurmd") >= (25, 11):
	# Ticket 23357: Use the node_name instead of hostname with %N
	expected_name = node_name
	else:
	expected_name = atf.get_node_parameter(node_name, "hostname")
	node_addr = atf.get_node_parameter(node_name, "address")
	if node_addr != expected_name and not atf.is_integer(node_addr[0]):
	expected_name = node_addr
	result_out = create_file(expected_name)
	wait_and_assert_files([result_out])
	remove_file(result_out)

	atf.submit_job_srun(f"--error={file_err} -N1 -O -w {node_name} true")
	result_err = create_file(expected_name, ERROR_TYPE)
	wait_and_assert_files([result_err])
	remove_file(result_err)

	# Test %A puts the Job array's master job allocation number in the file name
	array_size = 2
	file_out = create_file_path("A")
	file_err = create_file_path("A", ERROR_TYPE)
	file_in = "file_in.A.input"
	atf.make_bash_script(file_in, f"""srun -O --output={file_out} hostname""")
	job_id = atf.submit_job_sbatch(
	f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
	)
	atf.wait_for_job_state(job_id, "DONE")
	remove_file(file_in)
	result_out = create_file(str(job_id))
	wait_and_assert_files([result_out])
	remove_file(result_out)

	atf.make_bash_script(file_in, f"""srun -O --error={file_err} uid""")
	job_id = atf.submit_job_sbatch(
	f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
	)
	atf.wait_for_job_state(job_id, "DONE")
	remove_file(file_in)
	result_err = create_file(str(job_id), ERROR_TYPE)
	wait_and_assert_files([result_err])
	remove_file(result_err)

	# Test %a puts the Job array ID in the file name
	array_size = 2
	file_out = create_file_path("A.%a")
	file_err = create_file_path("A.%a", ERROR_TYPE)
	file_in = "file_in.A.a.input"
	atf.make_bash_script(file_in, f"""srun -O --output={file_out} hostname""")
	job_id = atf.submit_job_sbatch(
	f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
	)
	atf.wait_for_job_state(job_id, "DONE")
	tmp_dir_list = wait_and_assert_files(
	[create_file(f"{job_id}.{array_id}") for array_id in range(1, array_size + 1)]
	)
	for array_id in range(1, array_size + 1):
	id_file = create_file(str(job_id) + "." + str(array_id))
	assert (
	id_file in tmp_dir_list
	), f"%a: Job array file ({id_file}) was not created"
	remove_file(id_file)

	atf.make_bash_script(file_in, f"""srun -O --error={file_err} uid""")
	job_id = atf.submit_job_sbatch(
	f"-N1 --output=/dev/null --array=1-{array_size} {file_in}"
	)
	atf.wait_for_job_state(job_id, "DONE")
	tmp_dir_list = wait_and_assert_files(
	[
	create_file(f"{job_id}.{array_id}", ERROR_TYPE)
	for array_id in range(1, array_size + 1)
	]
	)
	for array_id in range(1, array_size + 1):
	id_file = create_file(str(job_id) + "." + str(array_id), ERROR_TYPE)
	assert (
	id_file in tmp_dir_list
	), f"%a: Job array file ({id_file}) was not created"
	remove_file(id_file)