python.Checks.SimpleCheck Class Reference

Inheritance diagram for python.Checks.SimpleCheck:

Collaboration diagram for python.Checks.SimpleCheck:

Public Member Functions
	__init__ (self, TestSetup setup, str name, str quantity, str unit, int field, float threshold)
bool	run (self, WorkflowTest test)
Optional[Path]	reference_file (self, "WorkflowTest" test, str file_name)

Public Attributes
	name = name
	quantity = quantity
	unit = unit
	field = field
	threshold = threshold
	setup = setup
	logger = setup.logger

Detailed Description

Run A Very Simple Check.

Definition at line 454 of file Checks.py.

Constructor & Destructor Documentation

__init__()

python.Checks.SimpleCheck.__init__	(		self,
		TestSetup	setup,
		str	name,
		str	quantity,
		str	unit,
		int	field,
		float	threshold )

Definition at line 457 of file Checks.py.

    def __init__(self, setup: TestSetup, name: str, quantity: str, unit: str, field: int, threshold: float):
        super().__init__(setup)
        self.name = name
        self.quantity = quantity
        self.unit = unit
        self.field = field
        self.threshold = threshold
 

Member Function Documentation

reference_file()

Optional[Path] python.Test.WorkflowCheck.reference_file ( self, "WorkflowTest" test, str file_name )

inherited

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

    def reference_file(self, test: "WorkflowTest", file_name: str) -> Optional[Path]:
        reference_path: Path = test.reference_path
        reference_file = reference_path / file_name
 
        # Read references from CVMFS
        if self.setup.validation_only:
            # Resolve the subfolder first. Results are stored like: main_folder/branch/test/version/.
            reference_revision = references_map[f"{test.ID}"]
            cvmfs_path = Path(references_CVMFS_path)
            rel_path = Path(self.setup.release_ID) / test.ID / reference_revision
            reference_path = cvmfs_path / rel_path
            reference_file = reference_path / file_name
 
            if not reference_path.exists():
                self.logger.error(f"CVMFS reference location {reference_path} does not exist!")
                return None
 
            if references_override_url is not None:
                import requests
 
                url = references_override_url
                if not url.endswith("/"): url += "/"
                url += str(rel_path / file_name)
                self.logger.info("Checking for reference override at %s", url)
                if requests.head(url).ok: # file exists at url
                    reference_file = Path.cwd() / f"reference_{file_name}"
                    self.logger.info("Downloading reference from %s to %s", url, reference_file)
                    r = requests.get(url, stream=True)
                    with reference_file.open('wb') as f:
                        for chunk in r.iter_content(chunk_size=1024):
                            if chunk: # filter out keep-alive new chunks
                                f.write(chunk)
                else:
                    self.logger.info("No reference override found")
 
        return reference_file
 
 

run()

bool python.Checks.SimpleCheck.run	(		self,
		WorkflowTest	test )

Definition at line 465 of file Checks.py.

    def run(self, test: WorkflowTest) -> bool:
        self.logger.info("-----------------------------------------------------")
        self.logger.info(f"Running {test.ID} {self.name} Check")
 
        result = True
        for step in test.steps:
            log_name = f"log.{step}"
            reference_log = test.reference_path / log_name
            validation_log = test.validation_path / log_name
 
            reference_value = 0
            with reference_log.open() as file:
                found = False
                for line in file:
                    if self.quantity in line:
                        reference_value = float(line.split()[self.field])
                        found = True
                        break
                if not found:
                    self.logger.error(f"No data available in {reference_log}. Job failed.")
                    return False
 
            validation_value = 0
            with validation_log.open() as file:
                found = False
                for line in file:
                    if self.quantity in line:
                        validation_value = float(line.split()[self.field])
                        found = True
                        break
                if not found:
                    self.logger.error(f"No data available in {validation_log}. Job failed.")
                    return False
 
            if reference_value != 0:
                factor = validation_value / reference_value
 
                # Error if the factor increases (very bad things)
                # Warning if the factor decreases (should be an understood feature)
                if factor > 1. + self.threshold:
                    self.logger.error(f"{self.quantity} in the {step} step with(out) your change is {validation_value} ({reference_value}) {self.unit}")
                    self.logger.error(f"Your change changes {self.quantity} by a factor {factor}")
                    self.logger.error("Is this an expected outcome of your change(s)?")
                    result = False
                    self.logger.error(f"{step}: {self.name}")
                    self.logger.error(f"ref  {reference_value} {self.unit}")
                    self.logger.error(f"val {validation_value} {self.unit}")
                if factor < 1. - self.threshold:
                    self.logger.warning(f"{self.quantity} in the {step} step with(out) your change is {validation_value} ({reference_value}) {self.unit}")
                    self.logger.warning(f"Your change changes {self.quantity} by a factor {factor}")
                    self.logger.warning("Is this an expected outcome of your change(s)?")
                    result = True
                    self.logger.warning(f"{step}: {self.name}")
                    self.logger.warning(f"ref  {reference_value} {self.unit}")
                    self.logger.warning(f"val {validation_value} {self.unit}")
 
        if result:
            self.logger.info("Passed!\n")
        else :
            self.logger.error("Failed!\n")
 
        return result
 
 

Member Data Documentation

field

python.Checks.SimpleCheck.field = field

Definition at line 462 of file Checks.py.

logger

python.Test.WorkflowCheck.logger = setup.logger

inherited

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

name

python.Checks.SimpleCheck.name = name

Definition at line 459 of file Checks.py.

quantity

python.Checks.SimpleCheck.quantity = quantity

Definition at line 460 of file Checks.py.

setup

python.Test.WorkflowCheck.setup = setup

inherited

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

threshold

python.Checks.SimpleCheck.threshold = threshold

Definition at line 463 of file Checks.py.

unit

python.Checks.SimpleCheck.unit = unit

Definition at line 461 of file Checks.py.

---

The documentation for this class was generated from the following file:

• Checks.py