• 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 dataclasses
8import datetime as dt
9from typing import Dict, Union
10
11# Arbitrary very large number that doesn't break any browser driver protocol.
12# chromedriver likely uses an uint32 ms internally, 2**30ms == 12days.
13SAFE_MAX_TIMEOUT_TIMEDELTA = dt.timedelta(milliseconds=2**30)
14
15
16@dataclasses.dataclass(frozen=True)
17class Timing:
18  cool_down_time: dt.timedelta = dt.timedelta(seconds=1)
19  # General purpose time unit.
20  unit: dt.timedelta = dt.timedelta(seconds=1)
21  # Used for upper bound / timeout limits independently.
22  timeout_unit: dt.timedelta = dt.timedelta()
23  run_timeout: dt.timedelta = dt.timedelta()
24  # Wait time after starting the browser and before running a workload.
25  start_delay: dt.timedelta = dt.timedelta()
26  # Wait time after running a workload and before stopping a browser.
27  stop_delay: dt.timedelta = dt.timedelta()
28
29  def __post_init__(self) -> None:
30    if self.cool_down_time.total_seconds() < 0:
31      raise ValueError(
32          f"Timing.cool_down_time must be >= 0, but got: {self.cool_down_time}")
33    if self.unit.total_seconds() <= 0:
34      raise ValueError(f"Timing.unit must be > 0, but got {self.unit}")
35    if self.timeout_unit:
36      if self.timeout_unit.total_seconds() <= 0:
37        raise ValueError(
38            f"Timing.timeout_unit must be > 0, but got {self.timeout_unit}")
39      if self.timeout_unit < self.unit:
40        raise ValueError(f"Timing.unit must be <= Timing.timeout_unit: "
41                         f"{self.unit} vs. {self.timeout_unit}")
42    if self.run_timeout.total_seconds() < 0:
43      raise ValueError(
44          f"Timing.run_timeout, must be >= 0, but got {self.run_timeout}")
45
46  def units(self, time: Union[float, int, dt.timedelta]) -> float:
47    if isinstance(time, dt.timedelta):
48      seconds = time.total_seconds()
49    else:
50      seconds = time
51    if seconds < 0:
52      raise ValueError(f"Unexpected negative time: {seconds}s")
53    return seconds / self.unit.total_seconds()
54
55  def _convert_to_seconds(
56      self, time_units: Union[float, int, dt.timedelta]) -> Union[float, int]:
57    if isinstance(time_units, dt.timedelta):
58      seconds = time_units.total_seconds()
59    else:
60      seconds = time_units
61    assert isinstance(seconds, (float, int))
62    if seconds < 0:
63      raise ValueError(f"Time-units must be >= 0, but got {seconds}")
64    return seconds
65
66  def timedelta(self, time_units: Union[float, int,
67                                        dt.timedelta]) -> dt.timedelta:
68    seconds_f = self._convert_to_seconds(time_units)
69    return self._to_safe_range(seconds_f * self.unit)
70
71  def timeout_timedelta(
72      self, time_units: Union[float, int, dt.timedelta]) -> dt.timedelta:
73    if self.has_no_timeout:
74      return SAFE_MAX_TIMEOUT_TIMEDELTA
75    seconds_f = self._convert_to_seconds(time_units)
76    return self._to_safe_range(seconds_f * (self.timeout_unit or self.unit))
77
78  def _to_safe_range(self, result: dt.timedelta) -> dt.timedelta:
79    if result > SAFE_MAX_TIMEOUT_TIMEDELTA:
80      return SAFE_MAX_TIMEOUT_TIMEDELTA
81    return result
82
83  @property
84  def has_no_timeout(self) -> bool:
85    return self.timeout_unit == dt.timedelta.max
86
87  def to_json(self) -> Dict[str, float]:
88    return {
89        "coolDownTime": self.cool_down_time.total_seconds(),
90        "unit": self.unit.total_seconds(),
91        "timeoutUnit": self.timeout_unit.total_seconds(),
92        "runTimeout": self.run_timeout.total_seconds(),
93    }
94