• 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 contextlib
9import datetime as dt
10import io
11import logging
12import pathlib
13from typing import Final, List, Optional, Sequence, Tuple
14from unittest import mock
15
16from pyfakefs import fake_filesystem_unittest
17from tests import test_helper
18from tests.crossbench import mock_browser
19from tests.crossbench.mock_helper import MockCLI, MockPlatform
20
21import crossbench
22from crossbench import path as pth
23from crossbench import plt
24from crossbench.benchmarks.loading.loadline_presets import \
25    LoadLineTabletBenchmark
26from crossbench.browsers.browser import Browser
27from crossbench.browsers.settings import Settings
28from crossbench.cli.cli import CrossBenchCLI
29from crossbench.cli.config.browser_variants import BrowserVariantsConfig
30from crossbench.cli.config.network import NetworkConfig
31from crossbench.cli.config.secrets import SecretsConfig
32
33
34class CrossbenchFakeFsTestCase(
35    fake_filesystem_unittest.TestCase, metaclass=abc.ABCMeta):
36
37  def setUp(self) -> None:
38    super().setUp()
39    self.setUpPyfakefs(modules_to_reload=[crossbench, mock_browser, pth])
40    # gettext is used extensively in argparse
41    gettext_patcher = mock.patch(
42        "gettext.dgettext", side_effect=lambda domain, message: message)
43    gettext_patcher.start()
44    self.addCleanup(gettext_patcher.stop)
45
46    sleep_patcher = mock.patch("time.sleep", return_value=None)
47    sleep_patcher.start()
48    self.addCleanup(sleep_patcher.stop)
49
50  def create_file(self,
51                  path_str: str,
52                  contents: Optional[str] = None) -> pathlib.Path:
53    path = pathlib.Path(path_str)
54    self.fs.create_file(path, contents=contents)
55    return path
56
57
58class BaseCrossbenchTestCase(CrossbenchFakeFsTestCase, metaclass=abc.ABCMeta):
59
60  def filter_splashscreen_urls(self, urls: Sequence[str]) -> List[str]:
61    return [url for url in urls if not url.startswith("data:")]
62
63  def setUp(self) -> None:
64    # Instantiate MockPlatform before setting up fake_filesystem so we can
65    # still interact with the original, real plt.Platform object for extracting
66    # basic system information.
67    self.platform = MockPlatform()  # pytype: disable=not-instantiable
68    super().setUp()
69    self._default_log_level = logging.getLogger().getEffectiveLevel()
70    logging.getLogger().setLevel(logging.CRITICAL)
71    for mock_browser_cls in mock_browser.ALL:
72      mock_browser_cls.setup_fs(self.fs)
73      self.assertTrue(mock_browser_cls.mock_app_path().exists())
74    self.out_dir = pathlib.Path("/tmp/results/test")
75    self.out_dir.parent.mkdir(parents=True)
76    self.fs.add_real_directory(
77        LoadLineTabletBenchmark.default_network_config_path().parent,
78        lazy_read=not test_helper.is_google_env())
79    if test_helper.is_google_env():
80      self.fs.add_real_directory("/build/cas")
81    self.browsers: List[mock_browser.MockBrowser] = [
82        mock_browser.MockChromeDev(
83            "dev", settings=Settings(platform=self.platform)),
84        mock_browser.MockChromeStable(
85            "stable", settings=Settings(platform=self.platform))
86    ]
87    mock_platform_patcher = mock.patch.object(plt, "PLATFORM", self.platform)
88    mock_platform_patcher.start()
89    self.addCleanup(mock_platform_patcher.stop)
90    for browser in self.browsers:
91      self.assertListEqual(browser.expected_js, [])
92    self.mock_args = mock.Mock(
93        wraps=False,
94        driver_path=None,
95        network_config=None,
96        browser_config=None,
97        viewport=None,
98        splash_screen=None,
99        secrets=SecretsConfig(),
100        wipe_system_user_data=False,
101        http_request_timeout=dt.timedelta(),
102        cache_dir=pathlib.Path("test_cache_dir"),
103        enable_features=None,
104        disable_features=None,
105        js_flags=None,
106        enable_field_trial_config=False,
107        network=NetworkConfig.default(),
108        probe=[],
109        other_browser_args=[],
110        driver_logging=False)
111
112  def tearDown(self) -> None:
113    logging.getLogger().setLevel(self._default_log_level)
114    self.assertListEqual(self.platform.sh_results, [])
115    super().tearDown()
116
117
118class SysExitTestException(Exception):
119
120  def __init__(self, exit_code=0):
121    super().__init__("sys.exit")
122    self.exit_code = exit_code
123
124
125class BaseCliTestCase(BaseCrossbenchTestCase):
126
127  SPLASH_URLS_LEN: Final[int] = 2
128
129  def setUp(self) -> None:
130    super().setUp()
131
132    # tabulate and textwrap can be slow for tests, let's mock them out.
133    def mock_tabulate(table, *args, **kwargs):
134      del args, kwargs
135      return str(table)
136
137    patcher = mock.patch("tabulate.tabulate", side_effect=mock_tabulate)
138    self.addCleanup(patcher.stop)
139    patcher.start()
140
141    def mock_wrap(text, *args, **kwargs):
142      del args, kwargs
143      return [text]
144
145    patcher = mock.patch("textwrap.wrap", side_effect=mock_wrap)
146    self.addCleanup(patcher.stop)
147    patcher.start()
148
149  def run_cli_output(self,
150                     *args,
151                     raises=None,
152                     enable_logging: bool = True) -> Tuple[MockCLI, str, str]:
153    with mock.patch(
154        "sys.stdout", new_callable=io.StringIO) as mock_stdout, mock.patch(
155            "sys.stderr", new_callable=io.StringIO) as mock_stderr:
156      cli = self.run_cli(*args, raises=raises, enable_logging=enable_logging)
157    stdout = mock_stdout.getvalue()
158    stderr = mock_stderr.getvalue()
159    # Make sure we don't accidentally reuse the buffers across run_cli calls.
160    mock_stdout.close()
161    mock_stderr.close()
162    return cli, stdout, stderr
163
164  def run_cli(self,
165              *args,
166              raises=None,
167              enable_logging: bool = False) -> MockCLI:
168    cli = MockCLI(platform=self.platform, enable_logging=enable_logging)
169    with mock.patch(
170        "sys.exit", side_effect=SysExitTestException), mock.patch.object(
171            plt, "PLATFORM", self.platform):
172      if raises:
173        with self.assertRaises(raises):
174          cli.run(args)
175      else:
176        cli.run(args)
177    return cli
178
179  def mock_chrome_stable(self):
180    return mock.patch.object(
181        BrowserVariantsConfig,
182        "get_browser_cls",
183        return_value=mock_browser.MockChromeStable)
184
185  @contextlib.contextmanager
186  def patch_get_browser(self, return_value: Optional[Sequence[Browser]] = None):
187    if not return_value:
188      return_value = self.browsers
189    with mock.patch.object(
190        CrossBenchCLI, "_get_browsers", return_value=return_value):
191      yield
192