diff --git a/integtest/no_duplicate_ta_test.py b/integtest/no_duplicate_ta_test.py new file mode 100644 index 00000000..9d03fddb --- /dev/null +++ b/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