1#!/usr/bin/env python3 2# 3# Copyright 2020 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import time 18 19from acts import signals 20from acts.controllers.access_point import setup_ap 21from acts.controllers.ap_lib import hostapd_constants 22from acts.controllers.ap_lib import hostapd_security 23from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest 24from acts.utils import rand_ascii_str 25 26DISCONNECTED = "Disconnected" 27CONNECTION_STOPPED = "ConnectionStopped" 28CONNECTIONS_ENABLED = "ConnectionsEnabled" 29CONNECTIONS_DISABLED = "ConnectionsDisabled" 30WPA2 = "wpa2" 31UPDATE_TIMEOUT_SEC = 5 32 33 34class StartStopClientConnectionsTest(WifiBaseTest): 35 """ Tests that we see the expected behavior with enabling and disabling 36 client connections 37 38 Test Bed Requirement: 39 * One or more Fuchsia devices 40 * One Access Point 41 """ 42 43 def setup_class(self): 44 super().setup_class() 45 # Start an AP with a hidden network 46 self.ssid = rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G) 47 self.access_point = self.access_points[0] 48 self.password = rand_ascii_str( 49 hostapd_constants.AP_PASSPHRASE_LENGTH_2G) 50 self.security_type = WPA2 51 security = hostapd_security.Security(security_mode=self.security_type, 52 password=self.password) 53 54 self.access_point.stop_all_aps() 55 # TODO(63719) use varying values for AP that shouldn't affect the test. 56 setup_ap(self.access_point, 57 'whirlwind', 58 hostapd_constants.AP_DEFAULT_CHANNEL_5G, 59 self.ssid, 60 security=security) 61 62 if len(self.fuchsia_devices) < 1: 63 raise EnvironmentError("No Fuchsia devices found.") 64 for fd in self.fuchsia_devices: 65 fd.configure_wlan(association_mechanism='policy', 66 preserve_saved_networks=True) 67 68 def setup_test(self): 69 for fd in self.fuchsia_devices: 70 if not fd.wlan_policy_controller.remove_all_networks(): 71 raise EnvironmentError( 72 "Failed to remove all networks in setup") 73 74 def teardown_class(self): 75 self.access_point.stop_all_aps() 76 77 def connect_and_validate(self, fd, ssid, security_type, expected_response): 78 """ Sends a connect request to the device and verifies we get a response 79 without error. This does not validate that a connection will be 80 attempted. This will fail the test if there is an error sending the 81 connect request, or if we don't get the expected connect response.""" 82 result_connect = fd.sl4f.wlan_policy_lib.wlanConnect( 83 ssid, security_type) 84 if result_connect.get("error") != None: 85 self.log.error("Error occurred requesting a connection: %s" % 86 result_connect.get("error")) 87 raise EnvironmentError("Failed to send connect request") 88 response = result_connect.get("result") 89 if response != expected_response: 90 self.log.error( 91 "Incorrect connect request response. Expected: \"%s\", Actual: %s" 92 % (expected_response, response)) 93 raise signals.TestFailure( 94 "Failed to get expected connect response") 95 96 def await_state_update(self, fd, desired_state, timeout): 97 """ This function polls the policy client state until it converges to 98 the caller's desired state. 99 100 Args: 101 fd: A FuchsiaDevice 102 desired_state: The expected client policy state. 103 timeout: Number of seconds to wait for the policy state to become 104 the desired_state. 105 Returns: 106 None assuming the desired state has been reached. 107 Raises: 108 TestFailure if the desired state is not reached by the timeout. 109 """ 110 start_time = time.time() 111 curr_state = None 112 while time.time() < start_time + timeout: 113 fd.sl4f.wlan_policy_lib.wlanSetNewListener() 114 curr_state = fd.sl4f.wlan_policy_lib.wlanGetUpdate() 115 if curr_state.get("error"): 116 self.log.error("Error occurred getting status update: %s" % 117 curr_state.get("error")) 118 raise EnvironmentError("Failed to get update") 119 120 if curr_state.get("result") and curr_state.get( 121 "result") == desired_state: 122 return 123 124 time.sleep(1) 125 126 self.log.error( 127 "Client state did not converge to the expected state in %s " 128 "seconds. Expected update: %s Actual update: %s" % 129 (timeout, desired_state, curr_state)) 130 raise signals.TestFailure("Client policy layer is in unexpected state") 131 132 def test_stop_client_connections_update(self): 133 for fd in self.fuchsia_devices: 134 if not fd.wlan_policy_controller.stop_client_connections(): 135 raise EnvironmentError("Failed to stop client connecions") 136 137 # Check that the most recent update says that the device is not 138 # connected to anything and client connections are disabled 139 expected_update = {"networks": [], "state": CONNECTIONS_DISABLED} 140 self.await_state_update(fd, expected_update, UPDATE_TIMEOUT_SEC) 141 142 def test_start_client_connections_update(self): 143 for fd in self.fuchsia_devices: 144 if not fd.wlan_policy_controller.start_client_connections(): 145 raise EnvironmentError("Failed to start client connecions") 146 147 # Check that the most recent update says that the device is not 148 # connected to anything and client connections are disabled 149 expected_update = {"networks": [], "state": CONNECTIONS_ENABLED} 150 self.await_state_update(fd, expected_update, UPDATE_TIMEOUT_SEC) 151 152 def test_stop_client_connections_rejects_connections(self): 153 # Test that if we turn client connections off, our requests to connect 154 # are rejected. 155 for fd in self.fuchsia_devices: 156 if not fd.wlan_policy_controller.stop_client_connections(): 157 raise EnvironmentError("Failed to stop client connecions") 158 159 # Save the network, otherwise connecting may fail because the 160 # network is not saved instead of client connections being off 161 if not fd.wlan_policy_controller.save_network( 162 self.ssid, self.security_type, password=self.password): 163 raise EnvironmentError("Failed to save network") 164 expected_response = "RejectedIncompatibleMode" 165 self.connect_and_validate(fd, self.ssid, self.security_type, 166 expected_response) 167 168 def test_start_stop_client_connections(self): 169 # Test that if we turn client connections on the device will connect, 170 # and if we turn of client connections the device will disconnect. 171 for fd in self.fuchsia_devices: 172 # Start client connections and check that we can 173 if not fd.wlan_policy_controller.save_network( 174 self.ssid, self.security_type, password=self.password): 175 raise EnvironmentError("Failed to save network") 176 if not fd.wlan_policy_controller.start_client_connections(): 177 raise EnvironmentError("Failed to start client connections") 178 179 expected_response = "Acknowledged" 180 self.connect_and_validate(fd, self.ssid, self.security_type, 181 expected_response) 182 183 if not fd.wlan_policy_controller.wait_for_connect( 184 self.ssid, self.security_type): 185 raise signals.TestFailure( 186 "Failed to connect after starting client connections") 187 188 # Stop client connections again and check that we disconnect 189 if not fd.wlan_policy_controller.stop_client_connections(): 190 raise EnvironmentError("Failed to stop client connecions") 191 if not fd.wlan_policy_controller.wait_for_disconnect( 192 self.ssid, self.security_type, DISCONNECTED, 193 CONNECTION_STOPPED): 194 raise signals.TestFailure( 195 "Failed to disconnect after client connections stopped") 196