3 # Copyright (c) 2020-2021 Virtuozzo International GmbH
34 def silent_unlink(path: Path) -> None:
41 def file_diff(file1: str, file2: str) -> List[str]:
42     with open(file1, encoding="utf-8") as f1, \
43          open(file2, encoding="utf-8") as f2:
56     """ Cache for elapsed time for tests, to show it during new test run
59     use it inside with-block or use save() after update().
61     def __init__(self, cache_file: str, env: TestEnv) -> None:  argument
62         self.env = env
63         self.cache_file = cache_file
64         self.cache: Dict[str, Dict[str, Dict[str, float]]]
67             with open(cache_file, encoding="utf-8") as f:
68                 self.cache = json.load(f)
70             self.cache = {}
72     def get(self, test: str,  argument
73             default: Optional[float] = None) -> Optional[float]:
74         if test not in self.cache:
77         if self.env.imgproto not in self.cache[test]:
80         return self.cache[test][self.env.imgproto].get(self.env.imgfmt,
83     def update(self, test: str, elapsed: float) -> None:  argument
84         d = self.cache.setdefault(test, {})
85         d.setdefault(self.env.imgproto, {})[self.env.imgfmt] = elapsed
87     def save(self) -> None:  argument
88         with open(self.cache_file, 'w', encoding="utf-8") as f:
89             json.dump(self.cache, f)
91     def __enter__(self) -> 'LastElapsedTime':  argument
92         return self
94     def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:  argument
95         self.save()
99     def __init__(self, status: str, description: str = '',  argument
101                  casenotrun: str = '', interrupted: bool = False) -> None:
102         self.status = status
103         self.description = description
104         self.elapsed = elapsed
105         self.diff = diff
106         self.casenotrun = casenotrun
107         self.interrupted = interrupted
114     def proc_run_test(test: str, test_field_width: int) -> TestResult:
118         return runner.run_test(test, test_field_width, mp=True)
120     def run_tests_pool(self, tests: List[str],  argument
121                        test_field_width: int, jobs: int) -> List[TestResult]:
123         # passing self directly to Pool.starmap() just doesn't work, because
126         TestRunner.shared_self = self
129             results = p.starmap(self.proc_run_test,
136     def __init__(self, env: TestEnv, tap: bool = False,  argument
137                  color: str = 'auto') -> None:
138         self.env = env
139         self.tap = tap
140         self.last_elapsed = LastElapsedTime('.last-elapsed-cache', env)
143         self.color = (color == 'on') or (color == 'auto' and
146         self._stack: contextlib.ExitStack
148     def __enter__(self) -> 'TestRunner':  argument
149         self._stack = contextlib.ExitStack()
150         self._stack.enter_context(self.env)
151         self._stack.enter_context(self.last_elapsed)
152         return self
154     def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:  argument
155         self._stack.close()
157     def test_print_one_line(self, test: str,  argument
164                             end: str = '\n') -> None:
165         """ Print short test info before/after test run """
166         test = os.path.basename(test)
171         if self.tap:
173                 print(f'ok {self.env.imgfmt} {test}')
175                 print(f'not ok {self.env.imgfmt} {test}')
177                 print(f'ok {self.env.imgfmt} {test} # SKIP')
194         if self.color:
209         print(f'{test:{test_field_width}} {col}{status:10}{col_end} '
213     def find_reference(self, test: str) -> str:  argument
214         if self.env.cachemode == 'none':
215             ref = f'{test}.out.nocache'
219         ref = f'{test}.out.{self.env.imgfmt}'
223         ref = f'{test}.{self.env.qemu_default_machine}.out'
227         return f'{test}.out'
229     def do_run_test(self, test: str) -> TestResult:  argument
231         Run one test
233         :param test: test file path
236         change ``self`` object in any way!
239         f_test = Path(test)
240         f_reference = Path(self.find_reference(test))
244                               description=f'No such test file: {f_test}')
255         env = self.env.prepare_subprocess(args)
257         # Split test directories, so that tests running in parallel don't
262                 f"{self.env.imgfmt}-{self.env.imgproto}-{f_test.name}")
274         with f_bad.open('w', encoding="utf-8") as f:
288         elapsed = round(time.time() - t0, 1)
298                 description=f_notrun.read_text(encoding='utf-8').strip())
302             casenotrun = f_casenotrun.read_text(encoding='utf-8')
319     def run_test(self, test: str,  argument
321                  mp: bool = False) -> TestResult:
323         Run one test and print short status
325         :param test: test file path
331         change ``self`` object in any way!
334         last_el = self.last_elapsed.get(test)
337         if not self.tap:
338             self.test_print_one_line(test=test,
345             testname = os.path.basename(test)
346             print(f'# running {self.env.imgfmt} {testname}')
348         res = self.do_run_test(test)
351         self.test_print_one_line(test=test,
359             if self.tap:
367     def run_tests(self, tests: List[str], jobs: int = 1) -> bool:  argument
373         if self.tap:
375             self.env.print_env('# ')
378             self.env.print_env()
383             results = self.run_tests_pool(tests, test_field_width, jobs)
391                 res = self.run_test(t, test_field_width)
404                     if self.tap:
412                 self.last_elapsed.update(t, res.elapsed)
418         if not self.tap: