• 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 datetime as dt
8import logging
9import sys
10from typing import TYPE_CHECKING, Any, Optional, Sequence, Union
11
12from crossbench import helper
13
14if TYPE_CHECKING:
15  from crossbench import plt
16  from crossbench.browsers.browser import Browser
17  from crossbench.exception import ExceptionAnnotationScope
18  from crossbench.runner.run import Run
19  from crossbench.runner.runner import Runner
20  from crossbench.runner.timing import Timing
21
22
23class Actions(helper.TimeScope):
24
25  _max_end_datetime: dt.datetime
26
27  def __init__(self,
28               message: str,
29               run: Run,
30               runner: Optional[Runner] = None,
31               browser: Optional[Browser] = None,
32               verbose: bool = False,
33               measure: bool = True,
34               timeout: dt.timedelta = dt.timedelta()):
35    assert message, "Actions need a name"
36    super().__init__(message)
37    self._exception_annotation: ExceptionAnnotationScope = run.exceptions.info(
38        f"Action: {message}")
39    self._run: Run = run
40    self._browser: Browser = browser or run.browser
41    self._runner: Runner = runner or run.runner
42    self._is_active: bool = False
43    self._verbose: bool = verbose
44    self._measure = measure
45    if timeout:
46      self._max_end_datetime = min(dt.datetime.now() + timeout,
47                                   run.max_end_datetime())
48    else:
49      self._max_end_datetime = run.max_end_datetime()
50
51  @property
52  def timing(self) -> Timing:
53    return self._runner.timing
54
55  @property
56  def run(self) -> Run:
57    return self._run
58
59  @property
60  def platform(self) -> plt.Platform:
61    return self._run.browser_platform
62
63  def __enter__(self) -> Actions:
64    self._exception_annotation.__enter__()
65    super().__enter__()
66    self._is_active = True
67    logging.debug("Action begin: %s", self._message)
68    if self._verbose:
69      logging.info(self._message.ljust(30))
70    else:
71      # Print message that doesn't overlap with helper.Spinner
72      sys.stdout.write(f"   {self._message.ljust(30)}\r")
73    return self
74
75  def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
76    self._is_active = False
77    self._exception_annotation.__exit__(exc_type, exc_value, exc_traceback)
78    super().__exit__(exc_type, exc_value, exc_traceback)
79    logging.debug("Action end: %s", self._message)
80    if self._measure:
81      self.run.durations[f"actions-duration {self.message}"] = self.duration
82
83  def _assert_is_active(self) -> None:
84    assert self._is_active, "Actions have to be used in a with scope"
85
86  def current_window_id(self) -> str:
87    return self._browser.current_window_id()
88
89  def switch_window(self, window_id: str) -> None:
90    self._browser.switch_window(window_id)
91
92  def js(self,
93         js_code: str,
94         timeout: Union[int, float, dt.timedelta] = 10,
95         arguments: Sequence[object] = (),
96         **kwargs) -> Any:
97    self._assert_is_active()
98    assert js_code, "js_code must be a valid JS script"
99    if kwargs:
100      js_code = js_code.format(**kwargs)
101    delta = self.timing.timeout_timedelta(timeout)
102    return self._browser.js(js_code, delta, arguments=arguments)
103
104  def wait_js_condition(self,
105                        js_code: str,
106                        min_wait: Union[dt.timedelta, float],
107                        timeout: Union[dt.timedelta, float],
108                        delay: Union[dt.timedelta, float] = 0) -> None:
109    wait_range = helper.WaitRange(
110        min=self.timing.timedelta(min_wait),
111        timeout=self.timing.timeout_timedelta(timeout),
112        delay=delay)
113    assert "return" in js_code, (
114        f"Missing return statement in js-wait code: {js_code}")
115    for _, time_left in wait_range.wait_with_backoff():
116      time_units = self.timing.units(time_left)
117      result = self.js(js_code, timeout=time_units, absolute_time=True)
118      if result:
119        return
120      assert result is False, (
121          f"js_code did not return a bool, but got: {result}\n"
122          f"js-code: {js_code}")
123
124  def show_url(self, url: str, target: Optional[str] = None) -> None:
125    self._assert_is_active()
126    if target and target in ("_blank", "_parent", "_top"):
127      # TODO: use target in the driver instead.
128      self.js(f"window.open('{url}','{target}');")
129    else:
130      if target not in (None, "_self", "_new_tab", "_new_window"):
131        raise ValueError(f"Invalid target: {target}")
132      self._browser.show_url(url, target=target)
133
134  def wait(
135      self, seconds: Union[dt.timedelta,
136                           float] = dt.timedelta(seconds=1)) -> None:
137    self._assert_is_active()
138    self.platform.sleep(seconds)
139