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