source/sphenixprod/test_check_downstream.py

0001 from collections import namedtuple
0002
0003 from check_downstream import build_eligible_units, find_flagged_units
0004
0005
0006 Row = namedtuple("Row", "dsttype runnumber segment events filename")
0007
0008
0009 REQUIRED = ["DST_TRIGGERED_EVENT_seb18", "DST_TRIGGERED_EVENT_seb20"]
0010
0011
0012 def test_matching_input_and_output_is_not_flagged():
0013     inputs = [
0014         Row("DST_TRIGGERED_EVENT_seb18", 82703, 0, 1000, "in18.root"),
0015         Row("DST_TRIGGERED_EVENT_seb20", 82703, 0, 1000, "in20.root"),
0016     ]
0017     outputs = [Row("DST_CALOFITTING", 82703, 0, 1000, "out.root")]
0018
0019     eligible = build_eligible_units(inputs, REQUIRED)
0020     assert find_flagged_units(eligible, outputs, ratio_cut=0.9) == []
0021
0022
0023 def test_missing_input_stream_is_not_eligible():
0024     inputs = [Row("DST_TRIGGERED_EVENT_seb18", 82703, 0, 1000, "in18.root")]
0025
0026     eligible = build_eligible_units(inputs, REQUIRED)
0027     assert eligible == {}
0028
0029
0030 def test_missing_output_is_flagged():
0031     inputs = [
0032         Row("DST_TRIGGERED_EVENT_seb18", 82703, 0, 1000, "in18.root"),
0033         Row("DST_TRIGGERED_EVENT_seb20", 82703, 0, 1000, "in20.root"),
0034     ]
0035
0036     eligible = build_eligible_units(inputs, REQUIRED)
0037     flagged = find_flagged_units(eligible, [], ratio_cut=0.9)
0038
0039     assert len(flagged) == 1
0040     assert flagged[0].reasons == ("missing_output",)
0041     assert flagged[0].report_line() == "00082703 00000 missing_output 1000 -1"
0042
0043
0044 def test_low_output_events_is_flagged():
0045     inputs = [
0046         Row("DST_TRIGGERED_EVENT_seb18", 82703, 0, 1000, "in18.root"),
0047         Row("DST_TRIGGERED_EVENT_seb20", 82703, 0, 1000, "in20.root"),
0048     ]
0049     outputs = [Row("DST_CALOFITTING", 82703, 0, 800, "out.root")]
0050
0051     eligible = build_eligible_units(inputs, REQUIRED)
0052     flagged = find_flagged_units(eligible, outputs, ratio_cut=0.9)
0053
0054     assert len(flagged) == 1
0055     assert flagged[0].reasons == ("low_output_events",)
0056
0057
0058 def test_input_mismatch_is_flagged():
0059     inputs = [
0060         Row("DST_TRIGGERED_EVENT_seb18", 82703, 0, 1000, "in18.root"),
0061         Row("DST_TRIGGERED_EVENT_seb20", 82703, 0, 999, "in20.root"),
0062     ]
0063     outputs = [Row("DST_CALOFITTING", 82703, 0, 1000, "out.root")]
0064
0065     eligible = build_eligible_units(inputs, REQUIRED)
0066     flagged = find_flagged_units(eligible, outputs, ratio_cut=0.9)
0067
0068     assert len(flagged) == 1
0069     assert flagged[0].reasons == ("input_mismatch",)
0070     assert flagged[0].input_events == -1
0071
0072
0073 def test_cut_segment_skips_non_divisible_segments():
0074     inputs = [
0075         Row("DST_TRIGGERED_EVENT_seb18", 82703, 1, 1000, "in18.root"),
0076         Row("DST_TRIGGERED_EVENT_seb20", 82703, 1, 1000, "in20.root"),
0077         Row("DST_TRIGGERED_EVENT_seb18", 82703, 2, 1000, "in18.root"),
0078         Row("DST_TRIGGERED_EVENT_seb20", 82703, 2, 1000, "in20.root"),
0079     ]
0080
0081     eligible = build_eligible_units(inputs, REQUIRED, cut_segment=2)
0082
0083     assert sorted(eligible) == [(82703, 2)]