python.Checks.FPECheck Class Reference

Inheritance diagram for python.Checks.FPECheck:

Collaboration diagram for python.Checks.FPECheck:

Public Member Functions
def	run (self, WorkflowTest test)
Optional[Path]	reference_file (self, "WorkflowTest" test, str file_name)

Public Attributes
	setup
	logger

Static Public Attributes
	ignoreTestRuns
	ignoreTestTypes
	ignoreTestIDs

Detailed Description

Run FPE check.

Definition at line 580 of file Checks.py.

Member Function Documentation

reference_file()

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

inherited

Definition at line 91 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()

def python.Checks.FPECheck.run	(		self,
		WorkflowTest	test
	)

Definition at line 588 of file Checks.py.

    def run(self, test: WorkflowTest):
        self.logger.info("-----------------------------------------------------")
        self.logger.info(f"Running {test.ID} FPE Check")
 
        result = True
        for step in test.steps:
            log = test.validation_path / f"log.{step}"
            fpes = {}
            stack_traces = {}
            with log.open() as file:
                last_stack_trace = None
                for line in file:
                    if "WARNING FPE" in line:
                        last_stack_trace = None
                        fpe = None
                        for part in reversed(line.split()):
                            if "[" in part:
                                fpe = part.strip().replace("[", "").replace("]", "")
                                break
                        if fpe:
                            if fpe in fpes:
                                fpes[fpe] += 1
                            else:
                                fpes[fpe] = 1
                                last_stack_trace = []
                                stack_traces[fpe] = last_stack_trace
                    elif "FPE stacktrace" in line and last_stack_trace is not None:
                        line = next(file)
                        last_stack_trace.append(line.strip()[9:])
 
            if fpes.keys():
                msgLvl = logging.WARNING if test.run in self.ignoreTestRuns or test.type in self.ignoreTestTypes or test.ID in self.ignoreTestIDs else logging.ERROR
                result = False
                self.logger.log(msgLvl, f" {step} validation test step FPEs")
                for fpe, count in sorted(fpes.items(), key=lambda item: item[1]):
                    self.logger.log(msgLvl, f"{count:>5}  {fpe}")
                for fpe in fpes.keys():
                    self.logger.log(msgLvl, "-----------------------------------------------------")
                    self.logger.log(msgLvl, f" first stack trace for algorithm {fpe}:")
                    for line in stack_traces[fpe]:
                        self.logger.log(msgLvl, line)
                    self.logger.log(msgLvl, "-----------------------------------------------------")
 
        if result:
            self.logger.info("Passed!\n")
        elif test.run in self.ignoreTestRuns or test.type in self.ignoreTestTypes or test.ID in self.ignoreTestIDs:
            self.logger.warning("Failed!")
            self.logger.warning("Check disabled due to irreproducibilities!\n")
            result = True
        else:
            self.logger.error("Failed!\n")
 
        return result

Member Data Documentation

ignoreTestIDs

python.Checks.FPECheck.ignoreTestIDs

static

Definition at line 586 of file Checks.py.

ignoreTestRuns

python.Checks.FPECheck.ignoreTestRuns

static

Definition at line 584 of file Checks.py.

ignoreTestTypes

python.Checks.FPECheck.ignoreTestTypes

static

Definition at line 585 of file Checks.py.

logger

python.Test.WorkflowCheck.logger

inherited

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

setup

python.Test.WorkflowCheck.setup

inherited

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

---

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

• Checks.py