1 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import atexit 6 import logging 7 import os 8 import urllib2 9 import urlparse 10 11 try: 12 from selenium import webdriver 13 except ImportError: 14 # Ignore import error, as this can happen when builder tries to call the 15 # setup method of test that imports chromedriver. 16 logging.error('selenium module failed to be imported.') 17 pass 18 19 from autotest_lib.client.bin import utils 20 from autotest_lib.client.common_lib.cros import chrome 21 22 CHROMEDRIVER_EXE_PATH = '/usr/local/chromedriver/chromedriver' 23 X_SERVER_DISPLAY = ':0' 24 X_AUTHORITY = '/home/chronos/.Xauthority' 25 26 27 class chromedriver(object): 28 """Wrapper class, a context manager type, for tests to use Chrome Driver.""" 29 30 def __init__(self, extra_chrome_flags=[], subtract_extra_chrome_flags=[], 31 extension_paths=[], username=None, password=None, 32 server_port=None, skip_cleanup=False, url_base=None, 33 extra_chromedriver_args=None, *args, **kwargs): 34 """Initialize. 35 36 @param extra_chrome_flags: Extra chrome flags to pass to chrome, if any. 37 @param subtract_extra_chrome_flags: Remove default flags passed to 38 chrome by chromedriver, if any. 39 @param extension_paths: A list of paths to unzipped extensions. Note 40 that paths to crx files won't work. 41 @param username: Log in using this username instead of the default. 42 @param password: Log in using this password instead of the default. 43 @param server_port: Port number for the chromedriver server. If None, 44 an available port is chosen at random. 45 @param skip_cleanup: If True, leave the server and browser running 46 so that remote tests can run after this script 47 ends. Default is False. 48 @param url_base: Optional base url for chromedriver. 49 @param extra_chromedriver_args: List of extra arguments to forward to 50 the chromedriver binary, if any. 51 """ 52 self._cleanup = not skip_cleanup 53 assert os.geteuid() == 0, 'Need superuser privileges' 54 55 # Log in with telemetry 56 self._chrome = chrome.Chrome(extension_paths=extension_paths, 57 username=username, 58 password=password, 59 extra_browser_args=extra_chrome_flags) 60 self._browser = self._chrome.browser 61 # Close all tabs owned and opened by Telemetry, as these cannot be 62 # transferred to ChromeDriver. 63 self._browser.tabs[0].Close() 64 65 # Start ChromeDriver server 66 self._server = chromedriver_server(CHROMEDRIVER_EXE_PATH, 67 port=server_port, 68 skip_cleanup=skip_cleanup, 69 url_base=url_base, 70 extra_args=extra_chromedriver_args) 71 72 # Open a new tab using Chrome remote debugging. ChromeDriver expects 73 # a tab opened for remote to work. Tabs opened using Telemetry will be 74 # owned by Telemetry, and will be inaccessible to ChromeDriver. 75 urllib2.urlopen('http://localhost:%i/json/new' % 76 utils.get_chrome_remote_debugging_port()) 77 78 chromeOptions = {'debuggerAddress': 79 ('localhost:%d' % 80 utils.get_chrome_remote_debugging_port())} 81 capabilities = {'chromeOptions':chromeOptions} 82 # Handle to chromedriver, for chrome automation. 83 try: 84 self.driver = webdriver.Remote(command_executor=self._server.url, 85 desired_capabilities=capabilities) 86 except NameError: 87 logging.error('selenium module failed to be imported.') 88 raise 89 90 91 def __enter__(self): 92 return self 93 94 95 def __exit__(self, *args): 96 """Clean up after running the test. 97 98 """ 99 if hasattr(self, 'driver') and self.driver: 100 self.driver.close() 101 del self.driver 102 103 if not hasattr(self, '_cleanup') or self._cleanup: 104 if hasattr(self, '_server') and self._server: 105 self._server.close() 106 del self._server 107 108 if hasattr(self, '_browser') and self._browser: 109 self._browser.Close() 110 del self._browser 111 112 def get_extension(self, extension_path): 113 """Gets an extension by proxying to the browser. 114 115 @param extension_path: Path to the extension loaded in the browser. 116 117 @return: A telemetry extension object representing the extension. 118 """ 119 return self._chrome.get_extension(extension_path) 120 121 122 @property 123 def chrome_instance(self): 124 """ The chrome instance used by this chrome driver instance. """ 125 return self._chrome 126 127 128 class chromedriver_server(object): 129 """A running ChromeDriver server. 130 131 This code is migrated from chrome: 132 src/chrome/test/chromedriver/server/server.py 133 """ 134 135 def __init__(self, exe_path, port=None, skip_cleanup=False, 136 url_base=None, extra_args=None): 137 """Starts the ChromeDriver server and waits for it to be ready. 138 139 Args: 140 exe_path: path to the ChromeDriver executable 141 port: server port. If None, an available port is chosen at random. 142 skip_cleanup: If True, leave the server running so that remote 143 tests can run after this script ends. Default is 144 False. 145 url_base: Optional base url for chromedriver. 146 extra_args: List of extra arguments to forward to the chromedriver 147 binary, if any. 148 Raises: 149 RuntimeError if ChromeDriver fails to start 150 """ 151 if not os.path.exists(exe_path): 152 raise RuntimeError('ChromeDriver exe not found at: ' + exe_path) 153 154 chromedriver_args = [exe_path] 155 if port: 156 # Allow remote connections if a port was specified 157 chromedriver_args.append('--whitelisted-ips') 158 else: 159 port = utils.get_unused_port() 160 chromedriver_args.append('--port=%d' % port) 161 162 self.url = 'http://localhost:%d' % port 163 if url_base: 164 chromedriver_args.append('--url-base=%s' % url_base) 165 self.url = urlparse.urljoin(self.url, url_base) 166 167 if extra_args: 168 chromedriver_args.extend(extra_args) 169 170 # TODO(ihf): Remove references to X after M45. 171 # Chromedriver will look for an X server running on the display 172 # specified through the DISPLAY environment variable. 173 os.environ['DISPLAY'] = X_SERVER_DISPLAY 174 os.environ['XAUTHORITY'] = X_AUTHORITY 175 176 self.bg_job = utils.BgJob(chromedriver_args, stderr_level=logging.DEBUG) 177 if self.bg_job is None: 178 raise RuntimeError('ChromeDriver server cannot be started') 179 180 try: 181 timeout_msg = 'Timeout on waiting for ChromeDriver to start.' 182 utils.poll_for_condition(self.is_running, 183 exception=utils.TimeoutError(timeout_msg), 184 timeout=10, 185 sleep_interval=.1) 186 except utils.TimeoutError: 187 self.close_bgjob() 188 raise RuntimeError('ChromeDriver server did not start') 189 190 logging.debug('Chrome Driver server is up and listening at port %d.', 191 port) 192 if not skip_cleanup: 193 atexit.register(self.close) 194 195 196 def is_running(self): 197 """Returns whether the server is up and running.""" 198 try: 199 urllib2.urlopen(self.url + '/status') 200 return True 201 except urllib2.URLError as e: 202 return False 203 204 205 def close_bgjob(self): 206 """Close background job and log stdout and stderr.""" 207 utils.nuke_subprocess(self.bg_job.sp) 208 utils.join_bg_jobs([self.bg_job], timeout=1) 209 result = self.bg_job.result 210 if result.stdout or result.stderr: 211 logging.info('stdout of Chrome Driver:\n%s', result.stdout) 212 logging.error('stderr of Chrome Driver:\n%s', result.stderr) 213 214 215 def close(self): 216 """Kills the ChromeDriver server, if it is running.""" 217 if self.bg_job is None: 218 return 219 220 try: 221 urllib2.urlopen(self.url + '/shutdown', timeout=10).close() 222 except: 223 pass 224 225 self.close_bgjob() 226
