Inheritance diagram for python.StandardTests.DerivationTest:

Collaboration diagram for python.StandardTests.DerivationTest:

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
	ID = ID
	tag = ID
	steps = steps
	command
list	output_checks = []
list	digest_checks = []
bool	skip_performance_checks = False
	run = run
	type = type
	setup = setup
	logger = setup.logger
Path	validation_path = self.setup.validation_run_path / f"run_{self.ID}"
Path	reference_path = self.setup.reference_run_path / f"run_{self.ID}"

Detailed Description

Derivations test.

Definition at line 212 of file StandardTests.py.

Constructor & Destructor Documentation

◆ init()

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

Definition at line 215 of file StandardTests.py.

    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: List[str], setup: TestSetup, extra_args: str = "") -> None:
        test_def = ID.split("_")
        multithreaded = False
        if test_def[-2] == "MT":
            multithreaded = True
            test_run = test_def.pop()
            test_def.pop()  # remove MT
            test_def.append(test_run)  # add run back to the end
        data_type = test_def[0].lower()
        formats = [format.upper() for format in test_def[1:-1]]
 
        threads = 0
        if setup.custom_threads is not None:
            threads = setup.custom_threads
 
        # If we're running a DAOD to DAOD job, the output stream is called D2AOD
        out_stream = "DAOD"
        if "inputDAOD_PHYSFile" in extra_args:
            out_stream = "D2AOD"
 
        if "maxEvents" not in extra_args:
            base_events = 100
            events = threads * base_events + 1
            extra_args += f" --maxEvents {events}"
 
            if not multithreaded:
                flush = 80
                format_flush = ", ".join([f"\"{out_stream}_{format}\": {flush}" for format in formats])
                extra_args += f" --preExec 'flags.Output.TreeAutoFlush={{{format_flush}}}'"
 
        if "inputAODFile" not in extra_args and "inputDAOD_PHYSFile" not in extra_args:
            extra_args += f" --inputAODFile {input_AOD[run][data_type]}"
 
        # could also use p5503
        if not multithreaded:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} Derivation_tf.py"
                f" --formats {' '.join(formats)}"
                " --multiprocess --multithreadedFileValidation True"
                f" --athenaMPMergeTargetSize '{out_stream}_*:0'"
                " --sharedWriter True"
                f" --output{out_stream}File myOutput.pool.root"
                f" --imf False {extra_args}")
        else:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} Derivation_tf.py"
                f" --formats {' '.join(formats)}"
                " --multithreaded"
                f" --output{out_stream}File myOutput.pool.root"
                f" --imf False {extra_args}")
 
        # skip performance checks for now
        self.skip_performance_checks = True
 
        self.output_checks = []
        for format in formats:
            self.output_checks.append(FrozenTier0PolicyCheck(setup, f"{out_stream}_{format}", 10))
            self.output_checks.append(MetadataCheck(setup, f"{out_stream}_{format}"))
 
        super().__init__("_".join(test_def), run, type, steps, setup)
 
 

Member Function Documentation

◆ run_checks()

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

inherited

Definition at line 200 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 163 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 178 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}\"")
 