New regression test to verify that we don't get duplicate TriggerActivities

integtest/no_duplicate_ta_test.py

@@ -0,0 +1,269 @@
+# 07-Jul-2026, KAB: the goal of this test is to check whether duplicate TriggerPrimitives
+# are being produced in a demo system. It does this by reducing the configured TriggerActivity
+# prescale value to 5 and checking if there are complaints in the log files about attemping
+# to insert duplicate TAs (which are produced from duplicate TPs) into the relevant latency buffers.
+#
+# We have been seeing this problem (failed attempts to insert duplicate TAs into the relevant
+# latency buffers) occasionally for quite some time in other regression tests, and we believe
+# that the problem is now fixed with the new "latency buffer returns an iterator" code in
+# DUNE-DAQ/datahandlinglibs#106.
+#
+import pytest
+import urllib.request
+
+import integrationtest.data_file_checks as data_file_checks
+import integrationtest.log_file_checks as log_file_checks
+import integrationtest.utility_functions as utility_functions
+import integrationtest.data_classes as data_classes
+import integrationtest.resource_validation as resource_validation
+from integrationtest.get_pytest_tmpdir import get_pytest_tmpdir
+from integrationtest.verbosity_helper import IntegtestVerbosityLevels
+
+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 = 2
+number_of_readout_apps = 1
+number_of_dataflow_apps = 2
+pulser_trigger_rate = 1.0  # Hz
+run_duration = 30  # seconds
+output_dir = "."
+
+# Default values for validation parameters
+expected_number_of_data_files = 2 * number_of_dataflow_apps
+check_for_logfile_errors = True
+expected_event_count = run_duration * pulser_trigger_rate / number_of_dataflow_apps
+expected_event_count_tolerance = expected_event_count / 10
+
+wibeth_frag_params = {
+    "fragment_type_description": "WIBEth",
+    "fragment_type": "WIBEth",
+    "expected_fragment_count": (number_of_data_producers * number_of_readout_apps),
+    "min_size_bytes": 7272,
+    "max_size_bytes": 14472,
+    "debug_mask": 0x0,
+}
+wibeth_tpset_params = {
+    "fragment_type_description": "TP Stream",
+    "fragment_type": "Trigger_Primitive",
+    "expected_fragment_count": number_of_readout_apps * 3,
+    "frag_counts_by_record_ordinal": {"first": {"min_count": 1, "max_count": number_of_readout_apps * 3},
+                                      "default": {"min_count": number_of_readout_apps * 3, "max_count": number_of_readout_apps * 3} },
+    "min_size_bytes": 72,
+    "max_size_bytes": 120000,
+    "debug_mask": 0x0,
+    "frag_sizes_by_record_ordinal": {  "first": {"min_size_bytes":    96, "max_size_bytes": 120000},
+                                      "second": {"min_size_bytes":    96, "max_size_bytes": 120000},
+                                        "last": {"min_size_bytes":    96, "max_size_bytes": 120000},
+                                     "default": {"min_size_bytes": 80000, "max_size_bytes": 120000} }
+}
+# sizes: 128 is for one TC with zero TAs inside it (72+56)
+#        208 is for one TC with one TA inside it (72+56+80)
+#        264 is for two TCs with one TA in one of them (72+56+80+56)
+triggercandidate_frag_params = {
+    "fragment_type_description": "Trigger Candidate",
+    "fragment_type": "Trigger_Candidate",
+    "expected_fragment_count": 1,
+    "min_size_bytes": 128,
+    "max_size_bytes": 264,
+    "debug_mask": 0x0,
+    "frag_sizes_by_TC_type": {"kPrescale": {"min_size_bytes": 208, "max_size_bytes": 264},
+                                "kRandom": {"min_size_bytes": 128, "max_size_bytes": 264},
+                                "default": {"min_size_bytes": 128, "max_size_bytes": 264} }
+}
+# sizes:  72 is for an empty TA fragment
+#        184 is for one TA with one TP inside it (72+88+24)
+#        296 is for two TAs with one TP in each of them (72+88+24+88+24)
+#        408 is for three TAs with one TP in each of them (72+88+24+88+24+88+24)
+triggeractivity_frag_params = {
+    "fragment_type_description": "Trigger Activity",
+    "fragment_type": "Trigger_Activity",
+    "expected_fragment_count": 1,
+    "min_size_bytes": 72,
+    "max_size_bytes": 408,
+    "debug_mask": 0x0,
+    "frag_sizes_by_TC_type": {"kPrescale": {"min_size_bytes": 184, "max_size_bytes": 408},
+                                "kRandom": {"min_size_bytes":  72, "max_size_bytes": 296},
+                                "default": {"min_size_bytes":  72, "max_size_bytes": 408} }
+}
+# sizes:  72 is for an empty TP fragment
+#        144 is for a fragment with three TPs in it (72+24+24+24)
+triggerprimitive_frag_params = {
+    "fragment_type_description": "Trigger Primitive",
+    "fragment_type": "Trigger_Primitive",
+    "expected_fragment_count": number_of_readout_apps * 3,
+    "min_size_bytes": 72,
+    "max_size_bytes": 144,
+}
+hsi_frag_params = {
+    "fragment_type_description": "HSI",
+    "fragment_type": "Hardware_Signal",
+    "expected_fragment_count": 0,
+    "min_size_bytes": 72,
+    "max_size_bytes": 100,
+}
+ignored_logfile_problems = {
+    "-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(8, 16)  # 3 for each data source (incl TPG) plus 2 more for everything else
+resource_validator.free_memory_needs(6, 10)  # 20% 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
+
+conf_dict = data_classes.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.dro_map_config.n_apps = number_of_readout_apps
+conf_dict.op_env = "integtest"
+conf_dict.config_session_name = "ta_testing"
+conf_dict.tpg_enabled = True
+conf_dict.n_df_apps = number_of_dataflow_apps
+conf_dict.frame_file = (
+    "asset://?checksum=dd156b4895f1b06a06b6ff38e37bd798"  # WIBEth All Zeros
+)
+
+conf_dict.config_substitutions.append(
+    data_classes.list_element_addition(
+        obj_class="Session",
+        obj_id=conf_dict.config_session_name,
+        rel_name="disabled",
+        additional_object_class="TPStreamWriterApplication",
+        additional_object_id="tp-stream-writer",
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="RandomTCMakerConf",
+        updates={"trigger_rate_hz": pulser_trigger_rate},
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="LatencyBuffer", updates={"size": 200000}
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="TAMakerPrescaleAlgorithm",
+        obj_id="dummy-ta-maker",
+        updates={"prescale": 5},
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="TCMakerPrescaleAlgorithm",
+        obj_id="tc-pass-through-algo",
+        updates={"prescale": 500},
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="TCDataProcessor",
+        obj_id="def-tc-processor",
+        updates={"merge_overlapping_tcs": 0},
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="DataStoreConf",
+        obj_id="default",
+        updates={"directory_path": output_dir},
+    )
+)
+conf_dict.config_substitutions.append(
+    data_classes.attribute_substitution(
+        obj_class="DataStoreConf",
+        obj_id="default_tp_store_conf",
+        updates={"directory_path": output_dir},
+    )
+)
+
+confgen_arguments = {"WIBEth_TPG_System": conf_dict}
+
+# The commands to run in dunerc, as a list
+dunerc_command_list = (
+    "boot conf wait 5".split()
+    + "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 ".split()
+    + "start --run-number 102 wait 1 enable-triggers wait ".split()
+    + [str(run_duration)]
+    + "disable-triggers wait 2 drain-dataflow wait 2 stop-trigger-sources stop ".split()
+    + " scrap terminate".split()
+)
+
+# 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):
+    local_expected_event_count = expected_event_count
+    local_event_count_tolerance = expected_event_count_tolerance
+    low_number_of_files = expected_number_of_data_files
+    high_number_of_files = expected_number_of_data_files
+    fragment_check_list = [triggercandidate_frag_params, hsi_frag_params]
+    local_expected_event_count += (
+        250
+        * number_of_data_producers
+        * number_of_readout_apps
+        * run_duration
+        / (100 * number_of_dataflow_apps)
+    )
+    local_event_count_tolerance += (
+        10
+        * number_of_data_producers
+        * number_of_readout_apps
+        * run_duration
+        / (100 * number_of_dataflow_apps)
+    )
+    fragment_check_list.append(wibeth_frag_params)
+    fragment_check_list.append(triggerprimitive_frag_params)
+    fragment_check_list.append(triggeractivity_frag_params)
+
+    # Run some tests on the output data file
+    assert (
+        len(run_dunerc.data_files) == high_number_of_files
+        or len(run_dunerc.data_files) == low_number_of_files
+    )
+
+    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, local_expected_event_count, local_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