1import os 2import subprocess 3import tempfile 4 5from mozprocess import ProcessHandler 6 7from tools.serve.serve import make_hosts_file 8 9from .base import Browser, require_arg, get_free_port, browser_command, ExecutorBrowser 10from .base import get_timeout_multiplier # noqa: F401 11from ..executors import executor_kwargs as base_executor_kwargs 12from ..executors.executorservodriver import (ServoWebDriverTestharnessExecutor, # noqa: F401 13 ServoWebDriverRefTestExecutor) # noqa: F401 14from ..process import cast_env 15 16here = os.path.join(os.path.split(__file__)[0]) 17 18__wptrunner__ = { 19 "product": "servodriver", 20 "check_args": "check_args", 21 "browser": "ServoWebDriverBrowser", 22 "executor": { 23 "testharness": "ServoWebDriverTestharnessExecutor", 24 "reftest": "ServoWebDriverRefTestExecutor", 25 }, 26 "browser_kwargs": "browser_kwargs", 27 "executor_kwargs": "executor_kwargs", 28 "env_extras": "env_extras", 29 "env_options": "env_options", 30 "timeout_multiplier": "get_timeout_multiplier", 31 "update_properties": "update_properties", 32} 33 34 35def check_args(**kwargs): 36 require_arg(kwargs, "binary") 37 38 39def browser_kwargs(test_type, run_info_data, config, **kwargs): 40 return { 41 "binary": kwargs["binary"], 42 "binary_args": kwargs["binary_args"], 43 "debug_info": kwargs["debug_info"], 44 "server_config": config, 45 "user_stylesheets": kwargs.get("user_stylesheets"), 46 "headless": kwargs.get("headless"), 47 } 48 49 50def executor_kwargs(test_type, server_config, cache_manager, run_info_data, **kwargs): 51 rv = base_executor_kwargs(test_type, server_config, 52 cache_manager, run_info_data, **kwargs) 53 return rv 54 55 56def env_extras(**kwargs): 57 return [] 58 59 60def env_options(): 61 return {"server_host": "127.0.0.1", 62 "testharnessreport": "testharnessreport-servodriver.js", 63 "supports_debugger": True} 64 65 66def update_properties(): 67 return (["debug", "os", "processor"], {"os": ["version"], "processor": ["bits"]}) 68 69 70def write_hosts_file(config): 71 hosts_fd, hosts_path = tempfile.mkstemp() 72 with os.fdopen(hosts_fd, "w") as f: 73 f.write(make_hosts_file(config, "127.0.0.1")) 74 return hosts_path 75 76class ServoWebDriverBrowser(Browser): 77 init_timeout = 300 # Large timeout for cases where we're booting an Android emulator 78 79 def __init__(self, logger, binary, debug_info=None, webdriver_host="127.0.0.1", 80 server_config=None, binary_args=None, user_stylesheets=None, headless=None): 81 Browser.__init__(self, logger) 82 self.binary = binary 83 self.binary_args = binary_args or [] 84 self.webdriver_host = webdriver_host 85 self.webdriver_port = None 86 self.proc = None 87 self.debug_info = debug_info 88 self.hosts_path = write_hosts_file(server_config) 89 self.server_ports = server_config.ports if server_config else {} 90 self.command = None 91 self.user_stylesheets = user_stylesheets if user_stylesheets else [] 92 self.headless = headless if headless else False 93 self.ca_certificate_path = server_config.ssl_config["ca_cert_path"] 94 95 def start(self, **kwargs): 96 self.webdriver_port = get_free_port() 97 98 env = os.environ.copy() 99 env["HOST_FILE"] = self.hosts_path 100 env["RUST_BACKTRACE"] = "1" 101 env["EMULATOR_REVERSE_FORWARD_PORTS"] = ",".join( 102 str(port) 103 for _protocol, ports in self.server_ports.items() 104 for port in ports 105 if port 106 ) 107 108 debug_args, command = browser_command( 109 self.binary, 110 self.binary_args + [ 111 "--hard-fail", 112 "--webdriver=%s" % self.webdriver_port, 113 "about:blank", 114 ], 115 self.debug_info 116 ) 117 118 if self.headless: 119 command += ["--headless"] 120 121 if self.ca_certificate_path: 122 command += ["--certificate-path", self.ca_certificate_path] 123 124 for stylesheet in self.user_stylesheets: 125 command += ["--user-stylesheet", stylesheet] 126 127 self.command = command 128 129 self.command = debug_args + self.command 130 131 if not self.debug_info or not self.debug_info.interactive: 132 self.proc = ProcessHandler(self.command, 133 processOutputLine=[self.on_output], 134 env=cast_env(env), 135 storeOutput=False) 136 self.proc.run() 137 else: 138 self.proc = subprocess.Popen(self.command, env=cast_env(env)) 139 140 self.logger.debug("Servo Started") 141 142 def stop(self, force=False): 143 self.logger.debug("Stopping browser") 144 if self.proc is not None: 145 try: 146 self.proc.kill() 147 except OSError: 148 # This can happen on Windows if the process is already dead 149 pass 150 151 def pid(self): 152 if self.proc is None: 153 return None 154 155 try: 156 return self.proc.pid 157 except AttributeError: 158 return None 159 160 def on_output(self, line): 161 """Write a line of output from the process to the log""" 162 self.logger.process_output(self.pid(), 163 line.decode("utf8", "replace"), 164 command=" ".join(self.command)) 165 166 def is_alive(self): 167 return self.proc.poll() is None 168 169 def cleanup(self): 170 self.stop() 171 os.remove(self.hosts_path) 172 173 def executor_browser(self): 174 assert self.webdriver_port is not None 175 return ExecutorBrowser, {"webdriver_host": self.webdriver_host, 176 "webdriver_port": self.webdriver_port, 177 "init_timeout": self.init_timeout} 178