• 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 datetime as dt
8import logging
9from typing import TYPE_CHECKING, Optional, Tuple, cast
10
11from immutabledict import immutabledict
12
13from crossbench.action_runner.action.action_type import ActionType
14from crossbench.action_runner.action.get import GetAction
15from crossbench.benchmarks.loading.page.base import Page, get_action_runner
16from crossbench.benchmarks.loading.playback_controller import \
17    PlaybackController
18from crossbench.benchmarks.loading.tab_controller import TabController
19
20if TYPE_CHECKING:
21  from crossbench.action_runner.base import ActionRunner
22  from crossbench.benchmarks.loading.config.blocks import ActionBlock
23  from crossbench.benchmarks.loading.config.login.custom import LoginBlock
24  from crossbench.cli.config.secrets import SecretsDict
25  from crossbench.runner.run import Run
26  from crossbench.types import JsonDict
27
28
29class InteractivePage(Page):
30
31  def __init__(self,
32               name: str,
33               blocks: Tuple[ActionBlock, ...],
34               setup: Optional[ActionBlock] = None,
35               login: Optional[LoginBlock] = None,
36               secrets: Optional[SecretsDict] = None,
37               playback: PlaybackController = PlaybackController.default(),
38               tabs: TabController = TabController.default(),
39               about_blank_duration: dt.timedelta = dt.timedelta(),
40               run_login: bool = True,
41               run_setup: bool = True):
42    assert name, "missing name"
43    self._name: str = name
44    assert isinstance(blocks, tuple)
45    self._blocks: Tuple[ActionBlock, ...] = blocks
46    assert self._blocks, "Must have at least 1 valid action"
47    assert not any(block.is_login for block in blocks), (
48        "No login blocks allowed as normal action block")
49    self._setup_block = setup
50    self._login_block = login
51    self._secrets: SecretsDict = secrets or immutabledict()
52    self._run_login = run_login
53    self._run_setup = run_setup
54    duration = self._get_duration()
55    super().__init__(self._name, duration, playback, tabs, about_blank_duration)
56
57  @property
58  def login_block(self) -> Optional[ActionBlock]:
59    return self._login_block
60
61  @property
62  def setup_block(self) -> Optional[ActionBlock]:
63    return self._setup_block
64
65  @property
66  def blocks(self) -> Tuple[ActionBlock, ...]:
67    return self._blocks
68
69  @property
70  def secrets(self) -> SecretsDict:
71    return self._secrets
72
73  @property
74  def first_url(self) -> str:
75    for block in self.blocks:
76      for action in block:
77        if action.TYPE == ActionType.GET:
78          return cast(GetAction, action).url
79    raise RuntimeError("No GET action with an URL found.")
80
81  def create_failure_artifacts(self,
82                               run: Run,
83                               message: str = "failure") -> None:
84    action_runner = get_action_runner(run)
85    try:
86      action_runner.screenshot_impl(run, message)
87    except Exception as e:  # pylint: disable=broad-except
88      logging.error("Failed to take a failure screenshot: %s", str(e))
89
90    try:
91      action_runner.dump_html_impl(run, message)
92    except Exception as e:  # pylint: disable=broad-except
93      logging.error("Failed to dump HTML on failure: %s", str(e))
94
95  def setup(self, run: Run) -> None:
96    action_runner = get_action_runner(run)
97    if self._run_login and (login_block := self.login_block):
98      action_runner.run_login(run, self, login_block)
99    if self._run_setup and (setup_block := self.setup_block):
100      action_runner.run_setup(run, self, setup_block)
101
102  def run(self, run: Run) -> None:
103    action_runner = get_action_runner(run)
104    multiple_tabs = self.tabs.multiple_tabs
105    for _ in self._playback:
106      action_runner.run_interactive_page(run, self, multiple_tabs)
107
108  def run_with(self, run: Run, action_runner: ActionRunner,
109               multiple_tabs: bool) -> None:
110    action_runner.run_interactive_page(run, self, multiple_tabs)
111
112  def details_json(self) -> JsonDict:
113    result = super().details_json()
114    result["actions"] = list(block.to_json() for block in self._blocks)
115    return result
116
117  def _get_duration(self) -> dt.timedelta:
118    duration = dt.timedelta()
119    for block in self._blocks:
120      duration += block.duration
121    return duration
122