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 5import logging 6 7from autotest_lib.client.bin import utils 8from autotest_lib.client.common_lib import error 9from autotest_lib.client.common_lib.cros import chrome 10 11NETWORK_TEST_EXTENSION_PATH = ('/usr/local/autotest/cros/networking/' 12 'chrome_testing/network_test_ext') 13 14class ChromeNetworkingTestContext(object): 15 """ 16 ChromeNetworkingTestContext handles creating a Chrome browser session and 17 launching a set of Chrome extensions on it. It provides handles for 18 telemetry extension objects, which can be used to inject JavaScript from 19 autotest. 20 21 Apart from user provided extensions, ChromeNetworkingTestContext always 22 loads the default network testing extension 'network_test_ext' which 23 provides some boilerplate around chrome.networkingPrivate calls. 24 25 Example usage: 26 27 context = ChromeNetworkingTestContext() 28 context.setup() 29 extension = context.network_test_extension() 30 extension.EvaluateJavaScript('var foo = 1; return foo + 1;') 31 context.teardown() 32 33 ChromeNetworkingTestContext also supports the Python 'with' syntax for 34 syntactic sugar. 35 36 """ 37 38 FIND_NETWORKS_TIMEOUT = 5 39 40 # Network type strings used by chrome.networkingPrivate 41 CHROME_NETWORK_TYPE_ETHERNET = 'Ethernet' 42 CHROME_NETWORK_TYPE_WIFI = 'WiFi' 43 CHROME_NETWORK_TYPE_BLUETOOTH = 'Bluetooth' 44 CHROME_NETWORK_TYPE_CELLULAR = 'Cellular' 45 CHROME_NETWORK_TYPE_VPN = 'VPN' 46 CHROME_NETWORK_TYPE_ALL = 'All' 47 48 def __init__(self, extensions=None, username=None, password=None, 49 gaia_login=False): 50 if extensions is None: 51 extensions = [] 52 extensions.append(NETWORK_TEST_EXTENSION_PATH) 53 self._extension_paths = extensions 54 self._username = username 55 self._password = password 56 self._gaia_login = gaia_login 57 self._chrome = None 58 59 def __enter__(self): 60 self.setup() 61 return self 62 63 def __exit__(self, *args): 64 self.teardown() 65 66 def _create_browser(self): 67 self._chrome = chrome.Chrome(logged_in=True, 68 gaia_login=self._gaia_login, 69 extension_paths=self._extension_paths, 70 username=self._username, 71 password=self._password) 72 73 # TODO(armansito): This call won't be necessary once crbug.com/251913 74 # gets fixed. 75 self._ensure_network_test_extension_is_ready() 76 77 def _ensure_network_test_extension_is_ready(self): 78 self.network_test_extension.WaitForJavaScriptCondition( 79 "typeof chromeTesting != 'undefined'", timeout=30) 80 81 def _get_extension(self, path): 82 if self._chrome is None: 83 raise error.TestFail('A browser session has not been setup.') 84 extension = self._chrome.get_extension(path) 85 if extension is None: 86 raise error.TestFail('Failed to find loaded extension "%s"' % path) 87 return extension 88 89 def setup(self, browser=None): 90 """ 91 Initializes a ChromeOS browser session that loads the given extensions 92 with private API priviliges. 93 94 @param browser: Chrome object to use, will create one if not provided. 95 96 """ 97 logging.info('ChromeNetworkingTestContext: setup') 98 99 if browser is None: 100 self._create_browser() 101 else: 102 self._chrome = browser 103 self.STATUS_PENDING = self.network_test_extension.EvaluateJavaScript( 104 'chromeTesting.STATUS_PENDING') 105 self.STATUS_SUCCESS = self.network_test_extension.EvaluateJavaScript( 106 'chromeTesting.STATUS_SUCCESS') 107 self.STATUS_FAILURE = self.network_test_extension.EvaluateJavaScript( 108 'chromeTesting.STATUS_FAILURE') 109 110 def teardown(self): 111 """ 112 Closes the browser session. 113 114 """ 115 logging.info('ChromeNetworkingTestContext: teardown') 116 if self._chrome: 117 self._chrome.browser.Close() 118 self._chrome = None 119 120 @property 121 def network_test_extension(self): 122 """ 123 @return Handle to the metworking test Chrome extension instance. 124 @raises error.TestFail if the browser has not been set up or if the 125 extension cannot get acquired. 126 127 """ 128 return self._get_extension(NETWORK_TEST_EXTENSION_PATH) 129 130 def call_test_function_async(self, function, *args): 131 """ 132 Asynchronously executes a JavaScript function that belongs to 133 "chromeTesting.networking" as defined in network_test_ext. The 134 return value (or call status) can be obtained at a later time via 135 "chromeTesting.networking.callStatus.<|function|>" 136 137 @param function: The name of the function to execute. 138 @param args: The list of arguments that are to be passed to |function|. 139 Note that strings in JavaScript are quoted using double quotes, 140 and this function won't convert string arguments to JavaScript 141 strings. To pass a string, the string itself must contain the 142 quotes, i.e. '"string"', otherwise the contents of the Python 143 string will be compiled as a JS token. 144 @raises exceptions.EvaluateException, in case of an error during JS 145 execution. 146 147 """ 148 arguments = ', '.join(str(i) for i in args) 149 extension = self.network_test_extension 150 extension.ExecuteJavaScript( 151 'chromeTesting.networking.' + function + '(' + arguments + ');') 152 153 def wait_for_condition_on_expression_result( 154 self, expression, condition, timeout): 155 """ 156 Blocks until |condition| returns True when applied to the result of the 157 JavaScript expression |expression|. 158 159 @param expression: JavaScript expression to evaluate. 160 @param condition: A function that accepts a single argument and returns 161 a boolean. 162 @param timeout: The timeout interval length, in seconds, after which 163 this method will raise an error. 164 @raises error.TestFail, if the conditions is not met within the given 165 timeout interval. 166 167 """ 168 extension = self.network_test_extension 169 def _evaluate_expr(): 170 return extension.EvaluateJavaScript(expression) 171 utils.poll_for_condition( 172 lambda: condition(_evaluate_expr()), 173 error.TestFail( 174 'Timed out waiting for condition on expression: ' + 175 expression), 176 timeout) 177 return _evaluate_expr() 178 179 def call_test_function(self, timeout, function, *args): 180 """ 181 Executes a JavaScript function that belongs to 182 "chromeTesting.networking" and blocks until the function has completed 183 its execution. A function is considered to have completed if the result 184 of "chromeTesting.networking.callStatus.<|function|>.status" equals 185 STATUS_SUCCESS or STATUS_FAILURE. 186 187 @param timeout: The timeout interval, in seconds, for which this 188 function will block. If the call status is still STATUS_PENDING 189 after the timeout expires, then an error will be raised. 190 @param function: The name of the function to execute. 191 @param args: The list of arguments that are to be passed to |function|. 192 See the docstring for "call_test_function_async" for a more 193 detailed description. 194 @raises exceptions.EvaluateException, in case of an error during JS 195 execution. 196 @raises error.TestFail, if the function doesn't finish executing within 197 |timeout|. 198 199 """ 200 self.call_test_function_async(function, *args) 201 return self.wait_for_condition_on_expression_result( 202 'chromeTesting.networking.callStatus.' + function, 203 lambda x: (x is not None and 204 x['status'] != self.STATUS_PENDING), 205 timeout) 206 207 def find_cellular_networks(self): 208 """ 209 Queries the current cellular networks. 210 211 @return A list containing the found cellular networks. 212 213 """ 214 return self.find_networks(self.CHROME_NETWORK_TYPE_CELLULAR) 215 216 def find_wifi_networks(self): 217 """ 218 Queries the current wifi networks. 219 220 @return A list containing the found wifi networks. 221 222 """ 223 return self.find_networks(self.CHROME_NETWORK_TYPE_WIFI) 224 225 def find_networks(self, network_type): 226 """ 227 Queries the current networks of the queried type. 228 229 @param network_type: One of CHROME_NETWORK_TYPE_* strings. 230 231 @return A list containing the found cellular networks. 232 233 """ 234 call_status = self.call_test_function( 235 self.FIND_NETWORKS_TIMEOUT, 236 'findNetworks', 237 '"' + network_type + '"') 238 if call_status['status'] == self.STATUS_FAILURE: 239 raise error.TestFail( 240 'Failed to get networks: ' + call_status['error']) 241 networks = call_status['result'] 242 if type(networks) != list: 243 raise error.TestFail( 244 'Expected a list, found "' + repr(networks) + '".') 245 return networks 246