1 # Copyright 2015 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 dbus 6 import logging 7 import tempfile 8 import time 9 10 from autotest_lib.client.bin import test 11 from autotest_lib.client.bin import utils 12 from autotest_lib.client.common_lib import error 13 from autotest_lib.client.common_lib.cros.network \ 14 import wifi_rack_constants as constants 15 from autotest_lib.client.cros.networking import wifi_proxy 16 from autotest_lib.client.cros.networking.chrome_testing \ 17 import chrome_networking_test_api as cnta 18 from autotest_lib.client.cros.networking.chrome_testing \ 19 import chrome_networking_test_context as cntc 20 21 22 class network_RackWiFiConnect(test.test): 23 """Client test to connect to various network services on WiFi rack. 24 25 After connection, we assert access to pages only accessible through the 26 connected network. 27 28 """ 29 version = 1 30 31 32 def _assert_access(self, test): 33 """Asset user can access page. 34 35 Verification URLs are either pages on WiFi rack's Apache server or 36 general Internet. 37 38 @param test string - testname of NetworkServices namedtuple 39 40 @return boolean - True if able to access, False otherwise 41 42 """ 43 for service_test in constants.NETWORK_SERVICES_TESTS: 44 if test == service_test.testname: 45 url, pattern = service_test.url, service_test.pattern 46 break 47 48 # Since this test runs OTA, allow 15 seconds of leeway 49 time.sleep(15) 50 51 wget_cmd = 'wget -O /tmp/wget.log %s' % url 52 for retry in range(3): 53 exit_status = utils.system(wget_cmd, ignore_status=True) 54 if not exit_status: 55 logging.debug('Able to wget URL.') 56 break 57 logging.error('Could not wget URL; trying again.') 58 59 grep_url_cmd = 'cat /tmp/wget.log | grep %s' % pattern 60 output_status = utils.system(grep_url_cmd, ignore_status=True) 61 if output_status: 62 logging.debug('Unable to access correct URL for %s', 63 service_test.testname) 64 return False 65 return True 66 67 68 def _connect(self, ssid, uname): 69 """Connect to particular network and assert access to page. 70 71 @param ssid string - predefined SSID from user's preferred networks 72 @param uname string - predefined username of managed user 73 74 @return boolean - True if able to connect, False otherwise 75 76 """ 77 start_time = time.time() 78 with cntc.ChromeNetworkingTestContext(username=uname, 79 password=constants.PASSWORD) as testing_context: 80 net_provider = cnta.ChromeNetworkProvider(testing_context) 81 enabled_devices = net_provider.get_enabled_devices() 82 if net_provider.WIFI_DEVICE not in enabled_devices: 83 net_provider.enable_network_device(net_provider.WIFI_DEVICE) 84 logging.info('Scanning for networks') 85 connect_to_service = None 86 while time.time() - start_time < constants.SCAN_RETRY_TIMEOUT: 87 net_provider.scan_for_networks(timeout=20) 88 logging.info('Attempting to connect to %s', ssid) 89 networks = net_provider.get_wifi_networks() 90 for service in networks: 91 if service['Name'] == ssid: 92 connect_to_service = service 93 if not connect_to_service: 94 logging.error('Unable to find %s', ssid) 95 continue 96 try: 97 net_provider.connect_to_network(connect_to_service) 98 logging.info('Successfully connected to network %s', ssid) 99 return True 100 except error.TestFail as e: 101 logging.error('Unable to connect to %s', ssid) 102 continue 103 return False 104 105 106 def _connect_and_assert(self, test, ssid, user): 107 """Verify connect and assert and write results to results/. 108 109 @param test string - testname of NetworkServices namedtuple 110 @param ssid string - predefined SSID from user's preferred networks 111 @param user string - predefined username of managed user 112 113 """ 114 tf = tempfile.NamedTemporaryFile(suffix='.txt', 115 prefix='connect_%s_' % test, 116 dir=self.resultsdir, 117 delete=False) 118 with tf as results: 119 if not self._connect(ssid, user): 120 results.write('%s FAILED to connect to SSID\n\n' % test) 121 elif not self._assert_access(test): 122 results.write('%s FAILED to access\n\n' % test) 123 else: 124 results.write('%s passed\n\n' % test) 125 126 127 def _to_wifi(self, proxy): 128 """Set service order to WiFi before Ethernet. 129 130 @param proxy WiFi Proxy object 131 132 """ 133 logging.info('Setting order to WiFi, prioritized over Ethernet.') 134 proxy.manager.SetServiceOrder(dbus.String('wifi,ethernet')) 135 136 137 def _to_ethernet(self, proxy): 138 """Set service order to default Ethernet before WiFi 139 140 @param proxy WiFi Proxy object 141 142 """ 143 logging.info('Setting back to default service order.') 144 proxy.manager.SetServiceOrder(dbus.String('ethernet,wifi')) 145 146 147 def run_once(self, test): 148 """Run the test. 149 150 @param test string - Set by the client test control file 151 152 """ 153 client_proxy = wifi_proxy.WifiProxy() 154 if test is not 'all': 155 logging.info('Running an individual control file.') 156 self._to_wifi(client_proxy) 157 for service_test in constants.NETWORK_SERVICES_TESTS: 158 if service_test.testname == test: 159 self._connect_and_assert(service_test.testname, 160 service_test.ssid, 161 service_test.user) 162 self._to_ethernet(client_proxy) 163 return 164 for service_test in constants.NETWORK_SERVICES_TESTS: 165 logging.info('==== Running current test %s ====', 166 service_test.testname) 167 self._to_wifi(client_proxy) 168 self._connect_and_assert(service_test.testname, 169 service_test.ssid, 170 service_test.user) 171 self._to_ethernet(client_proxy) 172 173 # Ensure DUT returns to normal service state 174 self._to_ethernet(client_proxy) 175
