blob: 479ad5c472ce2da74b20aef9da998e0103f6d618 [file] [log] [blame]
#!/usr/bin/python2
"""Webdriver-based benchmarks for partial layout.
Benchmarks for partial layout changes in Cobalt.
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import argparse
import importlib
import os
import re
import socket
import sys
import threading
import unittest
arg_parser = argparse.ArgumentParser(
description="Runs Webdriver-based Cobalt benchmarks")
arg_parser.add_argument(
"-p", "--platform",
help="Cobalt platform, eg 'linux-x64x11'."
"Fetched from environment if absent.")
arg_parser.add_argument(
"-e", "--executable",
help="Path to cobalt executable. "
"Auto-derived if absent.")
arg_parser.add_argument(
"-c", "--config",
choices=["debug", "devel", "qa", "gold"],
help="Build config (eg, 'qa' or 'devel'). Not used if "
"--executable is specified. Fetched from environment "
"if needed and absent.")
arg_parser.add_argument(
"-d", "--devkit_name",
help="Devkit or IP address for app_launcher."
"Current hostname used if absent.")
arg_parser.add_argument(
"-o", "--log_file", help="Logfile pathname. stdout if absent.")
# Pattern to match Cobalt log line for when the WebDriver port has been
# opened.
RE_WEBDRIVER_LISTEN = re.compile(r"Starting WebDriver server on port (\d+)$")
COBALT_WEBDRIVER_CAPABILITIES = {
"browserName": "cobalt",
"javascriptEnabled": True,
"platform": "LINUX"
}
_webdriver = None
def GetWebDriver():
"""Returns the active connect WebDriver instance."""
return _webdriver
def ImportSeleniumModule(submodule=None):
"""Dynamically imports a selenium.webdriver submodule.
This is done because selenium 3.0 is not commonly pre-installed
on workstations, and we want to have a friendly error message for that
case.
Args:
submodule: module subpath underneath "selenium.webdriver"
Returns:
appropriate module
"""
if submodule:
module_path = ".".join(("selenium", submodule))
else:
module_path = "selenium"
# As of this writing, Google uses selenium 3.0.0b2 internally, so
# thats what we will target here as well.
try:
module = importlib.import_module(module_path)
if submodule is None:
# Only the top-level module has __version__
if not module.__version__.startswith("3.0"):
raise ImportError("Not version 3.0.x")
except ImportError:
sys.stderr.write("Could not import {}\n"
"Please install selenium >= 3.0.0b2.\n"
"Commonly: \"sudo pip install 'selenium>=3.0.0b2'\"\n"
.format(module_path))
sys.exit(1)
return module
class CobaltRunner(object):
"""Runs a Cobalt browser w/ a WebDriver client attached."""
test_script_started = threading.Event()
should_exit = threading.Event()
selenium_webdriver_module = None
webdriver = None
launcher = None
log_file_path = None
thread = None
def __init__(self, platform, executable, devkit_name, log_file_path):
self.selenium_webdriver_module = ImportSeleniumModule("webdriver")
script_path = os.path.dirname(__file__)
sys.path.append(script_path + "/../../tools/lbshell/")
app_launcher = importlib.import_module("app_launcher")
self.launcher = app_launcher.CreateLauncher(
platform, executable, devkit_name=devkit_name, close_output_file=False)
self.launcher.SetArgs(["--enable_webdriver"])
self.launcher.SetOutputCallback(self._HandleLine)
self.log_file_path = log_file_path
self.log_file = None
def __enter__(self):
self.thread = threading.Thread(target=self.Run)
self.thread.start()
self.WaitForStart()
def __exit__(self, exc_type, exc_value, traceback):
self.SetShouldExit()
self.thread.join()
def _HandleLine(self, line):
"""Internal log line callback."""
done = self.should_exit.is_set()
# Wait for WebDriver port here then connect
if self.test_script_started.is_set():
return done
match = RE_WEBDRIVER_LISTEN.search(line)
if not match:
return done
port = match.group(1)
self._StartWebdriver(port)
return done
def SetShouldExit(self):
"""Indicates cobalt process should exit. Done at next log line output."""
self.should_exit.set()
def _GetIPAddress(self):
return self.launcher.GetIPAddress()
def _StartWebdriver(self, port):
global _webdriver
url = "http://{}:{}/".format(self._GetIPAddress(), port)
self.webdriver = self.selenium_webdriver_module.Remote(
url, COBALT_WEBDRIVER_CAPABILITIES)
_webdriver = self.webdriver
self.test_script_started.set()
def WaitForStart(self):
"""Waits for the webdriver client to attach to cobalt."""
self.test_script_started.wait()
def Run(self):
"""Thread run routine."""
# Use stdout if log_file_path is unspecified
# If log_file_path is specified, make sure to close it
to_close = None
try:
if self.log_file_path:
self.log_file = open(self.log_file_path, "w")
to_close = self.log_file
else:
self.log_file = sys.stdout
self.launcher.SetOutputFile(self.log_file)
self.launcher.Run()
# This is watched for in webdriver_benchmark_test.py
sys.stdout.write("partial_layout_benchmark TEST COMPLETE\n")
finally:
if to_close:
to_close.close()
return 0
def GetCobaltExecutablePath(platform, config):
"""Auto-derives a path to a cobalt executable."""
if config is None:
try:
config = os.environ["BUILD_TYPE"]
except KeyError:
sys.stderr.write("Must specify --config or --executable\n")
sys.exit(1)
script_dir = os.path.dirname(os.path.realpath(__file__))
out_dir = os.path.join(script_dir, "..", "..", "out")
executable_directory = os.path.join(out_dir, "{}_{}".format(platform, config))
return os.path.join(executable_directory, "cobalt")
def main():
args = arg_parser.parse_args()
# Keep unittest module from seeing these args
sys.argv = sys.argv[:1]
platform = args.platform
if platform is None:
try:
platform = os.environ["BUILD_PLATFORM"]
except KeyError:
sys.stderr.write("Must specify --platform\n")
sys.exit(1)
executable = args.executable
if executable is None:
executable = GetCobaltExecutablePath(platform, args.config)
devkit_name = args.devkit_name
if devkit_name is None:
devkit_name = socket.gethostname()
with CobaltRunner(platform, executable, devkit_name, args.log_file):
unittest.main()
return 0
if __name__ == "__main__":
sys.exit(main())