tor-browser

The Tor Browser
git clone https://git.dasho.dev/tor-browser.git
Log | Files | Refs | README | LICENSE

run_webpage_test.py (2586B)

      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 """Implements commands for running webpage tests."""
      5 
      6 import os
      7 import subprocess
      8 import time
      9 
     10 from contextlib import suppress
     11 from typing import List, Optional, Tuple
     12 
     13 import browser_runner
     14 from common import catch_sigterm, get_ip_address, wait_for_sigterm
     15 from test_runner import TestRunner
     16 
     17 _DEVTOOLS_PORT_FILE = 'webpage_test_runner.devtools.port'
     18 
     19 
     20 def capture_devtools_addr(proc: subprocess.Popen,
     21                          logs_dir: str) -> Tuple[str, int]:
     22    """Returns the devtools address and port initiated by the running |proc|.
     23    This function should only be used when the WebpageTestRunner is executed by
     24    a different process."""
     25    port_file = os.path.join(logs_dir, _DEVTOOLS_PORT_FILE)
     26 
     27    def try_reading_port():
     28        if not os.path.isfile(port_file):
     29            return None
     30        with open(port_file, encoding='utf-8') as inp:
     31            return inp.read().rsplit(':', 1)
     32 
     33    while True:
     34        result = try_reading_port()
     35        if result:
     36            return result
     37        proc.poll()
     38        assert not proc.returncode, 'Process stopped.'
     39        time.sleep(1)
     40 
     41 
     42 class WebpageTestRunner(TestRunner):
     43    """Test runner for running GPU tests."""
     44 
     45    def __init__(self, out_dir: str, test_args: List[str],
     46                 target_id: Optional[str], logs_dir: Optional[str]) -> None:
     47        super().__init__(out_dir, test_args, ['web_engine_shell'], target_id)
     48        self._runner = browser_runner.BrowserRunner(
     49            browser_runner.WEB_ENGINE_SHELL, target_id, out_dir)
     50        if logs_dir:
     51            self.port_file = os.path.join(logs_dir, _DEVTOOLS_PORT_FILE)
     52        else:
     53            self.port_file = None
     54 
     55    def run_test(self):
     56        catch_sigterm()
     57        self._runner.start()
     58        device_ip = get_ip_address(self._target_id, ipv4_only=True)
     59        addr = device_ip.exploded
     60        if device_ip.version == 6:
     61            addr = '[' + addr + ']'
     62        addr += ':' + str(self._runner.devtools_port)
     63        if self.port_file:
     64            with open(self.port_file, 'w') as out:
     65                out.write(addr)
     66        else:
     67            print('DevTools is running on ' + addr)
     68        try:
     69            wait_for_sigterm('shutting down the webpage.')
     70        finally:
     71            self._runner.close()
     72            if self.port_file:
     73                with suppress(OSError):
     74                    os.remove(self.port_file)
     75        return subprocess.CompletedProcess(args='', returncode=0)