• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2022 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
9import logging
10import os
11import shlex
12from typing import TYPE_CHECKING, Any, Iterable, Optional, Sequence, Tuple
13
14from ordered_set import OrderedSet
15
16from crossbench import path as pth
17from crossbench import plt
18from crossbench.browsers.settings import Settings
19from crossbench.flags.base import Flags, FlagsData, FlagsT
20
21if TYPE_CHECKING:
22  import re
23
24  from crossbench.browsers.attributes import BrowserAttributes
25  from crossbench.browsers.splash_screen import SplashScreen
26  from crossbench.browsers.viewport import Viewport
27  from crossbench.cli.config.secrets import Secret, SecretsDict
28  from crossbench.env import HostEnvironment
29  from crossbench.flags.chrome import ChromeFeatures
30  from crossbench.flags.js_flags import JSFlags
31  from crossbench.network.base import Network
32  from crossbench.probes.probe import Probe
33  from crossbench.runner.groups.session import BrowserSessionRunGroup
34  from crossbench.types import JsonDict
35
36
37class Browser(abc.ABC):
38
39  @classmethod
40  def default_flags(cls, initial_data: FlagsData = None) -> Flags:
41    return Flags(initial_data)
42
43  def __init__(self,
44               label: str,
45               path: Optional[pth.AnyPath] = None,
46               settings: Optional[Settings] = None):
47    self._settings = settings or Settings()
48    self._platform = self._settings.platform
49    self.label: str = label
50    self._unique_name: str = ""
51    self.app_name: str = self.type_name
52    self.version: str = "custom"
53    self.major_version: int = 0
54    self.app_path: pth.AnyPath = pth.AnyPath()
55    self.path = pth.AnyPath()
56    self._setup_path(path)
57    self._is_running: bool = False
58    self._pid: Optional[int] = None
59    self._probes: OrderedSet[Probe] = OrderedSet()
60    self._flags: Flags = self._setup_flags(self._settings)
61    self.log_file: Optional[pth.AnyPath] = None
62    self.cache_dir: Optional[pth.AnyPath] = self._settings.cache_dir
63    self.clear_cache_dir: bool = True
64    self._setup_cache_dir(self._settings)
65
66  def _setup_path(self, path: Optional[pth.AnyPath] = None) -> None:
67    if not path:
68      # TODO: separate class for remote browser (selenium) without an explicit
69      # binary path.
70      self.unique_name = f"{self.type_name}_{self.label}".lower()
71      return
72    self.path = self._resolve_binary(path)
73    # TODO clean up
74    if not self.platform.is_android:
75      assert self.path.is_absolute()
76    self.version = self._extract_version()
77    self.major_version = int(self.version.split(".")[0])
78    self.unique_name = f"{self.type_name}_v{self.major_version}_{self.label}"
79
80  def _setup_flags(self, settings: Settings) -> Flags:
81    assert not self._settings.js_flags, (
82        f"{self} doesn't support custom js_flags")
83    return self.default_flags(settings.flags)
84
85  def _setup_cache_dir(self, settings: Settings) -> None:
86    pass
87
88  @property
89  @abc.abstractmethod
90  def type_name(self) -> str:
91    pass
92
93  @property
94  @abc.abstractmethod
95  def attributes(self) -> BrowserAttributes:
96    pass
97
98  @property
99  def platform(self) -> plt.Platform:
100    return self._platform
101
102  @property
103  def host_platform(self) -> plt.Platform:
104    return self._platform.host_platform
105
106  @property
107  def unique_name(self) -> str:
108    return self._unique_name
109
110  @unique_name.setter
111  def unique_name(self, name: str) -> None:
112    assert name
113    # Replace any potentially unsafe chars in the name
114    self._unique_name = pth.safe_filename(name).lower()
115
116  @property
117  def network(self) -> Network:
118    return self._settings.network
119
120  @property
121  def secrets(self) -> SecretsDict:
122    return self._settings.secrets
123
124  @property
125  def splash_screen(self) -> SplashScreen:
126    return self._settings.splash_screen
127
128  @property
129  def viewport(self) -> Viewport:
130    return self._settings.viewport
131
132  @viewport.setter
133  def viewport(self, value: Viewport) -> None:
134    self._settings.viewport = value
135
136  @property
137  def wipe_system_user_data(self) -> bool:
138    return self._settings.wipe_system_user_data
139
140  @property
141  def http_request_timeout(self) -> dt.timedelta:
142    return self._settings.http_request_timeout
143
144  @property
145  def probes(self) -> Iterable[Probe]:
146    return iter(self._probes)
147
148  @property
149  def flags(self) -> Flags:
150    return self._flags
151
152  @property
153  def features(self) -> ChromeFeatures:
154    raise NotImplementedError(f"Unsupported feature flags on {self}.")
155
156  @property
157  def js_flags(self) -> JSFlags:
158    raise NotImplementedError(f"Unsupported feature flags on {self}.")
159
160  def user_agent(self) -> str:
161    return str(self.js("return window.navigator.userAgent"))
162
163  @property
164  def pid(self) -> Optional[int]:
165    return self._pid
166
167  @property
168  def is_running_process(self) -> Optional[bool]:
169    # TODO: activate this method again
170    if self.pid is None:
171      return None
172    info = self.platform.process_info(self.pid)
173    if info is None:
174      return None
175    if status := info.get("status"):
176      return status in ("running", "sleeping")
177    # TODO(cbruni): fix posix process_info for remote platforms where
178    # we don't get the status back.
179    return False
180
181  @property
182  def is_running(self) -> bool:
183    return self._is_running
184
185  def validate_env(self, env: HostEnvironment) -> None:
186    """Called before starting a browser / browser session to perform
187    a pre-run checklist."""
188
189  @property
190  def is_local(self) -> bool:
191    return self.platform.is_local
192
193  @property
194  def is_remote(self) -> bool:
195    return self.platform.is_remote
196
197  def set_log_file(self, path: pth.AnyPath) -> None:
198    self.log_file = path
199
200  @property
201  def stdout_log_file(self) -> pth.AnyPath:
202    assert self.log_file
203    return self.log_file.with_suffix(".stdout.log")
204
205  def _resolve_binary(self, path: pth.AnyPath) -> pth.AnyPath:
206    path = self.platform.absolute(path)
207    assert self.platform.exists(path), f"Binary at path={path} does not exist."
208    self.app_path = path
209    self.app_name = self.app_path.stem
210    if self.platform.is_macos:
211      path = self._resolve_macos_binary(path)
212    assert self.platform.is_file(path), (
213        f"Binary at path={path} is not a file.")
214    return path
215
216  def _resolve_macos_binary(self, path: pth.AnyPath) -> pth.AnyPath:
217    assert self.platform.is_macos
218    candidate = self.platform.search_binary(path)
219    if not candidate or not self.platform.is_file(candidate):
220      raise ValueError(f"Could not find browser executable in {path}")
221    return candidate
222
223  def attach_probe(self, probe: Probe) -> None:
224    if probe in self._probes:
225      raise ValueError(f"Cannot attach same probe twice: {probe}")
226    self._probes.add(probe)
227    probe.attach(self)
228
229  def details_json(self) -> JsonDict:
230    return {
231        "label": self.label,
232        "browser": self.type_name,
233        "unique_name": self.unique_name,
234        "app_name": self.app_name,
235        "version": self.version,
236        "flags": tuple(self.flags),
237        "js_flags": tuple(),
238        "path": os.fspath(self.path),
239        "clear_cache_dir": self.clear_cache_dir,
240        "major_version": self.major_version,
241        "log": {}
242    }
243
244  def validate_binary(self) -> None:
245    """ Helper method is called from the Runner before any Runs / Sessions
246    have started."""
247
248  def setup_binary(self) -> None:
249    """ This helper is called in the setup steps of each Session.
250    This can be used to install a custom binary on remote devices. """
251
252  def setup(self, session: BrowserSessionRunGroup) -> None:
253    assert not self._is_running, (
254        "Previously used browser was not correctly stopped.")
255    self.clear_cache()
256    self.start(session)
257    assert self._is_running
258
259  def is_logged_in(self, secret: Secret, strict: bool = False) -> bool:
260    """Determines whether the browser is already logged in with the given
261    credentials.
262
263    Args:
264      secret: The credentials to check.
265      strict: Whether or not to raise an error if login is impossible
266
267    Returns:
268      True if and only if the browser is already logged in with the account
269
270    Raises:
271      RuntimeError: If strict, when logging in with the given cridentials is
272      not possible.
273    """
274    del secret
275    del strict
276    return False
277
278  @abc.abstractmethod
279  def _extract_version(self) -> str:
280    pass
281
282  def clear_cache(self) -> None:
283    if self.clear_cache_dir and self.cache_dir:
284      self.platform.rm(self.cache_dir, missing_ok=True, dir=True)
285      self.platform.mkdir(self.cache_dir, parents=True)
286
287  @abc.abstractmethod
288  def start(self, session: BrowserSessionRunGroup) -> None:
289    pass
290
291  def _log_browser_start(self,
292                         args: Tuple[str, ...],
293                         driver_path: Optional[pth.AnyPath] = None) -> None:
294    logging.info("STARTING BROWSER Binary:  %s", self.path)
295    logging.info("STARTING BROWSER Version: %s", self.version)
296    if driver_path:
297      logging.info("STARTING BROWSER Driver:  %s", driver_path)
298    logging.info("STARTING BROWSER Network: %s", self.network)
299    logging.info("STARTING BROWSER Probes:  %s",
300                 ", ".join(p.NAME for p in self.probes))
301    logging.info("STARTING BROWSER Flags:   %s", shlex.join(args))
302
303  def _get_browser_flags_for_session(
304      self, session: BrowserSessionRunGroup) -> Tuple[str, ...]:
305    flags_copy: Flags = self.flags.copy()
306    flags_copy.update(session.extra_flags)
307    flags_copy.update(self.network.extra_flags(self.attributes))
308    flags_copy = self._filter_flags_for_run(flags_copy)
309    return tuple(flags_copy)
310
311  def _filter_flags_for_run(self, flags: FlagsT) -> FlagsT:
312    return flags
313
314  def quit(self) -> None:
315    assert self._is_running, "Browser is already stopped"
316    try:
317      self.force_quit()
318    finally:
319      self._pid = None
320
321  def force_quit(self) -> None:
322    if not self._is_running:
323      return
324    logging.info("Browser.force_quit()")
325    if self.platform.is_macos:
326      self.platform.exec_apple_script(f"""
327  tell application "{self.app_path}"
328    quit
329  end tell
330      """)
331    elif self._pid:
332      self.platform.terminate(self._pid)
333    self._is_running = False
334
335  @abc.abstractmethod
336  def js(
337      self,
338      script: str,
339      timeout: Optional[dt.timedelta] = None,
340      arguments: Sequence[object] = ()
341  ) -> Any:
342    pass
343
344  def run_script_on_new_document(self, script: str) -> None:
345    del script
346    raise NotImplementedError(
347        f"New document script injection is not supported by {self}")
348
349  def current_window_id(self) -> str:
350    raise NotImplementedError(f"current_window_id is not implemented by {self}")
351
352  def switch_window(self, window_id: str) -> None:
353    del window_id
354    raise NotImplementedError(f"switch_window is not implemented by {self}")
355
356  def switch_tab(
357      self,
358      title: Optional[re.Pattern] = None,
359      url: Optional[re.Pattern] = None,
360      tab_index: Optional[int] = None,
361      timeout: dt.timedelta = dt.timedelta(seconds=0)
362  ) -> None:
363    del title
364    del url
365    del tab_index
366    del timeout
367    raise NotImplementedError(f"Switching tabs is not supported by {self}")
368
369  @abc.abstractmethod
370  def show_url(self, url: str, target: Optional[str] = None) -> None:
371    pass
372
373  def switch_to_new_tab(self) -> None:
374    raise NotImplementedError(f"New tab is not supported by {self}")
375
376  def screenshot(self, path: pth.LocalPath) -> None:
377    # TODO: implement screenshot on browser and platform.
378    raise NotImplementedError(f"Taking screenshots is not supported by {self}")
379
380  def _sync_viewport_flag(self, flags: Flags, flag: str,
381                          is_requested_by_viewport: bool,
382                          replacement: Viewport) -> None:
383    if is_requested_by_viewport:
384      flags.set(flag)
385    elif flag in flags:
386      if self.viewport.is_default:
387        self.viewport = replacement
388      else:
389        raise ValueError(
390            f"{flag} conflicts with requested --viewport={self.viewport}")
391
392  def __str__(self) -> str:
393    platform_prefix = ""
394    if self.platform.is_remote:
395      platform_prefix = str(self.platform)
396    return f"{platform_prefix}{self.type_name.capitalize()}:{self.label}"
397
398  def __hash__(self) -> int:
399    # Poor-man's hash, browsers should be unique.
400    return hash(id(self))
401
402  def performance_mark(self, name: str):
403    self.js("performance.mark(arguments[0]);", arguments=[name])
404