TrigValTools: Add python-only step to TrigValSteering

Add a `PyStep` class that allows to implement Python-only "steps" via
a user-defined function.

Trigger/TrigValidation/TrigValTools/python/TrigValSteering/CheckSteps.py

@@ -13,6 +13,7 @@ import json
 import glob
 
 from TrigValTools.TrigValSteering.Step import Step, get_step_from_list
+from TrigValTools.TrigValSteering.ExecStep import ExecStep
 from TrigValTools.TrigValSteering.Common import art_input_eos, art_input_cvmfs, running_in_CI
 
 class RefComparisonStep(Step):
@@ -629,16 +630,18 @@ class MessageCountStep(Step):
             self.print_on_fail = self.required
         if self.print_on_fail:
             self.args += ' --saveAll'
+
+        max_events = test.exec_steps[0].max_events if isinstance(test.exec_steps[0], ExecStep) else 0
         if 'WARNING' not in self.thresholds:
             self.thresholds['WARNING'] = 0
         if 'INFO' not in self.thresholds:
-            self.thresholds['INFO'] = test.exec_steps[0].max_events
+            self.thresholds['INFO'] = max_events
         if 'DEBUG' not in self.thresholds:
             self.thresholds['DEBUG'] = 0
         if 'VERBOSE' not in self.thresholds:
             self.thresholds['VERBOSE'] = 0
         if 'other' not in self.thresholds:
-            self.thresholds['other'] = test.exec_steps[0].max_events
+            self.thresholds['other'] = max_events
         super(MessageCountStep, self).configure(test)
 
     def run(self, dry_run=False):

Trigger/TrigValidation/TrigValTools/python/TrigValSteering/PyStep.py

+#
+# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+#
+
+"""
+Step implemented as python function
+"""
+
+from TrigValTools.TrigValSteering.Step import Step
+import contextlib
+import sys
+
+class PyStep(Step):
+   """Step calling a python function"""
+
+   def __init__(self, func, name=None):
+      super(PyStep, self).__init__(name)
+      self.func = func
+      self.output_stream = Step.OutputStream.STDOUT_ONLY
+      if self.name is None:
+         self.name = self.func.__name__
+
+   def run(self, dry_run=False):
+
+      self.log.info('Running %s step', self.name)
+
+      dest = sys.stdout
+      if self.output_stream == self.OutputStream.NO_PRINT:
+         dest = None
+      elif self.output_stream in [self.OutputStream.FILE_ONLY, self.OutputStream.FILE_AND_STDOUT]:
+         dest = open(self.get_log_file_name(), 'w')
+
+      if dry_run:
+         self.result = 0
+      else:
+         try:
+            with contextlib.redirect_stdout(dest), contextlib.redirect_stderr(dest):
+               self.result = self.func()
+
+            # Poor man's implementation of 'tee'
+            if self.output_stream == self.OutputStream.FILE_AND_STDOUT:
+               dest.close()
+               print(open(dest.name).read())
+
+            # In case function does not return a value, assume success
+            if self.result is None:
+               self.result = 0
+         except Exception as e:
+            self.log.error('Exception calling %s: %s', self.func.__name__, e)
+            self.result = 1
+
+      return self.result, f'# (internal) {self.func.__name__}'

Trigger/TrigValidation/TrigValTools/test/test_unit_trigvalsteering.py

@@ -4,7 +4,7 @@
 # This is not an ART test. This is a unit test of the framework used for
 # steering Trigger ART tests.
 
-from TrigValTools.TrigValSteering import Test, ExecStep, CheckSteps, Common
+from TrigValTools.TrigValSteering import Test, ExecStep, CheckSteps, PyStep, Common
 import logging
 
 import os
@@ -26,19 +26,33 @@ for test_type in ['athena','athenaHLT','Reco_tf','Trig_reco_tf']:
     ex.args = '--help'
     test.exec_steps.append(ex)
 
+def hello():
+    print('### Hello World')
+
+def hello_throw():
+    raise RuntimeError('Hello World')
+
+test.exec_steps.append(PyStep.PyStep(hello, name='hello_stdout'))
+test.exec_steps.append(PyStep.PyStep(hello_throw))
+s = PyStep.PyStep(hello, name='hello_file')
+s.output_stream = s.OutputStream.FILE_AND_STDOUT
+test.exec_steps.append(s)
+
+
 test.art_type = 'build'
 test.check_steps = CheckSteps.default_check_steps(test)
 
-regtest_ref_text = [
-    '### athena.log ###',
-    '### athena.Test_athena.log ###',
-    '### athenaHLT.Test_athenaHLT.log ###',
-    '### Reco_tf.Test_Reco_tf.log ###',
-    '### Trig_reco_tf.Test_Trig_reco_tf.log ###',
-]
+regtest_ref_text = """\
+### athena.log ###
+### athena.Test_athena.log ###
+### athenaHLT.Test_athenaHLT.log ###
+### Reco_tf.Test_Reco_tf.log ###
+### Trig_reco_tf.Test_Trig_reco_tf.log ###
+### hello_file.log ###
+### Hello World"""
 
 with open('regtest.ref','w') as f:
-    f.write('\n'.join(regtest_ref_text))
+    f.write(regtest_ref_text)
 
 refcomp = CheckSteps.RegTestStep('RefComp')
 refcomp.input_base_name = 'athena.merged'