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