• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2023 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5from __future__ import annotations
6
7import abc
8from typing import TYPE_CHECKING, Iterable, Optional
9
10from crossbench import exception
11from crossbench.probes.results import ProbeResult, ProbeResultDict
12
13if TYPE_CHECKING:
14  from crossbench.path import LocalPath
15  from crossbench.probes.probe import Probe
16  from crossbench.runner.run import Run
17  from crossbench.runner.runner import Runner
18  from crossbench.types import JsonDict
19
20
21class RunGroup(abc.ABC):
22
23  def __init__(self, throw: bool = False) -> None:
24    self._exceptions = exception.Annotator(throw)
25    self._path: Optional[LocalPath] = None
26    self._merged_probe_results: Optional[ProbeResultDict] = None
27
28  def _set_path(self, path: LocalPath) -> None:
29    assert self._path is None
30    self._path = path
31    self._merged_probe_results = ProbeResultDict(path)
32
33  @property
34  def results(self) -> ProbeResultDict:
35    assert self._merged_probe_results is not None
36    return self._merged_probe_results
37
38  @property
39  def path(self) -> LocalPath:
40    assert self._path
41    return self._path
42
43  @property
44  def throw(self) -> bool:
45    return self._exceptions.throw
46
47  @property
48  def exceptions(self) -> exception.Annotator:
49    return self._exceptions
50
51  @property
52  def is_success(self) -> bool:
53    return self._exceptions.is_success
54
55  @property
56  @abc.abstractmethod
57  def info_stack(self) -> exception.TInfoStack:
58    pass
59
60  @property
61  @abc.abstractmethod
62  def runs(self) -> Iterable[Run]:
63    pass
64
65  @property
66  def failed_runs(self) -> Iterable[Run]:
67    for run in self.runs:
68      if not run.is_success:
69        yield run
70
71  @property
72  def info(self) -> JsonDict:
73    return {
74        "runs": len(tuple(self.runs)),
75        "failed runs": len(tuple(self.failed_runs))
76    }
77
78  def get_local_probe_result_path(self,
79                                  probe: Probe,
80                                  exists_ok: bool = False) -> LocalPath:
81    new_file = self.path / probe.result_path_name
82    if not exists_ok:
83      assert not new_file.exists(), (
84          f"Merged file {new_file} for {self.__class__} exists already.")
85    return new_file
86
87  def get_local_probe_result_dir(self,
88                                 probe: Probe,
89                                 exists_ok: bool = True) -> LocalPath:
90    path = self.get_local_probe_result_path(probe, exists_ok)
91    path.mkdir(parents=True, exist_ok=exists_ok)
92    return path
93
94  def merge(self, probes: Iterable[Probe]) -> None:
95    assert self._merged_probe_results is not None
96    with self._exceptions.info(*self.info_stack):
97      for probe in reversed(tuple(probes)):
98        with self._exceptions.capture(f"Probe {probe.name} merge results"):
99          results = self._merge_probe_results(probe)
100          if results is None:
101            continue
102          self._merged_probe_results[probe] = results
103
104  @abc.abstractmethod
105  def _merge_probe_results(self, probe: Probe) -> ProbeResult:
106    pass
107