diff --git a/docs/InformationAboutSpecialVariables.md b/docs/InformationAboutSpecialVariables.md new file mode 100644 index 0000000..d44d8e5 --- /dev/null +++ b/docs/InformationAboutSpecialVariables.md @@ -0,0 +1,146 @@ +# Special variables that are used by the integrationtest infrastructure + +18-Aug-2026, Kurt Biery + +## Introduction + +In the Pytest files that we write (our integtests), there are several special variables that are used to communication information about the desired conditions of the testing to the `integrationtest` infrastructure. This information includes things such as configuration parameters and run control commands. + +This document describes the special variables that are currently available and how they can, and should, be used. + +### Computer resource validation parameters + +This is communicated by the `resource_validator` special variable. It should point to an instance of the `ResourceValidator` class. This class is defined in [integrationtest/src/integrationtest/resource_validation.py](https://github.com/DUNE-DAQ/integrationtest/blob/develop/src/integrationtest/resource_validation.py). + +(More details coming soon.) + +### Integrationtest and DAQ system configuration parameters + +This is communicated by the `confgen_arguments` special variable. + +(More details coming soon.) + +### Run control process manager type(s) + +This is communicated by the `process_manager_choices` special variable. + +(More details coming soon.) + +### Run control commands or full DAQ session ingredients + +These are communicated either by the `dunerc_command_list` or the `daq_session_ingredients` special variable. Only one of these two variables should be specified in a single integtest file, but if both of them happen to be specified in the same integtest, the `daq_session_ingredients` takes precedence. + +The purposes of these two variables are similar - both provide commands that should be run by one or more run control applications - but the `daq_session_ingredients` variable is more powerful in that it allows users to specify one or more applications that should be run, instead of simply using the `drunc-unified-shell`. + +Information about `dunerc_command_list`: + +* this is the variable that has been used historically, and many of our existing integtests use it. +* it is expected to contain a Python list of the commands (strings) that are passed to run control in "batch" mode + * some examples: + * `dunerc_command_list = ("boot conf start --run-number 101 wait 1 enable-triggers wait ".split() + [str(run_duration)] + "disable-triggers wait 2 drain-dataflow wait 2 stop-trigger-sources stop scrap terminate".split())` + * `dunerc_command_list = ["boot", "conf", "start", "--run-number", "101", "wait", str(1), "enable-triggers", "wait", str(20), "disable-triggers", "stop-run", "shutdown"]` +* in addition to containing a single list of commands (as shown above), this variable can contain a dictionary of one or more lists of commands. With this functionality, multiple DAQ sessions with different sets of commands can be run from an single integtest. + * here is an example of this type declaration: + * `dunerc_command_list = {"DAQ_Session_1": ["boot", "conf", "start", "--run-number", "101", "wait", str(1), "enable-triggers", "wait", str(20), "disable-triggers", "stop-run", "shutdown"], "DAQ Session 2": ["boot", "conf", "start", "--run-number", "101", "wait", str(3), "enable-triggers", "wait", str(20), "disable-triggers", "stop-run", "scrap", "terminate"]}` + +Information about `daq_session_ingredients`: +* this variable was recently introduced so that developers of integtests can specify multiple control applications to be run in a given (integtest) DAQ session +* at the moment, this variable needs to contain a dictionary with one or more elements, and each element should contain a string key (with a word or phrase that describes the DAQ session) and an instance of the `DAQSessionIngredients` class as the value. The `DAQSessionIngredients` class is defined in [integrationtest/src/integrationtest/data_classes.py](https://github.com/DUNE-DAQ/integrationtest/blob/0fe60d9b1c1aa697ec9524c4aaf1507aaa3c6b2a/src/integrationtest/data_classes.py#L139). +* the [basic_multiapp_test.py](https://github.com/DUNE-DAQ/drunc/blob/kbiery/multi_ctrl_proc_support/src/drunc/integtest/basic_multiapp_test.py) regression test in the `drunc` repo has an example of specifying three applications to be run in the DAQ session and specifying commands that are sent to two of those applications. + * For reference, the relevant lines from `basic_multiapp_test.py` are copied below. +* in these instructions, I have tried to use the word "application" to mean a C++ program or a Python script that has been developed to perform one or more functions. And, I have tried to use the word "process" to mean an instance of an application that is running as part of a DAQ session. Apologies if this model is not strictly used everywhere. +* the `DAQSessionIngredients` class has data members that allow developers to specify the applications that should be run and the commands that should be sent to the processes. In this class, applications are represented by instances of the `DAQControlApplication` class and commands are listed in instances of the `DAQCommandSet` class. The `DAQCommandSet` has a field that specifies the process that we want to send the commands to. + * reference information: + +```python +@dataclass +class DAQSessionIngredients: + applications: list[DAQControlApplication] + commands: list[DAQCommandSet] + +@dataclass +class DAQControlApplication: + alias: str # a short-hand name for the process that is started + startup_strings: list[str] # the elements of the command string that should be used to start the application + wait_time_after_start: int = 2 # seconds to sleep after spawning the process + +@dataclass +class DAQCommandSet: + target: str # the name of the process that should receive the commands + command_list: list[str] $ the list of commands, e.g. ["boot", "conf"] + wait_params: CommandWaitParameters = field(default_factory=lambda: CommandWaitParameters()) + +@dataclass +class CommandWaitParameters: # please see the comments below for information about this class, etc. + wait_for_command_completion: bool = True + style: CommandWaitStyle = CommandWaitStyle.TIME + timeout_waiting_for_first_msg: int = 2 # seconds + wait_time_after_last_msg: int = 2 # seconds + timeout_waiting_for_exit: int = 5 # seconds + +class CommandWaitStyle(Enum): + ECHO = "echo" + TIME = "time" + TIME_PLUS_EXIT = "time_plus_exit" + NONE = "none" +``` + +* Here is some additional information about `CommandWaitParameters`: + * the commands that are specified in a `DAQCommandSet` are sent individually to the target process without any delay between them. So, we typically send all of the commands in the set in a fraction of a second, while the target process could take tens of seconds to execute all of them. + * when there is only one control process in an integtest, this rapid-fire approach may be all that we need, because a single process handles the throttling of the commands, running them one after another. However, when there are multiple control processes in an integtest, we may want to send a set of commands to Process1, wait for those to finish, and only then send a set of commands to Process2. This demonstrates a need to allow an `integrationtest` developer to specify whether they want the integrationtest infrastructure to wait for each command set to finish before moving on to the next set of commands, and if so, what style of waiting they would like be used. This is the motivation for the `CommandWaitParameters` class. + * of course, there are also situations in which we want to wait for all of the requested commands to finish running even when there is only one control process in the integtest. For example, we will likely want to allow a single process to finish executing all of the requested commands before the `integrationtest` infrastructure starts shutting down that process. + * the currently-supported wait styles are ECHO, TIME, and TIME_PLUS_EXIT. + * the ECHO wait style makes use of the `echo` command that is available in some of our control applications to clearly identify when a set of commands has finished. So, if a user specifies a command set that contains commands `['boot', 'conf']` and has a wait style of ECHO, the `integrationtest` infrastructure appends an `echo` command with a special string to the set, i.e. `['boot', 'conf', 'echo ""']`. When the `integrationtest` infrastructure sees the special string in the output of the target process, it knows that the command set has finished. + * this wait style is the most robust since we know that all of the commands before the `echo` command have been run when the `echo` results are seen in the process output. However, some applications don't provide `echo` functionality. + * the TIME wait style simply waits for configured amounts of time for console output to start and then stop. The idea here is to use the console output as an indicator of activity, and when the console output stops, presume that activity related to the requested command(s) has stopped. + * the TIME_PLUS_EXIT wait style is intended to be used with "exit" commands. The idea here is to wait for console output to stop and then wait for the process to exit (within a configurable timeout). +* There are several strings that are dynamically determined by the `integrationtest` infrastructure that we may want to include in the `startup_strings` field in our `DAQControlApplication` declarations. To take this into account, placeholder strings have been defined. These placeholder strings can be used in `DAQControlApplication` declarations and the `integrationtest` infrastructure will substitute the appropriate value at runtime. The placeholders that are currently available are the following: + * `` - the process manager type that should be used in the DAQ session + * recall that the `integrationtest` infrastructure has support for user-specified (dynamic) process manager types. If we don't want to make use of that functionality, we can hard-code the process manager type in our `DAQControlApplication.startup_strings`. Of course, that reduces flexibility, but there may be cases where it would make sense. + * `` - the configuration data file that the infrastructure has created for the integtest + * this placeholder string should always be used since the `integrationtest` infrastructure creates a new, temporary config data file for each running of an integtest + * `` - the name of the configuration session that should be used for the DAQ session + * this could be hard-coded, but it is safer to let it get filled in dynamically + * `` - the name that should be used to identify the DAQ session + * this placeholder can be used, or the name of the DAQ session could be hard-coded in the `startup_strings` +* when an integration test is run with verbosity level of 4 or greater, the command lines that are used to start the applications are printed on the console, and this output can be used to check if the desired substitutions were made + +Here is a snippet of code from the `basic_multapp_test.py` that shows how the `DAQSessionIngredients` are constructed in that integtest: + +```python +# The commands to run in dunerc and the process manager shell +dunerc_commands_1 = ( + "boot conf start --run-number 101 wait 1 enable-triggers wait ".split() + + [str(run_duration)] + ["disable-triggers"] +) +dunerc_commands_2 = ( + "drain-dataflow stop-trigger-sources stop wait 2 scrap terminate".split() +) +pmshell_command = ["ps"] + +# Find a free network port to use for the process manager +pm_port = find_free_port(50020, 52000) + +# The command lines that should be used to start the applications +procmsg_startup_commands = ["drunc-process-manager", "", str(pm_port)] +pmapp = DAQControlApplication("pm", procmsg_startup_commands) + +pmshell_startup_commands = ["drunc-process-manager-shell", f"grpc://localhost:{pm_port}"] +pmshellapp = DAQControlApplication("pmshell", pmshell_startup_commands) + +drunc_startup_commands = ["drunc-unified-shell", f"grpc://localhost:{pm_port}", "", "", ""] +druncapp = DAQControlApplication("drunc", drunc_startup_commands) + +# Packaging up the commands into DAQCommandSets +cmd_set_1 = DAQCommandSet("drunc", dunerc_commands_1, CommandWaitParameters(style=CommandWaitStyle.ECHO)) +cmd_set_2 = DAQCommandSet("pmshell", pmshell_command, CommandWaitParameters(style=CommandWaitStyle.TIME)) +cmd_set_3 = DAQCommandSet("drunc", dunerc_commands_2, CommandWaitParameters(style=CommandWaitStyle.ECHO)) + +# Putting everything together into a DAQSessionIngredients object +app_list = [ pmapp, pmshellapp, druncapp ] +cmd_set_list = [ cmd_set_1, cmd_set_2, cmd_set_3 ] +dsi = DAQSessionIngredients(app_list, cmd_set_list) + +# Declare the special variable that tells the integrationtest infrastructure what we want to run +daq_session_ingredients = {"MultiRCAppSession": dsi} +``` diff --git a/src/integrationtest/async_proc_mgmt.py b/src/integrationtest/async_proc_mgmt.py new file mode 100644 index 0000000..3838db4 --- /dev/null +++ b/src/integrationtest/async_proc_mgmt.py @@ -0,0 +1,347 @@ +import pytest +import asyncio +import getpass +import pathlib +import re +import time +from integrationtest.data_classes import * +from integrationtest.verbosity_helper import * +from datetime import datetime, timezone +from typing import Final + +import functools +print = functools.partial(print, flush=True) # always flush print() output + +PROCESS_ECHO_STRING: Final[str] = "*** COMMAND HAS COMPLETED ***" + + +async def read_stream(stream, process_name, app_exe_name, print_proc_name, run_dir, + shared_data: CommandProcessingSharedData, verbosity_level): + """Asynchronously reads lines from a stream and processes them immediately.""" + full_output = "" + observed_command_prompt = "" + + # store the full output in a log file to be checked for problems and for later reference + with open(f"{run_dir}/log_{getpass.getuser()}_{app_exe_name}_console_output.txt", "w", encoding="utf-8") as ff: + while True: + line = await stream.readline() + if not line: + break + decoded_line = line.decode() + async with shared_data.lock: + shared_data.last_msg_time = time.time() + + # if the special end-of-command string has been echo-ed by the process, + # send the relevant signal to any waiting task by setting the completion event + if PROCESS_ECHO_STRING in decoded_line: + shared_data.cmd_cmplt_evt.set() + continue + + # process the output of the "help" command, if requested + async with shared_data.lock: + if shared_data.parsing_of_help_output_in_progress: + trimmed_line = decoded_line.strip() + if len(trimmed_line) == 0: + continue + if "documented commands" in trimmed_line.lower(): + continue + if "=====" in trimmed_line: + continue + if trimmed_line.endswith(r">"): + observed_command_prompt = trimmed_line + continue + if verbosity_level >= IntegtestVerbosityLevels.full_output: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] Help command output: {decoded_line}") + the_cmds = trimmed_line.split() + tmp_list = shared_data.results_of_parsing_help_output + the_cmds + shared_data.results_of_parsing_help_output = sorted(set(tmp_list)) + continue + + # print out each line of output, subject to the verbosity level that the user has + # requested, as well as writing it to a log file and adding it to a string that + # we pass back to the user + should_be_printed = verbosity_level >= IntegtestVerbosityLevels.full_output + + # check for errors and warnings for all verbosity levels + if not should_be_printed: + lc_line = decoded_line.lower() + if ("error" in lc_line and (not "In error" in decoded_line and not "Endpoint" in decoded_line)) \ + or "warning" in lc_line or "critical" in lc_line: + should_be_printed = True + + # check for basic transition messages, if that level of verbosity is requested + if not should_be_printed: + if verbosity_level >= IntegtestVerbosityLevels.drunc_boot_terminate: + if "Booting session" in decoded_line or \ + ("Current FSM status is " in decoded_line and ("initial" in decoded_line or "running" in decoded_line)): + should_be_printed = True + + # check for all transition messages, if that level of verbosity is requested + if not should_be_printed: + if verbosity_level >= IntegtestVerbosityLevels.drunc_transitions: + if "Booting session" in decoded_line or "Running transition" in decoded_line \ + or ("wait" in decoded_line and "running" in decoded_line) or "exit code" in decoded_line: + should_be_printed = True + + # remove the application command prompt from the front of the line, if needed + if len(observed_command_prompt) > 0 and decoded_line.startswith(observed_command_prompt): + tmp_line = decoded_line.removeprefix(observed_command_prompt) + decoded_line = tmp_line.lstrip() + + # actually do the printout + if should_be_printed: + async with shared_data.lock: + if shared_data.number_of_lines_printed_to_the_console == 0: + print("++++++++++ DAQ Session BEGIN ++++++++++", flush=True) + if print_proc_name: + print(f"[{process_name}] {decoded_line}", end='', flush=True) + else: + print(decoded_line, end='', flush=True) + shared_data.number_of_lines_printed_to_the_console += 1 + + clean_line = re.sub(r"\x1b\[[0-9;]*m", "", decoded_line) + ff.write(f"{clean_line}") + ff.flush() + full_output += clean_line + + return full_output + + +async def wait_for_console_output_lull(start_time, wait_params: CommandWaitParameters, + shared_data: CommandProcessingSharedData): + now = time.time() + while True: + async with shared_data.lock: + if shared_data.last_msg_time <= start_time: + if now - start_time > wait_params.timeout_waiting_for_first_msg: + break + else: + if now - shared_data.last_msg_time >= wait_params.wait_time_after_last_msg: + break + await asyncio.sleep(0.25) + now = time.time() + + +async def send_commands(target_proc_info, proc_name, shared_data: CommandProcessingSharedData, + cmd_list, wait_params, verbosity_level): + target_proc = target_proc_info.process + if target_proc.returncode is not None: # Check if process is still running + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] Error: {proc_name} has already exited, unable to send \"{cmd_list}\".") + return + + # send the requested commands + cmd_start_time = time.time() + for cmd in cmd_list: + target_proc.stdin.write((cmd + "\n").encode()) + await target_proc.stdin.drain() + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] Sent command to {proc_name}: {cmd}") + else: + async with shared_data.lock: + if shared_data.number_of_lines_printed_to_the_console == 0: + print(".", end="") + + # wait for the command(s) to finish, if requested + if not wait_params.wait_for_command_completion: + return + if wait_params.style == CommandWaitStyle.TIME_PLUS_EXIT: + # The idea behind this command style is that we want to wait until the process has + # exited and we want to support 'exit' timeout values that are not long and arbitrary. + # In order to do that, we wait for a lull in the console output before waiting + # for the process exit. So, the exit timeout can hopefully be relative to the + # finishing of the console output. + # Of course, if the app doesn't support the "exit" command, there is no sense in + # waiting for the process to respond to it. But, we tell users that we skipped it. + await wait_for_console_output_lull(cmd_start_time, wait_params, shared_data) + if "exit" in target_proc_info.supported_commands: + sleep_interval: float = wait_params.timeout_waiting_for_exit / 10 + for idx in range(10): + if target_proc.returncode is not None: + break + await asyncio.sleep(sleep_interval) + if target_proc.returncode is None: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] WARNING: timeout waiting for {proc_name} to exit in response to {cmd_list}") + else: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] The {proc_name} process doesn't support the 'exit' command, so waiting for exit was skipped") + elif wait_params.style == CommandWaitStyle.ECHO: + if "echo" in target_proc_info.supported_commands: + shared_data.cmd_cmplt_evt.clear() + target_proc.stdin.write((f"echo '{PROCESS_ECHO_STRING}'\n").encode()) + await target_proc.stdin.drain() + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] Sent command to {proc_name}: echo '{PROCESS_ECHO_STRING}'") + await shared_data.cmd_cmplt_evt.wait() + shared_data.cmd_cmplt_evt.clear() + else: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] The {proc_name} process doesn't support the 'echo' command, using TIME wait instead'") + await wait_for_console_output_lull(cmd_start_time, wait_params, shared_data) + else: # treat everything else as wait_params.style == CommandWaitStyle.TIME: + await wait_for_console_output_lull(cmd_start_time, wait_params, shared_data) + + +async def intg_process_manager(daq_session_ingredients: DAQSessionIngredients, run_dir, + verbosity_level): + processes = {} + tasks = {} + command_completion_event = asyncio.Event() + proc_results = {} + shared_data: CommandProcessingSharedData = CommandProcessingSharedData() + + # 1. Start all subprocesses + for session_app in daq_session_ingredients.applications: + proc_name = session_app.alias + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print() + print(f"[integtest_proc_mgmt {now_string}] Starting \"{session_app.startup_strings}\" with process name \"{proc_name}\"...") + #print() + else: + print(".", end="") + proc = await asyncio.create_subprocess_exec( + *session_app.startup_strings, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + cwd=run_dir + ) + processes[proc_name] = RunningProcessInfo(proc) + + # 2. Schedule output reading tasks to run concurrently + tasks[proc_name] = asyncio.create_task(read_stream(proc.stdout, proc_name, session_app.startup_strings[0], + (len(daq_session_ingredients.applications)>1), + run_dir, shared_data, verbosity_level + )) + + time.sleep(session_app.wait_time_after_start) + + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print() + print(f"[integtest_proc_mgmt {now_string}] Started {len(processes)} process(es).") + print() + else: + async with shared_data.lock: + if shared_data.number_of_lines_printed_to_the_console == 0: + print(".", end="") + + # determine the supported commands for each app (using the 'help' command) + help_cmd = ["help"] + help_cmd_wait_params = CommandWaitParameters(timeout_waiting_for_first_msg=2) + await wait_for_console_output_lull(time.time(), help_cmd_wait_params, shared_data) + for proc_name, proc_info in processes.items(): + async with shared_data.lock: + shared_data.results_of_parsing_help_output = [] + shared_data.parsing_of_help_output_in_progress = True + await send_commands(proc_info, proc_name, shared_data, help_cmd, + help_cmd_wait_params, verbosity_level) + async with shared_data.lock: + shared_data.parsing_of_help_output_in_progress = False + proc_info.supported_commands = shared_data.results_of_parsing_help_output + shared_data.results_of_parsing_help_output = [] + + # 3. Send the commands to the running process(es) + return_code = 0 + try: + for cmd_set in daq_session_ingredients.commands: + target = cmd_set.target + if target in processes: + proc_info = processes[target] + + # re-organize the DAQ commands to provide valid combinations, if needed + working_cmd_list = [] + working_cmd = "" + # we work backward thru the list so that we can add arguments to commands + for daq_cmd in reversed(cmd_set.command_list): + daq_cmd = daq_cmd.strip() + # if the number of words is > 1, then we trust that the user specified the full command + if len(daq_cmd.split()) > 1: + working_cmd_list.append(daq_cmd) + else: + # check if the "cmd" is a number; if so, we expect it to be an argument + try: + int(daq_cmd) + working_cmd = " " + daq_cmd + working_cmd + continue + except: + pass + # check if the "cmd" starts with double-dash; if so, consider it an argument + if daq_cmd.startswith("--"): + working_cmd = " " + daq_cmd + working_cmd + continue + # check if the "cmd" is not one of the known supported commands + # if not, we consider it an argument to an application command + if len(proc_info.supported_commands) > 0: + if daq_cmd not in proc_info.supported_commands: + working_cmd = " " + daq_cmd + working_cmd + continue + + # Here we assemble valid application commands. + # If there is a non-empty "working" cmd string, add it to the + # current command as its arguments. Otherwise, the "cmd" stands + # alone and gets added to the list with no arguments. + if len(working_cmd) > 0: + working_cmd = daq_cmd + working_cmd + working_cmd_list.append(working_cmd) + working_cmd = "" + else: + working_cmd_list.append(daq_cmd) + # restore the intended order of the commands to be sent to the process + # (The "reversed" function returns an iterator that can only be used once. + # It seems safer to assign a fully-formed list to the this variable, + # so, we create a new list from the iterator.) + reformatted_cmd_list = list(reversed(working_cmd_list)) + + await send_commands(proc_info, target, shared_data, reformatted_cmd_list, + cmd_set.wait_params, verbosity_level) + + else: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"[integtest_proc_mgmt {now_string}] Error: Process '{target}' not found.") + + except asyncio.CancelledError: + print(f"\n[integtest_proc_mgmt {now_string}] Received CancelledError...") + pass + finally: + async with shared_data.lock: + if shared_data.number_of_lines_printed_to_the_console > 0: + print("---------- DAQ Session END ----------", flush=True) + print("", flush=True) + elif verbosity_level >= IntegtestVerbosityLevels.drunc_boot_terminate: + # huh? + print("", flush=True) + + # 4. Cleanup and terminate remaining processes + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"\n[integtest_proc_mgmt {now_string}] Shutting down processes...") + for proc_name, proc_info in reversed(processes.items()): + if proc_info.process.returncode is None: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + now_string = datetime.now(timezone.utc).strftime("%H:%M:%SZ") + print(f"\n[integtest_proc_mgmt {now_string}] Terminating the {proc_name} process...") + proc_info.process.terminate() + await proc_info.process.wait() + proc_results[proc_name] = {"returncode": proc_info.process.returncode} + + # Cancel background reading tasks + for proc_name, task in tasks.items(): + try: + task.cancel() + process_output = await task + proc_results[proc_name]["stdout"] = process_output + except asyncio.CancelledError: + proc_results[proc_name]["stdout"] = "asyncio.CancelledError" + except asyncio.InvalidStateError: + proc_results[proc_name]["stdout"] = "asyncio.InvalidStateError" + except asyncio.TimeoutError: + proc_results[proc_name]["stdout"] = "asyncio.TimeoutError" + + return proc_results diff --git a/src/integrationtest/data_classes.py b/src/integrationtest/data_classes.py index 33431bf..b5da940 100644 --- a/src/integrationtest/data_classes.py +++ b/src/integrationtest/data_classes.py @@ -1,5 +1,6 @@ from dataclasses import dataclass, field from enum import Enum +import asyncio @dataclass class DROMap_config: @@ -126,3 +127,49 @@ class CreateConfigResult: data_dirs: list[str] tpstream_data_dirs: list[str] trmon_data_dirs: list[str] + + +class CommandWaitStyle(Enum): + ECHO = "echo" + TIME = "time" + TIME_PLUS_EXIT = "time_plus_exit" + NONE = "none" + +@dataclass +class CommandWaitParameters: + wait_for_command_completion: bool = True + style: CommandWaitStyle = CommandWaitStyle.TIME + timeout_waiting_for_first_msg: int = 2 # seconds + wait_time_after_last_msg: int = 2 # seconds + timeout_waiting_for_exit: int = 5 # seconds + +@dataclass +class DAQControlApplication: + alias: str + startup_strings: list[str] + wait_time_after_start: int = 2 # seconds + +@dataclass +class DAQCommandSet: + target: str + command_list: list[str] + wait_params: CommandWaitParameters = field(default_factory=lambda: CommandWaitParameters()) + +@dataclass +class DAQSessionIngredients: + applications: list[DAQControlApplication] + commands: list[DAQCommandSet] + +@dataclass +class RunningProcessInfo: + process: asyncio.subprocess.Process + supported_commands: list[str] = field(default_factory=list) + +@dataclass +class CommandProcessingSharedData: + lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False) + cmd_cmplt_evt: asyncio.Event = field(default_factory=asyncio.Event, repr=False) + last_msg_time: int = 0 + number_of_lines_printed_to_the_console: int = 0 + parsing_of_help_output_in_progress: bool = False + results_of_parsing_help_output: list[str] = field(default_factory=list) diff --git a/src/integrationtest/integrationtest_commandline.py b/src/integrationtest/integrationtest_commandline.py index 1421e2b..9fda40d 100644 --- a/src/integrationtest/integrationtest_commandline.py +++ b/src/integrationtest/integrationtest_commandline.py @@ -49,13 +49,6 @@ def pytest_addoption(parser): help="This controls the volume of messages that are printed out by the integration test infrastructure (1 is lowest, 6 is highest)", required=False ) - parser.addoption( - "--dunerc-fullprint-watch-string", - action="store", - default="", - help="A phrase that, if found in run control messages, will trigger the printout of all RC messages", - required=False - ) parser.addoption( "--remove-hdf5-files", action="store", diff --git a/src/integrationtest/integrationtest_drunc.py b/src/integrationtest/integrationtest_drunc.py index e9911ed..a6e7bdd 100644 --- a/src/integrationtest/integrationtest_drunc.py +++ b/src/integrationtest/integrationtest_drunc.py @@ -5,25 +5,18 @@ import os import re import sys +import time +import asyncio +import random +import json +import copy from io import StringIO import conffwk from integrationtest.integrationtest_commandline import file_exists from integrationtest.resource_validation import ResourceValidator -from integrationtest.verbosity_helper import ( - VerbosityHelper, - IntegtestVerbosityLevels, -) -from integrationtest.data_classes import ( - CreateConfigResult, - config_substitution, - attribute_substitution, - relationship_substitution, - list_element_substitution, - list_element_addition, - ConnSvcControl, - integtest_params_for_generated_dunedaq_config, - integtest_params_for_predefined_dunedaq_config, -) +from integrationtest.verbosity_helper import * +from integrationtest.data_classes import * +from integrationtest.async_proc_mgmt import * from integrationtest.utility_functions import delete_file from daqconf.generate_hwmap import generate_hwmap from daqconf.generate import ( @@ -48,14 +41,11 @@ get_session_env_var, ) from daqconf.get_session_apps import get_segment_apps -import time -import random -import json -# keep track of the number of parametrizations (for various display uses) -total_paramtrization_combinations = 0 -parametrization_counter = 0 +# keep track of the number of parameterizations (for various display uses) +total_parameterization_combinations = 0 +parameterization_counter = 0 def parametrize_fixture_with_items(metafunc, fixture, itemsname): @@ -121,16 +111,24 @@ def pytest_generate_tests(metafunc): parametrize_fixture_with_items(metafunc, "create_config_files", "confgen_arguments") parametrize_fixture_with_items(metafunc, "process_manager_type", "process_manager_choices") - parametrize_fixture_with_items(metafunc, "run_dunerc", "dunerc_command_list") + if hasattr(metafunc.module, "daq_session_ingredients"): + parametrize_fixture_with_items(metafunc, "run_dunerc", "daq_session_ingredients") + else: + parametrize_fixture_with_items(metafunc, "run_dunerc", "dunerc_command_list") - # determine the number of different parametrizations + # determine the number of different parameterizations # (recall that this fixture is called once per pytest function in each integtest) # (we only need to calculate this value once, so we check the initial value of zero) - global total_paramtrization_combinations - if total_paramtrization_combinations == 0: - total_paramtrization_combinations = len(metafunc.module.confgen_arguments) * len(metafunc.module.process_manager_choices) - if type(metafunc.module.dunerc_command_list) is dict: - total_paramtrization_combinations *= len(metafunc.module.dunerc_command_list) + global total_parameterization_combinations + if total_parameterization_combinations == 0: + total_parameterization_combinations = len(metafunc.module.confgen_arguments) * len(metafunc.module.process_manager_choices) + if hasattr(metafunc.module, "daq_session_ingredients"): + if type(metafunc.module.daq_session_ingredients) is dict: + total_parameterization_combinations *= len(metafunc.module.daq_session_ingredients) + elif hasattr(metafunc.module, "dunerc_command_list"): + if type(metafunc.module.dunerc_command_list) is dict: + total_parameterization_combinations *= len(metafunc.module.dunerc_command_list) + @pytest.fixture(scope="module") def process_manager_type(request): @@ -150,7 +148,6 @@ def create_config_files(request, tmp_path_factory, check_system_resources): produced by one pytest module """ - dummy_resource_check = check_system_resources integtest_params = request.param #if isinstance(integtest_params, integtest_params_for_generated_dunedaq_config): @@ -175,12 +172,12 @@ def create_config_files(request, tmp_path_factory, check_system_resources): integtest_params.daq_session_name = integtest_params.config_session_name # 26-Mar-2026, KAB: suppress output messages, if requested - integtest_verbosity_level = int(request.config.getoption("--integtest-verbosity")) - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + verbosity_level = int(request.config.getoption("--integtest-verbosity")) + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: print("", flush=True) original_stdout = sys.stdout - if integtest_verbosity_level < IntegtestVerbosityLevels.full_output: - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + if verbosity_level < IntegtestVerbosityLevels.full_output: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: print("----------------------------------------", flush=True) print("*** Messages related to configuration generation have been suppressed ***", flush=True) print("----------------------------------------", flush=True) @@ -384,7 +381,7 @@ def apply_update(obj, substitution): ) # restore the usual stdout behavior, if needed - if integtest_verbosity_level < IntegtestVerbosityLevels.full_output: + if verbosity_level < IntegtestVerbosityLevels.full_output: sys.stdout = original_stdout else: print("", flush=True) @@ -403,8 +400,14 @@ def run_dunerc(request, create_config_files, process_manager_type, trace_debug_s """ run_control_commands = request.param + # determine which type of request this is, either a list of commands for dunerc or a more + # sophisticated list of applications to be started and the commands to be sent to them + user_supplied_apps = False + if type(run_control_commands) is DAQSessionIngredients: + user_supplied_apps = True + no_integtest_connsvc = request.config.getoption("--no-integtest-connsvc") - integtest_verbosity_level = int(request.config.getoption("--integtest-verbosity")) + verbosity_level = int(request.config.getoption("--integtest-verbosity")) if no_integtest_connsvc and \ isinstance(create_config_files.integtest_params, integtest_params_for_generated_dunedaq_config): @@ -412,21 +415,25 @@ def run_dunerc(request, create_config_files, process_manager_type, trace_debug_s run_dir = tmp_path_factory.mktemp("run") - global total_paramtrization_combinations - if total_paramtrization_combinations > 1: - global parametrization_counter - parametrization_counter += 1 - if parametrization_counter > 1: - if integtest_verbosity_level > IntegtestVerbosityLevels.just_errors_and_warnings and \ - integtest_verbosity_level < IntegtestVerbosityLevels.integtest_debug: + global total_parameterization_combinations + if total_parameterization_combinations > 1: + global parameterization_counter + parameterization_counter += 1 + if parameterization_counter > 1: + if verbosity_level > IntegtestVerbosityLevels.just_errors_and_warnings and \ + verbosity_level < IntegtestVerbosityLevels.integtest_debug: print("", flush=True) print("", flush=True) - if integtest_verbosity_level > IntegtestVerbosityLevels.just_errors_and_warnings: + if verbosity_level > IntegtestVerbosityLevels.just_errors_and_warnings: current_test = os.environ.get("PYTEST_CURRENT_TEST") - match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d\].*", current_test) if match_obj: current_test = match_obj.group(1) + else: + match_obj = re.search(r".*\[(.+)\].*", current_test) + if match_obj: + current_test = match_obj.group(1) print(f"-> {current_test} <-") @@ -466,7 +473,7 @@ def run_dunerc(request, create_config_files, process_manager_type, trace_debug_s and create_config_files.integtest_params.connsvc_control == ConnSvcControl.INTEGRATIONTEST ): # start connsvc - if integtest_verbosity_level >= IntegtestVerbosityLevels.full_output: + if verbosity_level >= IntegtestVerbosityLevels.full_output: print( f"Starting Connectivity Service on port {create_config_files.integtest_params.connsvc_port}" ) @@ -523,7 +530,7 @@ class RunResult: # suppress output, if requested original_stdout = sys.stdout - if integtest_verbosity_level < IntegtestVerbosityLevels.full_output: + if verbosity_level < IntegtestVerbosityLevels.full_output: sys.stdout = catcher = StringIO() for path in rawdata_paths: @@ -588,103 +595,80 @@ class RunResult: file_obj.unlink(True) # missing is OK # restore the usual stdout behavior, if needed - if integtest_verbosity_level < IntegtestVerbosityLevels.full_output: + if verbosity_level < IntegtestVerbosityLevels.full_output: sys.stdout = original_stdout + exit_cmd = DAQCommandSet("drunc", [ "exit" ], CommandWaitParameters(style=CommandWaitStyle.TIME_PLUS_EXIT)) + if user_supplied_apps: + dsi = copy.deepcopy(run_control_commands) + for app in dsi.applications: + # replace placeholders in the command startup strings + for idx in range(len(app.startup_strings)): + if app.startup_strings[idx] == "": + app.startup_strings[idx] = str(process_manager_type) + continue + if app.startup_strings[idx] == "": + app.startup_strings[idx] = str(create_config_files.dunedaq_config_file) + continue + if app.startup_strings[idx] == "": + app.startup_strings[idx] = str(create_config_files.integtest_params.config_session_name) + continue + if app.startup_strings[idx] == "": + app.startup_strings[idx] = str(create_config_files.integtest_params.daq_session_name) + continue + + # include requested options in the startup strings for the apps that support them + # (This relies on the apps failing if we pass them unsupported options and not failing + # if the requested options are supported. Of course, this is not perfect because there + # can be multiple options in a single list, and we don't do the work to see if some of + # them are supported but not others.) + if len(dunerc_option_strings) > 0: + help_cmds = [app.startup_strings[0]] + dunerc_option_strings + ["--help"] + result = subprocess.run(help_cmds, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + if result.returncode == 0: + app.startup_strings[1:1] = dunerc_option_strings + if len(create_config_files.integtest_params.dunerc_cmd_args) > 0: + help_cmds = [app.startup_strings[0]] + create_config_files.integtest_params.dunerc_cmd_args + ["--help"] + result = subprocess.run(help_cmds, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + if result.returncode == 0: + app.startup_strings[1:1] = create_config_files.integtest_params.dunerc_cmd_args + + # add an exit command to the end of the command list, for each app, in reverse order + for app in reversed(dsi.applications): + tmp_exit_cmd = copy.deepcopy(exit_cmd) + tmp_exit_cmd.target = app.alias + dsi.commands.append(tmp_exit_cmd) + + else: + popen_command_list = [dunerc] + create_config_files.integtest_params.dunerc_cmd_args \ + + dunerc_option_strings + [process_manager_type] + [str(create_config_files.dunedaq_config_file)] \ + + [str(create_config_files.integtest_params.config_session_name)] \ + + [str(create_config_files.integtest_params.daq_session_name)] + + dsapp = DAQControlApplication("drunc", popen_command_list) + + requested_cmds = DAQCommandSet("drunc", run_control_commands, CommandWaitParameters(style=CommandWaitStyle.ECHO)) + + app_list = [ dsapp ] + cmd_set_list = [ requested_cmds, exit_cmd ] + dsi = DAQSessionIngredients(app_list, cmd_set_list) + result = RunResult() time_before = time.time() - # 25-Mar-2026, KAB: use subprocess.Popen to manage the run control session so that we can - # capture the console output and pass it back to the user for inspection and validation. - popen_command_list = [dunerc] + create_config_files.integtest_params.dunerc_cmd_args \ - + dunerc_option_strings + [process_manager_type] + [str(create_config_files.dunedaq_config_file)] \ - + [str(create_config_files.integtest_params.config_session_name)] \ - + [str(create_config_files.integtest_params.daq_session_name)] + run_control_commands - rc_process = subprocess.Popen( - popen_command_list, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=1, - cwd=run_dir - ) - # print out each line of captured output, subject to the verbosity level that the - # user has requested, as well as add it to the string that we pass back to the user - tmp_string = request.config.getoption("--dunerc-fullprint-watch-string") - full_printout_watch_string = tmp_string.replace("_SPC_", " ") - full_printout_activated = False - number_of_lines_printed_to_the_console = 0 - full_output = "" - for line in rc_process.stdout: - should_be_printed = integtest_verbosity_level >= IntegtestVerbosityLevels.full_output or \ - full_printout_activated - - # check for a user-specified string that triggers full printout - # (this check needs to come first so that it sees the initial value of "should_be_printed") - if should_be_printed == False and len(full_printout_watch_string) > 0: - if re.search(full_printout_watch_string, line): - if number_of_lines_printed_to_the_console == 0: - print("\n++++++++++ DRUNC Session BEGIN ++++++++++", flush=True) - else: - print("++++++++++ Switching to full DRUNC output mode ++++++++++", flush=True) - print( - f"+++ Displaying all DRUNC messages based on the presence of phrase \"{full_printout_watch_string}\" +++", - flush=True - ) - print(full_output) # messages captured so far - full_printout_activated = True - should_be_printed = True - number_of_lines_printed_to_the_console = 1 # probably more, but good enough - - # check for errors and warnings for all verbosity levels - if should_be_printed == False: - lc_line = line.lower() - if ("error" in lc_line and (not "In error" in line and not "Endpoint" in line)) \ - or "warning" in lc_line or "critical" in lc_line: - should_be_printed = True - - # check for basic transition messages, if that level of verbosity is requested - if should_be_printed == False: - if integtest_verbosity_level >= IntegtestVerbosityLevels.drunc_boot_terminate: - if "Booting session" in line or \ - ("Current FSM status is " in line and ("initial" in line or "running" in line)): - should_be_printed = True - - # check for all transition messages, if that level of verbosity is requested - if should_be_printed == False: - if integtest_verbosity_level >= IntegtestVerbosityLevels.drunc_transitions: - if "Booting session" in line or "Running transition" in line \ - or ("wait" in line and "running" in line) or "exit code" in line: - should_be_printed = True - - # actually do the printout - if should_be_printed: - if number_of_lines_printed_to_the_console == 0: - print( - "++++++++++ DRUNC Session BEGIN ++++++++++", flush=True - ) # Apparently need to flush before subprocess.run - print(line, end='', flush=True) - number_of_lines_printed_to_the_console += 1 - full_output += line - - rc_process.communicate() - proc_returncode = rc_process.returncode - - # store the full dunerc console output in a log file for reference and checking - with open(f"{run_dir}/log_{getpass.getuser()}_drunc_console_output.txt", "w", encoding="utf-8") as ff: - no_ansi_output = re.sub(r"\x1b\[[0-9;]*m", "", full_output) - ff.write(no_ansi_output) - - # construct a CompletedProcess instance to be passed back to the user. In this way, - # user code does not need to change in response to the change in this code from - # using subprocess.run() to subprocess.Popen(). - result.completed_process = subprocess.CompletedProcess( - popen_command_list, - returncode=proc_returncode, - stdout=full_output - ) + proc_results = asyncio.run(intg_process_manager(dsi, run_dir, verbosity_level)) + time_after = time.time() + # construct a CompletedProcess instance for each application that was run. + result.completed_processes = {} + for app in dsi.applications: + result.completed_processes[app.alias] = subprocess.CompletedProcess( + app.startup_strings, + returncode=proc_results[app.alias]["returncode"], + stdout=proc_results[app.alias]["stdout"] + ) + if connsvc_obj is not None: time.sleep(1) connsvc_obj.send_signal(2) @@ -700,12 +684,6 @@ class RunResult: ) subprocess.run(["killall", "gunicorn", "drunc-controller"]) - if number_of_lines_printed_to_the_console > 0: - print("---------- DRUNC Session END ----------", flush=True) - print("", flush=True) - elif integtest_verbosity_level >= IntegtestVerbosityLevels.drunc_boot_terminate: - print("", flush=True) - result.confgen_config = create_config_files.integtest_params result.config_session_name = create_config_files.integtest_params.config_session_name result.daq_session_name = create_config_files.integtest_params.daq_session_name @@ -732,7 +710,7 @@ class RunResult: # 10-Dec-2025, KAB: added the DAQ session overall time so that we can use this # information in fine-tuning the allowed ranges in time-based checking of test results. result.daq_session_overall_time = time_after - time_before - result.verbosity_helper = VerbosityHelper(integtest_verbosity_level) + result.verbosity_helper = VerbosityHelper(verbosity_level) # pass the names of the HDF5 files to the 'cleanup' fixture cleanup_hdf5_files["raw"] = result.data_files @@ -755,16 +733,16 @@ def check_system_resources(request): the recommended resources are not present, then a warning is printed """ skip_resource_checks = request.config.getoption("--skip-resource-checks") - integtest_verbosity_level = int(request.config.getoption("--integtest-verbosity")) + verbosity_level = int(request.config.getoption("--integtest-verbosity")) # print out a couple of blank lines to help with formatting - if integtest_verbosity_level > IntegtestVerbosityLevels.just_errors_and_warnings: + if verbosity_level > IntegtestVerbosityLevels.just_errors_and_warnings: print("", flush=True) print("", flush=True) resval = getattr(request.module, "resource_validator", ResourceValidator()) - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: resval_debug_string = resval.get_debug_string() print(resval_debug_string) @@ -777,7 +755,7 @@ def check_system_resources(request): del request.session.items[1:] pytest.skip(f"\n\N{LARGE YELLOW CIRCLE} {resval_report_string}") if not resval.recommended_resources_are_present: - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: resval_report_string = resval.get_recommended_resources_report() print(f"\n*** Note: {resval_report_string}") @@ -786,7 +764,7 @@ def check_system_resources(request): # 16-Feb-2026, KAB: added a printout for recommended resources after the "yield" # statement so that it gets printed out at the end of the output that the user sees. if not resval.recommended_resources_are_present: - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: resval_report_string = resval.get_recommended_resources_report() print(f"\n*** Note: {resval_report_string}") @@ -912,7 +890,7 @@ def cleanup_hdf5_files(request, create_config_files): yield file_lists # here is where the work gets done... - integtest_verbosity_level = int(request.config.getoption("--integtest-verbosity")) + verbosity_level = int(request.config.getoption("--integtest-verbosity")) user_requests_hdf5_file_removal = request.config.getoption("--remove-hdf5-files") the_test_requests_hdf5_file_removal = create_config_files.integtest_params.remove_hdf5_files @@ -944,7 +922,7 @@ def cleanup_hdf5_files(request, create_config_files): pathlist_string += " " + str(data_file.parent) if pathlist_string and filelist_string: - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: print("============================================") print("Listing the hdf5 files before deleting them:") print("============================================") @@ -960,7 +938,7 @@ def cleanup_hdf5_files(request, create_config_files): for data_file in file_lists["trmon"]: delete_file(data_file) - if integtest_verbosity_level >= IntegtestVerbosityLevels.integtest_debug: + if verbosity_level >= IntegtestVerbosityLevels.integtest_debug: print("--------------------") os.system(f"df -h {pathlist_string}") print("============================================") diff --git a/src/integrationtest/log_file_checks.py b/src/integrationtest/log_file_checks.py index d9402fd..2d4261b 100644 --- a/src/integrationtest/log_file_checks.py +++ b/src/integrationtest/log_file_checks.py @@ -116,14 +116,17 @@ def logs_are_error_free(log_file_names, show_all_problems=True, print_logfilenam ["LogLevel=error", r'key:\s+"DUNEDAQ_ERS_', r"drunc.utils.ConnectivityServiceClient\s+404 Client Error: NOT FOUND for url:"] ) - local_excl_string_map.setdefault("drunc", []).extend( + local_excl_string_map.setdefault("drunc-unified-shell", []).extend( ["LogLevel=error", r'key:\s+"DUNEDAQ_ERS_', r"DUNEDAQ_ERS_.*erstrace", "export DUNEDAQ_ERS_", r"NewConnectionError.* Failed to establish a new connection: \[Errno 111\] Connection refused", r"drunc.utils.ConnectivityServiceClient\s+404 Client Error: NOT FOUND for url:"] ) + local_excl_string_map.setdefault("drunc-process-manager", []).extend( + ["LogLevel=error", r'key:\s+"DUNEDAQ_ERS_', r"DUNEDAQ_ERS_.*erstrace", "export DUNEDAQ_ERS_"] + ) # 21-Jul-2026, KAB: phrases that we always want to exclude - local_excl_string_map.setdefault("drunc", []).extend(["Substate.*In error.*Endpoint"]) + local_excl_string_map.setdefault("drunc-unified-shell", []).extend(["Substate.*In error.*Endpoint"]) all_ok=True #print("") # Clear potential dot from pytest diff --git a/src/integrationtest/utility_functions.py b/src/integrationtest/utility_functions.py index f754388..a716b0b 100644 --- a/src/integrationtest/utility_functions.py +++ b/src/integrationtest/utility_functions.py @@ -15,7 +15,7 @@ def basic_checks(run_dunerc, caplog, print_test_name: bool = True): if print_test_name and run_dunerc.verbosity_helper.compare_level(IntegtestVerbosityLevels.drunc_transitions): # print the name of the current test current_test = os.environ.get("PYTEST_CURRENT_TEST") - match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d\].*", current_test) if match_obj: current_test = match_obj.group(1) banner_line = re.sub(".", "=", current_test) @@ -24,9 +24,10 @@ def basic_checks(run_dunerc, caplog, print_test_name: bool = True): print(banner_line) # Check that dunerc completed correctly - if run_dunerc.completed_process.returncode != 0: - fail_msg = f"The run control session returned a non-zero status code ({run_dunerc.completed_process.returncode})." - pytest.fail(fail_msg, pytrace=False) + for proc_name, cmplt_proc in run_dunerc.completed_processes.items(): + if cmplt_proc.returncode != 0: + fail_msg = f"The {proc_name} process returned a non-zero status code ({cmplt_proc.returncode})." + pytest.fail(fail_msg, pytrace=False) # Check that there weren't any warnings or errors during setup setup_logs = caplog.get_records("setup") diff --git a/src/integrationtest/verbosity_helper.py b/src/integrationtest/verbosity_helper.py index ebea7d2..cda4ed2 100644 --- a/src/integrationtest/verbosity_helper.py +++ b/src/integrationtest/verbosity_helper.py @@ -6,8 +6,8 @@ class IntegtestVerbosityLevels: just_errors_and_warnings: int = 1 # shows just errors and warnings drunc_boot_terminate: int = 2 # adds a small subset of drunc transitions drunc_transitions: int = 3 # adds all drunc transitions plus validation check results - integtest_debug: int = 4 # adds ResourceValidation debug info and any other integtest debug - full_output: int = 5 # shows everything + integtest_debug: int = 4 # adds ResourceValidation debug info and other integtest debug + full_output: int = 5 # shows all drunc output drunc_debug: int = 6 # enables drunc debug messages class VerbosityHelper: