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