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