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 5 import logging 6 import time 7 import urlparse 8 9 from autotest_lib.client.bin import test 10 from autotest_lib.client.common_lib import error 11 from autotest_lib.client.cros import backchannel, network 12 from autotest_lib.client.cros.cellular import cell_tools 13 from autotest_lib.client.cros.networking import shill_context 14 15 # Import 'flimflam_test_path' first in order to import 'flimflam'. 16 # pylint: disable=W0611 17 from autotest_lib.client.cros import flimflam_test_path 18 # pylint: enable=W0611 19 import flimflam 20 21 22 # TODO(benchan): Use the log scopes defined in ShillProxy after 23 # migrating from FlimFlam to ShillProxy. 24 SHILL_LOG_SCOPES = 'dbus+device+dhcp+manager+portal+service+wimax' 25 26 27 class 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
