Source code
Revision control
Copy as Markdown
Other Tools
# Copyright 2024 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
""" Starts a web engine shell on an existing fuchsia device, and returns a
ChromeDriver instance to control it."""
import logging
import os
import subprocess
from contextlib import AbstractContextManager
from typing import List
# From vpython wheel.
# pylint: disable=import-error
from selenium import webdriver
from selenium.webdriver import ChromeOptions
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
# pylint: enable=import-error
from common import get_ffx_isolate_dir, get_free_local_port
from isolate_daemon import IsolateDaemon
from run_webpage_test import capture_devtools_addr
LOG_DIR = os.environ.get('ISOLATED_OUTDIR', '/tmp')
class ChromeDriverWrapper(AbstractContextManager):
"""Manages the web engine shell on the device and the chromedriver
communicating with it. This class expects the chromedriver exists at
clang_x64/stripped/chromedriver in output dir."""
def __init__(self, extra_args: List[str] = None):
# The reference of the webdriver.Chrome instance.
self._driver = None
# Creates the isolate dir for daemon to ensure it can be shared across
# the processes. Note, it has no effect if isolate_dir has already been
# set.
self._isolate_dir = IsolateDaemon.IsolateDir()
# The process of the run_test.py webpage.
self._proc: subprocess.Popen = None
# Extra arguments sent to run_test.py webpage process.
self._extra_args = extra_args or []
def __enter__(self):
"""Starts the run_test.py and the chromedriver connecting to it, must be
executed before other commands."""
self._isolate_dir.__enter__()
logging.warning('ffx daemon is running in %s', get_ffx_isolate_dir())
self._proc = subprocess.Popen([
os.path.join(os.path.dirname(os.path.abspath(__file__)),
'run_test.py'), 'webpage', '--out-dir=.',
'--browser=web-engine-shell', '--device', f'--logs-dir={LOG_DIR}'
] + self._extra_args,
env={
**os.environ, 'CHROME_HEADLESS': '1'
})
address, port = capture_devtools_addr(self._proc, LOG_DIR)
logging.warning('DevTools is now running on %s:%s', address, port)
options = ChromeOptions()
options.debugger_address = f'{address}:{str(port)}'
# The port webdriver running on is not very interesting, the _driver
# instance will be used directly. So a random free local port is used.
self._driver = webdriver.Chrome(options=options,
service=Service(
os.path.join(
'clang_x64', 'stripped',
'chromedriver'),
get_free_local_port()))
self._driver.__enter__()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
"""Stops the run_test.py and the chromedriver, cannot perform other
commands afterward."""
try:
return self._driver.__exit__(exc_type, exc_val, exc_tb)
finally:
self._proc.terminate()
self._proc.wait()
self._isolate_dir.__exit__(exc_type, exc_val, exc_tb)
def __getattr__(self, name):
"""Forwards function calls to the underlying |_driver| instance."""
return getattr(self._driver, name)
# Explicitly override find_element_by_id to avoid importing selenium
# packages in the caller files.
# The find_element_by_id in webdriver.Chrome is deprecated.
# DeprecationWarning: find_element_by_* commands are deprecated. Please
# use find_element() instead
def find_element_by_id(self, id_str):
"""Returns the element in the page with id |id_str|."""
return self._driver.find_element(By.ID, id_str)