ATLAS Offline Software
Loading...
Searching...
No Matches
python.Checks.FailedOrPassedCheck Class Reference
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 = setup
 logger = 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 93 of file Tools/WorkflowTestRunner/python/Test.py.

93 def reference_file(self, test: "WorkflowTest", file_name: str) -> Optional[Path]:
94 reference_path: Path = test.reference_path
95 reference_file = reference_path / file_name
96
97 # Read references from CVMFS
98 if self.setup.validation_only:
99 # Resolve the subfolder first. Results are stored like: main_folder/branch/test/version/.
100 reference_revision = references_map[f"{test.ID}"]
101 cvmfs_path = Path(references_CVMFS_path)
102 rel_path = Path(self.setup.release_ID) / test.ID / reference_revision
103 reference_path = cvmfs_path / rel_path
104 reference_file = reference_path / file_name
105
106 if not reference_path.exists():
107 self.logger.error(f"CVMFS reference location {reference_path} does not exist!")
108 return None
109
110 if references_override_url is not None:
111 import requests
112
113 url = references_override_url
114 if not url.endswith("/"): url += "/"
115 url += str(rel_path / file_name)
116 self.logger.info("Checking for reference override at %s", url)
117 if requests.head(url).ok: # file exists at url
118 reference_file = Path.cwd() / f"reference_{file_name}"
119 self.logger.info("Downloading reference from %s to %s", url, reference_file)
120 r = requests.get(url, stream=True)
121 with reference_file.open('wb') as f:
122 for chunk in r.iter_content(chunk_size=1024):
123 if chunk: # filter out keep-alive new chunks
124 f.write(chunk)
125 else:
126 self.logger.info("No reference override found")
127
128 return reference_file
129
130

◆ run()

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

Definition at line 14 of file Checks.py.

14 def run(self, test: WorkflowTest) -> bool:
15 result = True
16 printed_trf = False
17 for step in test.steps:
18 self.logger.info("-----------------------------------------------------")
19 log = test.validation_path / f"log.{step}"
20 counter = 0
21 warnings = []
22 errors = []
23 if log.exists():
24 with log.open() as file:
25 for line in file:
26 if ("ERROR" in line and "| ERROR |" not in line) or ("FATAL" in line and "| FATAL |" not in line):
27 errors.append(line[9:].strip())
28 elif "WARNING" in line and "| WARNING |" not in line:
29 warnings.append(line[9:].strip())
30 elif '"successful run"' in line:
31 counter += 1
32 elif step == "DQHistogramMerge" and "Writing file: myHIST.root" in line: # DQ merge hack
33 counter += 1
34
35 if warnings:
36 self.logger.info(f"{step} validation test step WARNINGS")
37 warnings = list(dict.fromkeys(warnings))
38 for w in warnings:
39 self.logger.info(f" {w}")
40 self.logger.info("-----------------------------------------------------")
41
42 if errors:
43 self.logger.info(f"{step} validation test step ERRORS")
44 errors = list(dict.fromkeys(errors))
45 for e in errors:
46 self.logger.info(f" {e}")
47 self.logger.info("-----------------------------------------------------")
48
49 if counter and not errors:
50 self.logger.info(f"{step} validation test step successful")
51
52 if step == "DQHistogramMerge":
53 self.logger.info(f"Full {step} step log:")
54 with log.open() as file:
55 for line in file:
56 self.logger.print(f" {line.strip()}")
57 self.logger.info("-----------------------------------------------------")
58 else:
59 result = False
60 if log.exists():
61 printed_trf = True # if one step fails, next steps are expected so don't be too verbose
62 self.logger.error(f"{step} validation test step failed")
63 self.logger.error(f"Full {step} step log:")
64 with log.open() as file:
65 for line in file:
66 self.logger.print(f" {line.strip()}")
67 self.logger.info("-----------------------------------------------------")
68 else:
69 self.logger.error(f"{step} validation test step did not run")
70 if not printed_trf:
71 printed_trf = True
72 self.logger.error("Full transform log:")
73 with (test.validation_path / f"{test.ID}.log").open() as file:
74 for line in file:
75 self.logger.print(f" {line.strip()}")
76 self.logger.info("-----------------------------------------------------")
77
78 if self.setup.validation_only:
79 continue # Skip checking reference test because in this mode the clean tests have not been run
80
81 log = test.reference_path / f"log.{step}"
82 counter = 0
83 with log.open() as file:
84 for line in file:
85 if '"successful run"' in line:
86 counter += 1
87 elif (step == "DQHistogramMerge" # DQ merge hack
88 and "Writing file: myHIST.root" in line):
89 counter += 1
90
91 if counter:
92 self.logger.info(f"{step} reference test step successful")
93 else:
94 self.logger.error(f"{step} reference test step failed")
95 result = False
96
97 if result:
98 self.logger.info(f"All {test.ID} athena steps completed successfully\n")
99 else :
100 self.logger.error(f"One or more {test.ID} Athena steps failed. Please investigate the cause.\n")
101
102 return result
103
104
void print(char *figname, TCanvas *c1)
Definition run.py:1

Member Data Documentation

◆ logger

python.Test.WorkflowCheck.logger = setup.logger
inherited

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

◆ setup

python.Test.WorkflowCheck.setup = setup
inherited

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


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