• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2024 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
8import datetime as dt
9from typing import TYPE_CHECKING, Any, Dict, Type, TypeVar
10
11from crossbench import exception
12from crossbench.action_runner.action.action_type import ActionType
13from crossbench.config import ConfigObject, ConfigParser
14from crossbench.parse import DurationParser, NumberParser, ObjectParser
15
16if TYPE_CHECKING:
17  from crossbench.action_runner.base import ActionRunner
18  from crossbench.runner.run import Run
19  from crossbench.types import JsonDict
20
21
22class ActionTypeConfigParser(ConfigParser):
23  """Custom ConfigParser for ActionType that works on
24  Action Configs. This way we can pop the 'value' or 'type' key from the
25  config dict."""
26
27  def __init__(self):
28    super().__init__("ActionType parser", ActionType)
29    self.add_argument(
30        "action",
31        aliases=("type",),
32        type=ObjectParser.non_empty_str,
33        required=True)
34
35  def new_instance_from_kwargs(self, kwargs: Dict[str, Any]) -> ActionType:
36    return ActionType(kwargs["action"])
37
38
39_ACTION_TYPE_CONFIG_PARSER = ActionTypeConfigParser()
40
41ACTION_TIMEOUT = dt.timedelta(seconds=20)
42
43ActionT = TypeVar("ActionT", bound="Action")
44
45# Lazily initialized Action class lookup.
46ACTIONS: Dict[ActionType, Type[Action]] = {}
47
48
49class Action(ConfigObject, metaclass=abc.ABCMeta):
50  TYPE: ActionType = ActionType.GET
51
52  @classmethod
53  def parse_str(cls, value: str) -> Action:
54    return ACTIONS[ActionType.GET].parse_str(value)
55
56  @classmethod
57  def parse_dict(cls: Type[ActionT], config: Dict[str, Any]) -> ActionT:
58    action_type: ActionType = _ACTION_TYPE_CONFIG_PARSER.parse(config)
59    action_cls: Type[ActionT] = ACTIONS[action_type]
60    with exception.annotate_argparsing(
61        f"Parsing Action details  ...{{ action: \"{action_type}\", ...}}:"):
62      action = action_cls.config_parser().parse(config)
63    assert isinstance(action, cls), f"Expected {cls} but got {type(action)}"
64    return action
65
66  @classmethod
67  def config_parser(cls: Type[ActionT]) -> ConfigParser[ActionT]:
68    parser = ConfigParser(f"{cls.__name__} parser", cls)
69    parser.add_argument(
70        "index", type=NumberParser.positive_zero_int, required=False, default=0)
71    parser.add_argument(
72        "timeout",
73        type=DurationParser.positive_duration,
74        default=ACTION_TIMEOUT)
75    return parser
76
77  def __init__(self, timeout: dt.timedelta = ACTION_TIMEOUT, index: int = 0):
78    self._timeout: dt.timedelta = timeout
79    self._index = index
80    self.validate()
81
82  @property
83  def index(self) -> int:
84    return self._index
85
86  @property
87  def duration(self) -> dt.timedelta:
88    return dt.timedelta(milliseconds=10)
89
90  @property
91  def timeout(self) -> dt.timedelta:
92    return self._timeout
93
94  @property
95  def has_timeout(self) -> bool:
96    return self._timeout != dt.timedelta.max
97
98  @abc.abstractmethod
99  def run_with(self, run: Run, action_runner: ActionRunner) -> None:
100    pass
101
102  def validate(self) -> None:
103    if self._timeout.total_seconds() < 0:
104      raise ValueError(
105          f"{self}.timeout should be positive, but got {self.timeout}")
106
107  def to_json(self) -> JsonDict:
108    return {"type": str(self.TYPE), "timeout": self.timeout.total_seconds()}
109
110  def __str__(self) -> str:
111    return type(self).__name__
112
113  def __eq__(self, other: object) -> bool:
114    if isinstance(other, Action):
115      return self.to_json() == other.to_json()
116    return False
117