| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import logging |
| |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib.cros import chrome |
| |
| NETWORK_TEST_EXTENSION_PATH = ('/usr/local/autotest/cros/networking/' |
| 'chrome_testing/network_test_ext') |
| |
| class ChromeNetworkingTestContext(object): |
| """ |
| ChromeNetworkingTestContext handles creating a Chrome browser session and |
| launching a set of Chrome extensions on it. It provides handles for |
| telemetry extension objects, which can be used to inject JavaScript from |
| autotest. |
| |
| Apart from user provided extensions, ChromeNetworkingTestContext always |
| loads the default network testing extension 'network_test_ext' which |
| provides some boilerplate around chrome.networkingPrivate calls. |
| |
| Example usage: |
| |
| context = ChromeNetworkingTestContext() |
| context.setup() |
| extension = context.network_test_extension() |
| extension.EvaluateJavaScript('var foo = 1; return foo + 1;') |
| context.teardown() |
| |
| ChromeNetworkingTestContext also supports the Python 'with' syntax for |
| syntactic sugar. |
| |
| """ |
| |
| FIND_NETWORKS_TIMEOUT = 5 |
| |
| # Network type strings used by chrome.networkingPrivate |
| CHROME_NETWORK_TYPE_ETHERNET = 'Ethernet' |
| CHROME_NETWORK_TYPE_WIFI = 'WiFi' |
| CHROME_NETWORK_TYPE_BLUETOOTH = 'Bluetooth' |
| CHROME_NETWORK_TYPE_CELLULAR = 'Cellular' |
| CHROME_NETWORK_TYPE_VPN = 'VPN' |
| CHROME_NETWORK_TYPE_ALL = 'All' |
| |
| def __init__(self, extensions=None, username=None, password=None, |
| gaia_login=False): |
| if extensions is None: |
| extensions = [] |
| extensions.append(NETWORK_TEST_EXTENSION_PATH) |
| self._extension_paths = extensions |
| self._username = username |
| self._password = password |
| self._gaia_login = gaia_login |
| self._chrome = None |
| |
| def __enter__(self): |
| self.setup() |
| return self |
| |
| def __exit__(self, *args): |
| self.teardown() |
| |
| def _create_browser(self): |
| self._chrome = chrome.Chrome(logged_in=True, |
| gaia_login=self._gaia_login, |
| extension_paths=self._extension_paths, |
| username=self._username, |
| password=self._password) |
| |
| # TODO(armansito): This call won't be necessary once crbug.com/251913 |
| # gets fixed. |
| self._ensure_network_test_extension_is_ready() |
| |
| def _ensure_network_test_extension_is_ready(self): |
| self.network_test_extension.WaitForJavaScriptCondition( |
| "typeof chromeTesting != 'undefined'", timeout=30) |
| |
| def _get_extension(self, path): |
| if self._chrome is None: |
| raise error.TestFail('A browser session has not been setup.') |
| extension = self._chrome.get_extension(path) |
| if extension is None: |
| raise error.TestFail('Failed to find loaded extension "%s"' % path) |
| return extension |
| |
| def setup(self, browser=None): |
| """ |
| Initializes a ChromeOS browser session that loads the given extensions |
| with private API priviliges. |
| |
| @param browser: Chrome object to use, will create one if not provided. |
| |
| """ |
| logging.info('ChromeNetworkingTestContext: setup') |
| |
| if browser is None: |
| self._create_browser() |
| else: |
| self._chrome = browser |
| self.STATUS_PENDING = self.network_test_extension.EvaluateJavaScript( |
| 'chromeTesting.STATUS_PENDING') |
| self.STATUS_SUCCESS = self.network_test_extension.EvaluateJavaScript( |
| 'chromeTesting.STATUS_SUCCESS') |
| self.STATUS_FAILURE = self.network_test_extension.EvaluateJavaScript( |
| 'chromeTesting.STATUS_FAILURE') |
| |
| def teardown(self): |
| """ |
| Closes the browser session. |
| |
| """ |
| logging.info('ChromeNetworkingTestContext: teardown') |
| if self._chrome: |
| self._chrome.browser.Close() |
| self._chrome = None |
| |
| @property |
| def network_test_extension(self): |
| """ |
| @return Handle to the metworking test Chrome extension instance. |
| @raises error.TestFail if the browser has not been set up or if the |
| extension cannot get acquired. |
| |
| """ |
| return self._get_extension(NETWORK_TEST_EXTENSION_PATH) |
| |
| def call_test_function_async(self, function, *args): |
| """ |
| Asynchronously executes a JavaScript function that belongs to |
| "chromeTesting.networking" as defined in network_test_ext. The |
| return value (or call status) can be obtained at a later time via |
| "chromeTesting.networking.callStatus.<|function|>" |
| |
| @param function: The name of the function to execute. |
| @param args: The list of arguments that are to be passed to |function|. |
| Note that strings in JavaScript are quoted using double quotes, |
| and this function won't convert string arguments to JavaScript |
| strings. To pass a string, the string itself must contain the |
| quotes, i.e. '"string"', otherwise the contents of the Python |
| string will be compiled as a JS token. |
| @raises exceptions.EvaluateException, in case of an error during JS |
| execution. |
| |
| """ |
| arguments = ', '.join(str(i) for i in args) |
| extension = self.network_test_extension |
| extension.ExecuteJavaScript( |
| 'chromeTesting.networking.' + function + '(' + arguments + ');') |
| |
| def wait_for_condition_on_expression_result( |
| self, expression, condition, timeout): |
| """ |
| Blocks until |condition| returns True when applied to the result of the |
| JavaScript expression |expression|. |
| |
| @param expression: JavaScript expression to evaluate. |
| @param condition: A function that accepts a single argument and returns |
| a boolean. |
| @param timeout: The timeout interval length, in seconds, after which |
| this method will raise an error. |
| @raises error.TestFail, if the conditions is not met within the given |
| timeout interval. |
| |
| """ |
| extension = self.network_test_extension |
| def _evaluate_expr(): |
| return extension.EvaluateJavaScript(expression) |
| utils.poll_for_condition( |
| lambda: condition(_evaluate_expr()), |
| error.TestFail( |
| 'Timed out waiting for condition on expression: ' + |
| expression), |
| timeout) |
| return _evaluate_expr() |
| |
| def call_test_function(self, timeout, function, *args): |
| """ |
| Executes a JavaScript function that belongs to |
| "chromeTesting.networking" and blocks until the function has completed |
| its execution. A function is considered to have completed if the result |
| of "chromeTesting.networking.callStatus.<|function|>.status" equals |
| STATUS_SUCCESS or STATUS_FAILURE. |
| |
| @param timeout: The timeout interval, in seconds, for which this |
| function will block. If the call status is still STATUS_PENDING |
| after the timeout expires, then an error will be raised. |
| @param function: The name of the function to execute. |
| @param args: The list of arguments that are to be passed to |function|. |
| See the docstring for "call_test_function_async" for a more |
| detailed description. |
| @raises exceptions.EvaluateException, in case of an error during JS |
| execution. |
| @raises error.TestFail, if the function doesn't finish executing within |
| |timeout|. |
| |
| """ |
| self.call_test_function_async(function, *args) |
| return self.wait_for_condition_on_expression_result( |
| 'chromeTesting.networking.callStatus.' + function, |
| lambda x: (x is not None and |
| x['status'] != self.STATUS_PENDING), |
| timeout) |
| |
| def find_cellular_networks(self): |
| """ |
| Queries the current cellular networks. |
| |
| @return A list containing the found cellular networks. |
| |
| """ |
| return self.find_networks(self.CHROME_NETWORK_TYPE_CELLULAR) |
| |
| def find_wifi_networks(self): |
| """ |
| Queries the current wifi networks. |
| |
| @return A list containing the found wifi networks. |
| |
| """ |
| return self.find_networks(self.CHROME_NETWORK_TYPE_WIFI) |
| |
| def find_networks(self, network_type): |
| """ |
| Queries the current networks of the queried type. |
| |
| @param network_type: One of CHROME_NETWORK_TYPE_* strings. |
| |
| @return A list containing the found cellular networks. |
| |
| """ |
| call_status = self.call_test_function( |
| self.FIND_NETWORKS_TIMEOUT, |
| 'findNetworks', |
| '"' + network_type + '"') |
| if call_status['status'] == self.STATUS_FAILURE: |
| raise error.TestFail( |
| 'Failed to get networks: ' + call_status['error']) |
| networks = call_status['result'] |
| if type(networks) != list: |
| raise error.TestFail( |
| 'Expected a list, found "' + repr(networks) + '".') |
| return networks |
