Added an integtest to demonstrate user-specified control applications...

src/drunc/integtest/basic_multiapp_test.py

@@ -0,0 +1,168 @@
+# 05-Aug-2026, KAB: the goal of this test is to validate and demonstrate the use of multiple
+# user-specified applications running in the DAQ session that is part of this test.
+#
+# This integtest was created by copying the small_footprint_quick_test from the daqsystemtest
+# repo and converting the assignment of the run control commands to make use of the new
+# "daq_session_ingredients" special integtest variable.
+#
+import pytest
+import urllib.request
+
+from integrationtest.data_classes import *
+import integrationtest.data_file_checks as data_file_checks
+import integrationtest.log_file_checks as log_file_checks
+import integrationtest.resource_validation as resource_validation
+import integrationtest.utility_functions as utility_functions
+from integrationtest.get_pytest_tmpdir import get_pytest_tmpdir
+from integrationtest.verbosity_helper import IntegtestVerbosityLevels
+from daqconf.utils import find_free_port
+
+import functools
+print = functools.partial(print, flush=True)  # always flush print() output
+
+pytest_plugins = "integrationtest.integrationtest_drunc"
+
+# Values that help determine the running conditions
+number_of_data_producers = 1
+run_duration = 20  # seconds
+
+# Default values for validation parameters
+expected_number_of_data_files = 1
+check_for_logfile_errors = True
+expected_event_count = run_duration
+expected_event_count_tolerance = 2
+wibeth_frag_params = {
+    "fragment_type_description": "WIBEth",
+    "fragment_type": "WIBEth",
+    "expected_fragment_count": number_of_data_producers,
+    "min_size_bytes": 14472,
+    "max_size_bytes": 21672,
+}
+triggercandidate_frag_params = {
+    "fragment_type_description": "Trigger Candidate",
+    "fragment_type": "Trigger_Candidate",
+    "expected_fragment_count": 1,
+    "min_size_bytes": 128,
+    "max_size_bytes": 216,
+}
+hsi_frag_params = {
+    "fragment_type_description": "HSI",
+    "fragment_type": "Hardware_Signal",
+    "expected_fragment_count": 1,
+    "min_size_bytes": 100,
+    "max_size_bytes": 100,
+}
+ignored_logfile_problems = {
+    "connectionservice": [
+        "Searching for connections matching uid_regex<errored_frames_q> and data_type Unknown"
+    ],
+    "-controller": [
+        "Worker with pid \\d+ was terminated due to signal 1",
+        "Connection '.*' not found on the application registry",
+    ],
+    "connectivity-service": [
+        "errorlog: -",
+    ],
+}
+
+# Determine if this computer has enough resources for these tests
+resource_validator = resource_validation.ResourceValidator()
+resource_validator.cpu_count_needs(4, 8)  # 2 for each data source plus 2 more for everything else
+resource_validator.free_memory_needs(4, 6)  # 33% more than what we observe being used ('free -h')
+actual_output_path = get_pytest_tmpdir()
+resource_validator.free_disk_space_needs(actual_output_path, 1)  # more than what we observe
+
+# The arguments to pass to the config generator, excluding the json
+# output directory (the test framework handles that)
+
+conf_dict = integtest_params_for_generated_dunedaq_config()
+conf_dict.object_databases = ["config/daqsystemtest/integrationtest-objects.data.xml"]
+conf_dict.dro_map_config.n_streams = number_of_data_producers
+conf_dict.op_env = "integtest"
+conf_dict.config_session_name = "smallfootprint"
+conf_dict.tpg_enabled = False
+utility_functions.enable_fake_hsi_trigger(conf_dict, trigger_rate=1.0)
+
+conf_dict.config_substitutions.append(
+    attribute_substitution(obj_class="LatencyBuffer", updates={"size": 50000})
+)
+
+confgen_arguments = {"SmallFootprint": conf_dict}
+
+# 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", "<proc_mgr_choice>", str(pm_port)]
+pmapp = DAQSessionApp("pm", procmsg_startup_commands)
+
+pmshell_startup_commands = ["drunc-process-manager-shell", f"grpc://localhost:{pm_port}"]
+pmshellapp = DAQSessionApp("pmshell", pmshell_startup_commands)
+
+drunc_startup_commands = ["drunc-unified-shell", f"grpc://localhost:{pm_port}", "<config_data_file>", "<config_session_name>", "<daq_session_name>"]
+druncapp = DAQSessionApp("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}
+
+
+# The tests themselves
+
+
+def test_dunerc_success(run_dunerc, caplog):
+    # checks for run control success, problems during pytest setup, etc.
+    utility_functions.basic_checks(run_dunerc, caplog, print_test_name=False)
+
+
+def test_log_files(run_dunerc):
+    if check_for_logfile_errors:
+        # Check that there are no warnings or errors in the log files
+        assert log_file_checks.logs_are_error_free(
+            run_dunerc.log_files, True, True, ignored_logfile_problems,
+            verbosity_helper=run_dunerc.verbosity_helper
+        )
+
+
+def test_data_files(run_dunerc):
+    # Run some tests on the output data file
+    assert len(run_dunerc.data_files) == expected_number_of_data_files
+
+    fragment_check_list = [triggercandidate_frag_params, hsi_frag_params]
+    fragment_check_list.append(wibeth_frag_params)  # WIBEth
+
+    all_ok = True
+    for idx in range(len(run_dunerc.data_files)):
+        data_file = data_file_checks.DataFile(run_dunerc.data_files[idx], run_dunerc.verbosity_helper)
+        all_ok &= data_file_checks.sanity_check(data_file)
+        all_ok &= data_file_checks.check_file_attributes(data_file)
+        all_ok &= data_file_checks.check_event_count(
+            data_file, expected_event_count, expected_event_count_tolerance
+        )
+        for jdx in range(len(fragment_check_list)):
+            all_ok &= data_file_checks.check_fragment_count(
+                data_file, fragment_check_list[jdx]
+            )
+            all_ok &= data_file_checks.check_fragment_sizes(
+                data_file, fragment_check_list[jdx]
+            )
+    assert all_ok

src/drunc/integtest/controller_test.py

@@ -161,7 +161,7 @@ def boot_status_table(run_dunerc):
     Scoped to the module so every test in this file can compare against the
     same baseline without re-parsing stdout each time.
     """
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
     return get_status_table_after_echo(lines, "post_boot")
 
 
@@ -223,7 +223,7 @@ def test_dunerc_success(run_dunerc) -> None:
     print(current_test)
     print(banner_line)
 
-    assert run_dunerc.completed_process.returncode == 0
+    assert run_dunerc.completed_processes["drunc"].returncode == 0
 
 
 def test_log_files(run_dunerc) -> None:
@@ -250,5 +250,5 @@ def test_log_files(run_dunerc) -> None:
 @pytest.mark.parametrize("params", _FSM_COMMANDS, ids=lambda p: p.marker)
 def test_fsm_command(run_dunerc, boot_status_table, params: FsmCommandParams) -> None:
     """Checks that each FSM command executes successfully and transitions all processes to the expected state."""
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
     _check_command(lines, boot_status_table, params)

src/drunc/integtest/process_manager_test.py

@@ -155,7 +155,7 @@ def test_dunerc_success(run_dunerc) -> None:
     print(banner_line)
 
     # Check that dunerc completed correctly
-    assert run_dunerc.completed_process.returncode == 0
+    assert run_dunerc.completed_processes["drunc"].returncode == 0
 
 
 def test_log_files(run_dunerc) -> None:
@@ -194,7 +194,7 @@ def test_log_files(run_dunerc) -> None:
 
 def test_boot(run_dunerc) -> None:
     """Checks that boot starts the managed processes and exposes UUIDs in ps."""
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
 
     # Check if no processes running in session works
     pre_boot_idx = require_line_containing(
@@ -229,7 +229,7 @@ def test_unknown_log_command(run_dunerc) -> None:
     test_str = (
         "Bad query for logs: The process corresponding to the query doesn't exist"
     )
-    assert test_str in run_dunerc.completed_process.stdout
+    assert test_str in run_dunerc.completed_processes["drunc"].stdout
 
 
 def test_root_controller_logs(run_dunerc) -> None:
@@ -239,7 +239,7 @@ def test_root_controller_logs(run_dunerc) -> None:
     - there are exactly 5 lines between those two lines
     - among those 5 lines, the one from "drunc.controller.core.init_controller" ends with "Controller ready"
     """
-    lines = run_dunerc.completed_process.stdout.splitlines()
+    lines = run_dunerc.completed_processes["drunc"].stdout.splitlines()
 
     # 1) Find the header/footer lines
     header_idx = require_line_containing(
@@ -277,7 +277,7 @@ def test_root_controller_logs(run_dunerc) -> None:
 
 def test_wait_command_duration_from_logs(run_dunerc) -> None:
     """Checks that the wait command logs the expected duration and elapsed time."""
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
 
     echo_idx = require_echo_marker_index(lines, "test_wait")
 
@@ -336,7 +336,7 @@ def test_wait_command_duration_from_logs(run_dunerc) -> None:
 
 def test_restart_mlt_logs(run_dunerc) -> None:
     """Checks that restarting mlt produces the expected restart, exit, and boot logs."""
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
 
     echo_idx = require_echo_marker_index(lines, "pre_restart_mlt")
 
@@ -382,7 +382,7 @@ def test_restart_mlt_logs(run_dunerc) -> None:
 
 def test_kill_removes_mlt_from_ps_table(run_dunerc) -> None:
     """Checks that killing mlt removes it from the subsequent ps table."""
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
 
     ps_before_kill = get_ps_table_after_echo(lines, "test_kill_mlt")
     ps_after_kill = get_ps_table_after_echo(lines, "test_kill_mlt_post")
@@ -395,7 +395,7 @@ def test_kill_removes_mlt_from_ps_table(run_dunerc) -> None:
 
 def test_mlt_recovers_after_kill(run_dunerc) -> None:
     """Checks that mlt is present again after the recovery restart sequence."""
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
     ps_after_recovery = get_ps_table_after_echo(lines, "test_recovery_post")
     assert_process_presence(ps_after_recovery, "mlt", context="after recovery")
 
@@ -404,7 +404,7 @@ def test_flush(run_dunerc) -> None:
     """Checks that flush work by crashing mlt, seeing that the process exists,
     and then flushing to show its gone"""
 
-    lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines()
+    lines = strip_ansi(run_dunerc.completed_processes["drunc"].stdout).splitlines()
     ps_initial = get_ps_table_after_echo(lines, "test_flush")
     assert_process_presence(ps_initial, "mlt", context="before crash")