Inheritance diagram for python.Checks.FailedOrPassedCheck:

Collaboration diagram for python.Checks.FailedOrPassedCheck:

Public Member Functions
bool	run (self, WorkflowTest test)

Optional[Path]	reference_file (self, "WorkflowTest" test, str file_name)

Public Attributes
	setup

	logger

Detailed Description

Was the q test successful? To check simply count the number of lines containing the string "successful run".

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

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

Definition at line 14 of file Checks.py.

     def run(self, test: WorkflowTest) -> bool:
         result = True
         printed_trf = False
         for step in test.steps:
             self.logger.info("-----------------------------------------------------")
             log = test.validation_path / f"log.{step}"
             counter = 0
             warnings = []
             errors = []
             if log.exists():
                 with log.open() as file:
                     for line in file:
                         if ("ERROR" in line and "| ERROR |" not in line) or ("FATAL" in line and "| FATAL |" not in line):
                             errors.append(line[9:].strip())
                         elif "WARNING" in line and "| WARNING |" not in line:
                             warnings.append(line[9:].strip())
                         elif '"successful run"' in line:
                             counter += 1
                         elif step == "DQHistogramMerge" and "Writing file: myHIST.root" in line:  # DQ merge hack
                             counter += 1
  
             if warnings:
                 self.logger.info(f"{step} validation test step WARNINGS")
                 warnings = list(dict.fromkeys(warnings))
                 for w in warnings:
                     self.logger.info(f"  {w}")
                 self.logger.info("-----------------------------------------------------")
  
             if errors:
                 self.logger.info(f"{step} validation test step ERRORS")
                 errors = list(dict.fromkeys(errors))
                 for e in errors:
                     self.logger.info(f"  {e}")
                 self.logger.info("-----------------------------------------------------")
  
             if counter and not errors:
                 self.logger.info(f"{step} validation test step successful")
  
                 if step == "DQHistogramMerge":
                     self.logger.info(f"Full {step} step log:")
                     with log.open() as file:
                         for line in file:
                             self.logger.print(f"  {line.strip()}")
                     self.logger.info("-----------------------------------------------------")
             else:
                 result = False
                 if log.exists():
                     printed_trf = True  # if one step fails, next steps are expected so don't be too verbose
                     self.logger.error(f"{step} validation test step failed")
                     self.logger.error(f"Full {step} step log:")
                     with log.open() as file:
                         for line in file:
                             self.logger.print(f"  {line.strip()}")
                     self.logger.info("-----------------------------------------------------")
                 else:
                     self.logger.error(f"{step} validation test step did not run")
                     if not printed_trf:
                         printed_trf = True
                         self.logger.error("Full transform log:")
                         with (test.validation_path / f"{test.ID}.log").open() as file:
                             for line in file:
                                 self.logger.print(f"  {line.strip()}")
                         self.logger.info("-----------------------------------------------------")
  
             if self.setup.validation_only:
                 continue  # Skip checking reference test because in this mode the clean tests have not been run
  
             log = test.reference_path / f"log.{step}"
             counter = 0
             with log.open() as file:
                 for line in file:
                     if '"successful run"' in line:
                         counter += 1
                     elif (step == "DQHistogramMerge"  # DQ merge hack
                           and "Writing file: myHIST.root" in line):
                         counter += 1
  
             if counter:
                 self.logger.info(f"{step} reference test step successful")
             else:
                 self.logger.error(f"{step} reference test step failed")
                 result = False
  
         if result:
             self.logger.info(f"All {test.ID} athena steps completed successfully\n")
         else :
             self.logger.error(f"One or more {test.ID} Athena steps failed. Please investigate the cause.\n")
  
         return result
  
  