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
6import time
7import urlparse
8
9from autotest_lib.client.bin import test
10from autotest_lib.client.common_lib import error
11from autotest_lib.client.cros import backchannel, network
12from autotest_lib.client.cros.cellular import cell_tools
13from autotest_lib.client.cros.networking import shill_context
14
15# Import 'flimflam_test_path' first in order to import 'flimflam'.
16# pylint: disable=W0611
17from autotest_lib.client.cros import flimflam_test_path
18# pylint: enable=W0611
19import flimflam
20
21
22# TODO(benchan): Use the log scopes defined in ShillProxy after
23# migrating from FlimFlam to ShillProxy.
24SHILL_LOG_SCOPES = 'dbus+device+dhcp+manager+portal+service+wimax'
25
26
27class network_WiMaxSmoke(test.test):
28    """Verifies that a WiMAX module can connect to a WiMAX network.
29
30    The test attempts to connect to a WiMAX network. It assumes that a WiMAX
31    module is plugged in to the DUT and a WiMAX network is available. It repeats
32    the connect/disconnect sequence several times. Each time after connecting to
33    the WiMAX network, it fetches some data from a URL to confirm the WiMAX
34    connectivity.
35
36    """
37    version = 1
38
39    def _connect_service(self):
40        """Connects the WiMAX service under test.
41
42        Raises:
43            error.TestFail if it fails to connect the service before timeout.
44
45        """
46        logging.info('ConnectService: %s', self._service.object_path)
47
48        service_properties = self._service.GetProperties()
49        parameters = {
50            'Type': 'wimax',
51            'Name': str(service_properties['Name']),
52            'NetworkId': str(service_properties['NetworkId']),
53            'EAP.Identity': 'test',
54            'EAP.Password': 'test',
55        }
56        logging.info('parameters : %s', parameters)
57        self._flim.manager.ConfigureService(parameters)
58
59        success, status = self._flim.ConnectService(
60            service=self._service,
61            config_timeout=self._connect_timeout)
62        if not success:
63            raise error.TestFail('Could not connect: %s.' % status)
64
65        logging.info('Waiting for portal or online state.')
66        portal_or_online_states = ['portal', 'online']
67        state = self._flim.WaitForServiceState(
68            service=self._service,
69            expected_states=portal_or_online_states,
70            timeout=self._connect_timeout,
71            ignore_failure=True)[0]
72        if not state in portal_or_online_states:
73            raise error.TestFail('Still in state %s' % state)
74
75
76    def _disconnect_service(self):
77        """Disconnects the WiMAX service under test.
78
79        Raises:
80            error.TestFail if it fails to disconnect the service before
81                timeout.
82
83        """
84        logging.info('DisonnectService: %s', self._service.object_path)
85
86        success, status = self._flim.DisconnectService(
87            service=self._service,
88            wait_timeout=self._disconnect_timeout)
89        if not success:
90            raise error.TestFail('Could not disconnect: %s.' % status)
91
92
93    def _test_connectivity(self):
94        """Tests network connectivity over WiMAX.
95
96        Test network connectivity over WiMAX as follows:
97            - Connecting the WiMAX service
98            - Fetching data from a URL
99            - Disconnecting the WiMAX service
100
101        Raises:
102            error.TestFail if any of the steps above fails.
103
104        """
105        if self._sleep_kludge:
106            logging.info('Sleeping for %.1f seconds', self._sleep_kludge)
107            time.sleep(self._sleep_kludge)
108
109        self._connect_service()
110
111        device = self._flim.GetObjectInterface(
112            'Device', self._service.GetProperties()['Device'])
113        interface = device.GetProperties()['Interface']
114        logging.info('Expected interface for %s: %s',
115                     self._service.object_path, interface)
116        network.CheckInterfaceForDestination(
117            urlparse.urlparse(self._fetch_url_pattern).hostname,
118            interface)
119
120        fetch_time = network.FetchUrl(self._fetch_url_pattern,
121                                      self._bytes_to_fetch,
122                                      self._fetch_timeout)
123        self.write_perf_keyval({
124            'seconds_WiMAX_fetch_time': fetch_time,
125            'bytes_WiMAX_bytes_received': self._bytes_to_fetch,
126            'bits_second_WiMAX_speed': 8 * self._bytes_to_fetch / fetch_time
127        })
128
129        self._disconnect_service()
130
131
132    def run_once(self, **kwargs):
133        # Number of connectivity test runs.
134        self._connect_count = kwargs.get('connect_count', 5)
135
136        # Number of seconds to sleep between connect and disconnect operations.
137        self._sleep_kludge = kwargs.get('sleep_kludge', 5)
138
139        # URL pattern to fetch data from during each connectivity test run.
140        self._fetch_url_pattern = \
141            kwargs.get('fetch_url_pattern', network.FETCH_URL_PATTERN_FOR_TEST)
142
143        # Timeout in seconds for connect and disconnect operations, and
144        # fetching data from a URL.
145        self._connect_timeout = kwargs.get('connect_timeout', 10)
146        self._disconnect_timeout = kwargs.get('disconnect_timeout', 10)
147        self._fetch_timeout = kwargs.get('fetch_timeout', 120)
148
149        # Number of bytes to fetch during each connectivity test run.
150        self._bytes_to_fetch = kwargs.get('bytes_to_fetch', 64 * 1024)
151
152        with backchannel.Backchannel():
153            with cell_tools.OtherDeviceShutdownContext('wimax'):
154                # TODO(benchan): Replace FlimFlam with ShillProxy.
155                self._flim = flimflam.FlimFlam()
156                self._flim.SetDebugTags(SHILL_LOG_SCOPES)
157                self._service = self._flim.FindWimaxService()
158                if not self._service:
159                    raise error.TestError('Could not find a WiMAX service.')
160
161                with shill_context.ServiceAutoConnectContext(
162                    self._flim.FindWimaxService, False):
163                    self._disconnect_service()
164                    for _ in xrange(self._connect_count):
165                        self._test_connectivity()
166