Inheritance diagram for python.StandardTests.QTest:

Collaboration diagram for python.StandardTests.QTest:

Public Member Functions
None	__init__ (self, str ID, WorkflowRun run, WorkflowType type, List[str] steps, TestSetup setup, str extra_args="")

None	run_reference (self)

None	run_validation (self)

bool	run_checks (self, List[WorkflowCheck] performance_checks)

Public Attributes
	command

	output_checks

	digest_checks

	ID

	tag

	steps

	skip_performance_checks

	run

	type

	setup

	logger

Detailed Description

General workflow q-test.

Definition at line 15 of file StandardTests.py.

Constructor & Destructor Documentation

◆ init()

None python.StandardTests.QTest.__init__	(		self,
		str	ID,
		WorkflowRun	run,
		WorkflowType	type,
		List[str]	steps,
		TestSetup	setup,
		str	extra_args = `""`
	)

Definition at line 18 of file StandardTests.py.

     def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: List[str], setup: TestSetup, extra_args: str = "") -> None:
         if "maxEvents" not in extra_args:
             if type == WorkflowType.MCPileUpReco or run == WorkflowRun.Run4:
                 extra_args += " --maxEvents 5"
             else:
                 extra_args += " --maxEvents 20"
  
         if type == WorkflowType.MCPileUpReco:
             if "inputHITSFile" not in extra_args:
                 extra_args += f" --inputHITSFile {input_HITS[run]}"
             if "inputRDO_BKGFile" not in extra_args:
                 extra_args += " --inputRDO_BKGFile ../run_d*/myRDO.pool.root"
  
         threads = 1
         threads_argument = '--multithreaded'
         if setup.custom_threads is not None:
             threads = setup.custom_threads
         if threads <= 0:
             threads_argument = ''
  
         self.command = \
             (f"ATHENA_CORE_NUMBER={threads} Reco_tf.py {threads_argument} --AMIConfig {ID}"
              f" --imf False {extra_args}")
  
         self.output_checks = []
         # TODO: disable RDO comparison for now
         # if type == WorkflowType.MCReco:
         #     self.output_checks.append(FrozenTier0PolicyCheck(setup, "RDO", 10))
         self.output_checks.append(FrozenTier0PolicyCheck(setup, "AOD", 60))
         self.output_checks.append(FrozenTier0PolicyCheck(setup, "ESD", 20))
         self.output_checks.append(MetadataCheck(setup, "AOD"))
  
         self.digest_checks = []
         if not setup.disable_output_checks:
             self.digest_checks.append(AODContentCheck(setup))
         self.digest_checks.append(AODDigestCheck(setup))
  
         super().__init__(ID, run, type, steps, setup)
  
  

Member Function Documentation

◆ run_checks()

bool python.Test.WorkflowTest.run_checks	(		self,
		List[WorkflowCheck]	performance_checks
	)

inherited

Definition at line 197 of file Tools/WorkflowTestRunner/python/Test.py.

     def run_checks(self, performance_checks: List[WorkflowCheck]) -> bool:
         self.logger.info("-----------------------------------------------------")
         self.logger.info(f"----------- Post-processing of {self.ID} Test -----------")
         result = True
  
         # digest checks
         for check in self.digest_checks:
             result = check.run(self) and result
  
         # output checks
         if not self.setup.disable_output_checks:
             for check in self.output_checks:
                 result = check.run(self) and result
  
         if self.setup.validation_only or self.skip_performance_checks:
             return result  # Performance checks against static references not possible
  
         # performance checks
         for check in performance_checks:
             result = check.run(self) and result
  
         return result

◆ run_reference()

None python.Test.WorkflowTest.run_reference ( self )

inherited

Definition at line 160 of file Tools/WorkflowTestRunner/python/Test.py.

     def run_reference(self) -> None:
         self.logger.info(f"Running reference in rel {self.setup.release_reference}")
         self.logger.info(f"\"{self.command}\"")
  
         self.reference_path.mkdir(parents=True, exist_ok=True)
  
         cmd = (f"cd {self.reference_path};"
                f"source $AtlasSetup/scripts/asetup.sh {self.setup.release_reference} >& /dev/null;")
         cmd += f"TRF_NOECHO=1 {self.command} > {self.ID}.log 2>&1"
  
         subprocess.call(cmd, shell=True)
  
         self.logger.info(f"Finished clean in rel {self.setup.release_reference}")
         self.logger.info(f"\"{self.command}\"")
  

◆ run_validation()

None python.Test.WorkflowTest.run_validation ( self )

inherited

Definition at line 175 of file Tools/WorkflowTestRunner/python/Test.py.

     def run_validation(self) -> None:
         self.logger.info(f"Running validation in rel {self.setup.release_validation}")
         self.logger.info(f"\"{self.command}\"")
  
         self.validation_path.mkdir(parents=True, exist_ok=True)
  
         cmd = f"cd {self.validation_path};"
         if self.setup.disable_release_setup or not self.setup.release_validation:
             pass
         elif "WorkDir_DIR" in environ:
             cmake_build_dir = environ["WorkDir_DIR"]
             cmd += (f"source $AtlasSetup/scripts/asetup.sh {self.setup.release_validation} >& /dev/null;"
                     f"source {cmake_build_dir}/setup.sh;")
         else:
             cmd += f"source $AtlasSetup/scripts/asetup.sh {self.setup.release_validation} >& /dev/null;"
         cmd += f"TRF_NOECHO=1 {self.command} > {self.ID}.log 2>&1"
  
         subprocess.call(cmd, shell=True)
  
         self.logger.info(f"Finished validation in rel {self.setup.release_validation}")
         self.logger.info(f"\"{self.command}\"")
  