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 mobly import signals 18import multiprocessing as mp 19import random 20import time 21 22from acts import utils 23from acts import asserts 24from acts.base_test import BaseTestClass 25from acts.controllers import iperf_server 26from acts.controllers import iperf_client 27from acts.controllers.access_point import setup_ap 28from acts.controllers.ap_lib import hostapd_constants 29from acts.controllers.ap_lib import hostapd_security 30from acts.controllers.ap_lib.hostapd_utils import generate_random_password 31from acts_contrib.test_utils.abstract_devices.wlan_device import create_wlan_device 32 33ANDROID_DEFAULT_WLAN_INTERFACE = 'wlan0' 34CONNECTIVITY_MODE_LOCAL = 'local_only' 35CONNECTIVITY_MODE_UNRESTRICTED = 'unrestricted' 36DEFAULT_AP_PROFILE = 'whirlwind' 37DEFAULT_IPERF_PORT = 5201 38DEFAULT_STRESS_TEST_ITERATIONS = 10 39DEFAULT_TIMEOUT = 30 40DEFAULT_IPERF_TIMEOUT = 60 41DEFAULT_NO_ADDR_EXPECTED_TIMEOUT = 5 42INTERFACE_ROLE_AP = 'Ap' 43INTERFACE_ROLE_CLIENT = 'Client' 44INTERFACE_ROLES = {INTERFACE_ROLE_AP, INTERFACE_ROLE_CLIENT} 45OPERATING_BAND_2G = 'only_2_4_ghz' 46OPERATING_BAND_5G = 'only_5_ghz' 47OPERATING_BAND_ANY = 'any' 48SECURITY_OPEN = 'none' 49SECURITY_WEP = 'wep' 50SECURITY_WPA = 'wpa' 51SECURITY_WPA2 = 'wpa2' 52SECURITY_WPA3 = 'wpa3' 53STATE_UP = True 54STATE_DOWN = False 55TEST_TYPE_ASSOCIATE_ONLY = 'associate_only' 56TEST_TYPE_ASSOCIATE_AND_PING = 'associate_and_ping' 57TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC = 'associate_and_pass_traffic' 58TEST_TYPES = { 59 TEST_TYPE_ASSOCIATE_ONLY, TEST_TYPE_ASSOCIATE_AND_PING, 60 TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC 61} 62 63 64def get_test_name_from_settings(settings): 65 return settings['test_name'] 66 67 68def get_ap_params_from_config_or_default(config): 69 """Retrieves AP parameters from ACTS config, or returns default settings. 70 71 Args: 72 config: dict, from ACTS config, that may contain custom ap parameters 73 74 Returns: 75 dict, containing all AP parameters 76 """ 77 profile = config.get('profile', DEFAULT_AP_PROFILE) 78 ssid = config.get( 79 'ssid', utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G)) 80 channel = config.get('channel', hostapd_constants.AP_DEFAULT_CHANNEL_2G) 81 security_mode = config.get('security_mode', None) 82 password = config.get('password', None) 83 if security_mode: 84 if not password: 85 password = generate_random_password(security_mode=security_mode) 86 security = hostapd_security.Security(security_mode, password) 87 else: 88 security = None 89 90 return { 91 'profile': profile, 92 'ssid': ssid, 93 'channel': channel, 94 'security': security, 95 'password': password 96 } 97 98 99def get_soft_ap_params_from_config_or_default(config): 100 """Retrieves SoftAp parameters from ACTS config or returns default settings. 101 102 Args: 103 config: dict, from ACTS config, that may contain custom soft ap 104 parameters 105 106 Returns: 107 dict, containing all soft AP parameters 108 """ 109 ssid = config.get( 110 'ssid', utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G)) 111 connectivity_mode = config.get('connectivity_mode', 112 CONNECTIVITY_MODE_LOCAL) 113 operating_band = config.get('operating_band', OPERATING_BAND_2G) 114 security_type = config.get('security_type', SECURITY_OPEN) 115 password = config.get('password', '') 116 117 # The SoftAP API uses 'open' security instead of None, '' password 118 # instead of None, and security_type instead of security_mode, hence 119 # the difference between ap_params and soft_ap_params 120 if security_type != SECURITY_OPEN and password == '': 121 password = generate_random_password(security_mode=security_type) 122 123 return { 124 'ssid': ssid, 125 'connectivity_mode': connectivity_mode, 126 'operating_band': operating_band, 127 'security_type': security_type, 128 'password': password 129 } 130 131 132class StressTestIterationFailure(Exception): 133 """Used to differentiate a subtest failure from an actual exception""" 134 pass 135 136 137class SoftApClient(object): 138 def __init__(self, device): 139 self.w_device = create_wlan_device(device) 140 self.ip_client = iperf_client.IPerfClientOverAdb(device.serial) 141 142 143class WlanInterface(object): 144 def __init__(self): 145 self.name = None 146 self.mac_addr = None 147 self.ipv4 = None 148 149 150class SoftApTest(BaseTestClass): 151 """Tests for Fuchsia SoftAP 152 153 Testbed requirement: 154 * One Fuchsia device 155 * At least one dlient (Android) device 156 * For multi-client tests, at least two dlient (Android) devices are 157 required. Test will be skipped if less than two client devices are 158 present. 159 * For any tests that exercise client-mode (e.g. toggle tests, simultaneous 160 tests), a physical AP (whirlwind) is also required. Those tests will be 161 skipped if physical AP is not present. 162 """ 163 def setup_class(self): 164 self.soft_ap_test_params = self.user_params.get( 165 'soft_ap_test_params', {}) 166 self.dut = create_wlan_device(self.fuchsia_devices[0]) 167 self.dut.device.netstack_lib.init() 168 169 # TODO(fxb/51313): Add in device agnosticity for clients 170 self.clients = [] 171 for device in self.android_devices: 172 self.clients.append(SoftApClient(device)) 173 self.primary_client = self.clients[0] 174 175 self.iperf_server_config = { 176 'user': self.dut.device.ssh_username, 177 'host': self.dut.device.ip, 178 'ssh_config': self.dut.device.ssh_config 179 } 180 self.iperf_server = iperf_server.IPerfServerOverSsh( 181 self.iperf_server_config, DEFAULT_IPERF_PORT, use_killall=True) 182 self.iperf_server.start() 183 184 try: 185 self.access_point = self.access_points[0] 186 except AttributeError: 187 self.access_point = None 188 189 self.ap_iperf_client = iperf_client.IPerfClientOverSsh( 190 self.user_params['AccessPoint'][0]['ssh_config']) 191 192 def teardown_class(self): 193 # Because this is using killall, it will stop all iperf processes 194 self.iperf_server.stop() 195 196 def setup_test(self): 197 for ad in self.android_devices: 198 ad.droid.wakeLockAcquireBright() 199 ad.droid.wakeUpNow() 200 for client in self.clients: 201 client.w_device.disconnect() 202 client.w_device.reset_wifi() 203 client.w_device.wifi_toggle_state(True) 204 self.stop_all_soft_aps() 205 if self.access_point: 206 self.access_point.stop_all_aps() 207 self.dut.disconnect() 208 209 def teardown_test(self): 210 for client in self.clients: 211 client.w_device.disconnect() 212 for ad in self.android_devices: 213 ad.droid.wakeLockRelease() 214 ad.droid.goToSleepNow() 215 self.stop_all_soft_aps() 216 if self.access_point: 217 self.access_point.stop_all_aps() 218 self.dut.disconnect() 219 220 def on_fail(self, test_name, begin_time): 221 self.dut.take_bug_report(test_name, begin_time) 222 self.dut.get_log(test_name, begin_time) 223 224 def start_soft_ap(self, settings): 225 """Starts a softAP on Fuchsia device. 226 227 Args: 228 settings: a dict containing softAP configuration params 229 ssid: string, SSID of softAP network 230 security_type: string, security type of softAP network 231 - 'none', 'wep', 'wpa', 'wpa2', 'wpa3' 232 password: string, password if applicable 233 connectivity_mode: string, connecitivity_mode for softAP 234 - 'local_only', 'unrestricted' 235 operating_band: string, band for softAP network 236 - 'any', 'only_5_ghz', 'only_2_4_ghz' 237 """ 238 ssid = settings['ssid'] 239 security_type = settings['security_type'] 240 password = settings.get('password', '') 241 connectivity_mode = settings['connectivity_mode'] 242 operating_band = settings['operating_band'] 243 244 self.log.info('Starting SoftAP on DUT with settings: %s' % settings) 245 246 response = self.dut.device.wlan_ap_policy_lib.wlanStartAccessPoint( 247 ssid, security_type, password, connectivity_mode, operating_band) 248 if response.get('error'): 249 raise EnvironmentError('SL4F: Failed to setup SoftAP. Err: %s' % 250 response['error']) 251 252 self.log.info('SoftAp network (%s) is up.' % ssid) 253 254 def stop_soft_ap(self, settings): 255 """ Stops a specific SoftAP On Fuchsia device. 256 257 Args: 258 settings: a dict containing softAP config params (see start_soft_ap) 259 for details 260 261 Raises: 262 EnvironmentError, if StopSoftAP call fails. 263 """ 264 ssid = settings['ssid'] 265 security_type = settings['security_type'] 266 password = settings.get('password', '') 267 268 response = self.dut.device.wlan_ap_policy_lib.wlanStopAccessPoint( 269 ssid, security_type, password) 270 if response.get('error'): 271 raise EnvironmentError('SL4F: Failed to stop SoftAP. Err: %s' % 272 response['error']) 273 274 def stop_all_soft_aps(self): 275 """ Stops all SoftAPs on Fuchsia Device. 276 277 Raises: 278 EnvironmentError, if StopAllAps call fails. 279 """ 280 response = self.dut.device.wlan_ap_policy_lib.wlanStopAllAccessPoint() 281 if response.get('error'): 282 raise EnvironmentError( 283 'SL4F: Failed to stop all SoftAPs. Err: %s' % 284 response['error']) 285 286 def associate_with_soft_ap(self, w_device, settings): 287 """Associates client device with softAP on Fuchsia device. 288 289 Args: 290 w_device: wlan_device to associate with the softAP 291 settings: a dict containing softAP config params (see start_soft_ap) 292 for details 293 294 Raises: 295 TestFailure, if association fails 296 """ 297 self.log.info( 298 'Attempting to associate client %s with SoftAP on FuchsiaDevice ' 299 '(%s).' % (w_device.device.serial, self.dut.device.ip)) 300 301 check_connectivity = settings[ 302 'connectivity_mode'] == CONNECTIVITY_MODE_UNRESTRICTED 303 associated = w_device.associate( 304 settings['ssid'], 305 target_pwd=settings.get('password'), 306 target_security=hostapd_constants. 307 SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get( 308 settings['security_type'], None), 309 check_connectivity=check_connectivity) 310 311 if not associated: 312 self.log.error('Failed to connect to SoftAp.') 313 return False 314 315 self.log.info('Client successfully associated with SoftAP.') 316 return True 317 318 def disconnect_from_soft_ap(self, w_device): 319 """Disconnects client device from SoftAP. 320 321 Args: 322 w_device: wlan_device to disconnect from SoftAP 323 """ 324 self.log.info('Disconnecting device %s from SoftAP.' % 325 w_device.device.serial) 326 w_device.disconnect() 327 328 def get_dut_interface_by_role(self, 329 role, 330 wait_for_addr_timeout=DEFAULT_TIMEOUT): 331 """Retrieves interface information from the FuchsiaDevice DUT based 332 on the role. 333 334 Args: 335 role: string, the role of the interface to seek (e.g. Client or Ap) 336 337 Raises: 338 ConnectionError, if SL4F calls fail 339 AttributeError, if device does not have an interface matching role 340 341 Returns: 342 WlanInterface object representing the interface matching role 343 """ 344 if not role in INTERFACE_ROLES: 345 raise ValueError('Unsupported interface role %s' % role) 346 347 interface = WlanInterface() 348 349 # Determine WLAN interface with role 350 wlan_ifaces = self.dut.device.wlan_lib.wlanGetIfaceIdList() 351 if wlan_ifaces.get('error'): 352 raise ConnectionError('Failed to get wlan interface IDs: %s' % 353 wlan_ifaces['error']) 354 355 for wlan_iface in wlan_ifaces['result']: 356 iface_info = self.dut.device.wlan_lib.wlanQueryInterface( 357 wlan_iface) 358 if iface_info.get('error'): 359 raise ConnectionError('Failed to query wlan iface: %s' % 360 iface_info['error']) 361 362 if iface_info['result']['role'] == role: 363 interface.mac_addr = iface_info['result']['mac_addr'] 364 break 365 else: 366 raise LookupError('Failed to find a %s interface.' % role) 367 368 # Retrieve interface info from netstack 369 netstack_ifaces = self.dut.device.netstack_lib.netstackListInterfaces() 370 if netstack_ifaces.get('error'): 371 raise ConnectionError('Failed to get netstack ifaces: %s' % 372 netstack_ifaces['error']) 373 374 # TODO(fxb/51315): Once subnet information is available in 375 # netstackListInterfaces store it to verify the clients ip address. 376 for netstack_iface in netstack_ifaces['result']: 377 if netstack_iface['mac'] == interface.mac_addr: 378 interface.name = netstack_iface['name'] 379 if len(netstack_iface['ipv4_addresses']) > 0: 380 interface.ipv4 = '.'.join( 381 str(byte) 382 for byte in netstack_iface['ipv4_addresses'][0]) 383 else: 384 interface.ipv4 = self.wait_for_ipv4_address( 385 self.dut, 386 interface.name, 387 timeout=wait_for_addr_timeout) 388 self.log.info('DUT %s interface: %s. Has ipv4 address %s' % 389 (role, interface.name, interface.ipv4)) 390 return interface 391 392 def wait_for_ipv4_address(self, 393 w_device, 394 interface_name, 395 timeout=DEFAULT_TIMEOUT): 396 # TODO(fxb/51315): Once subnet information is available in netstack, add a 397 # subnet verification here. 398 """ Waits for interface on a wlan_device to get an ipv4 address. 399 400 Args: 401 w_device: wlan_device to check interface 402 interface_name: name of the interface to check 403 timeout: seconds to wait before raising an error 404 405 Raises: 406 ValueError, if interface does not have an ipv4 address after timeout 407 """ 408 409 end_time = time.time() + timeout 410 while time.time() < end_time: 411 ips = w_device.get_interface_ip_addresses(interface_name) 412 if len(ips['ipv4_private']) > 0: 413 self.log.info('Device %s interface %s has ipv4 address %s' % 414 (w_device.device.serial, interface_name, 415 ips['ipv4_private'][0])) 416 return ips['ipv4_private'][0] 417 else: 418 time.sleep(1) 419 raise ConnectionError( 420 'After %s seconds, device %s still does not have an ipv4 address ' 421 'on interface %s.' % 422 (timeout, w_device.device.serial, interface_name)) 423 424 def get_ap_ipv4_address(self, channel, timeout=DEFAULT_TIMEOUT): 425 """Get APs ipv4 address (actual AP, not soft ap on DUT) 426 427 Args: 428 channel: int, channel of the network used to determine 429 which interface to use 430 """ 431 lowest_5ghz_channel = 36 432 if channel < lowest_5ghz_channel: 433 ap_interface = self.access_point.wlan_2g 434 else: 435 ap_interface = self.access_point.wlan_5g 436 end_time = time.time() + timeout 437 while time.time() < end_time: 438 ap_ipv4_addresses = utils.get_interface_ip_addresses( 439 self.access_point.ssh, ap_interface)['ipv4_private'] 440 if len(ap_ipv4_addresses) > 0: 441 return ap_ipv4_addresses[0] 442 else: 443 self.log.debug( 444 'Access point does not have an ipv4 address on interface ' 445 '%s. Retrying in 1 second.' % ap_interface) 446 else: 447 raise ConnectionError( 448 'Access point never had an ipv4 address on interface %s.' % 449 ap_interface) 450 451 def device_can_ping_addr(self, w_device, dest_ip, timeout=DEFAULT_TIMEOUT): 452 """ Verify wlan_device can ping a destination ip. 453 454 Args: 455 w_device: wlan_device to initiate ping 456 dest_ip: ip to ping from wlan_device 457 458 Raises: 459 TestFailure, if ping fails 460 """ 461 end_time = time.time() + timeout 462 while time.time() < end_time: 463 with utils.SuppressLogOutput(): 464 ping_result = w_device.can_ping(dest_ip) 465 466 if ping_result: 467 self.log.info('Ping successful from device %s to dest ip %s.' % 468 (w_device.identifier, dest_ip)) 469 return True 470 else: 471 self.log.debug( 472 'Device %s could not ping dest ip %s. Retrying in 1 second.' 473 % (w_device.identifier, dest_ip)) 474 time.sleep(1) 475 else: 476 self.log.info('Failed to ping from device %s to dest ip %s.' % 477 (w_device.identifier, dest_ip)) 478 return False 479 480 def run_iperf_traffic(self, ip_client, server_address, server_port=5201): 481 """Runs traffic between client and ap an verifies throughput. 482 483 Args: 484 ip_client: iperf client to use 485 server_address: ipv4 address of the iperf server to use 486 server_port: port of the iperf server 487 488 Raises: 489 TestFailure, if no traffic passes in either direction 490 """ 491 ip_client_identifier = self.get_iperf_client_identifier(ip_client) 492 493 self.log.info( 494 'Running traffic from iperf client %s to iperf server %s.' % 495 (ip_client_identifier, server_address)) 496 client_to_ap_path = ip_client.start( 497 server_address, '-i 1 -t 10 -J -p %s' % server_port, 498 'client_to_soft_ap') 499 500 client_to_ap_result = iperf_server.IPerfResult(client_to_ap_path) 501 if (not client_to_ap_result.avg_receive_rate): 502 raise ConnectionError( 503 'Failed to pass traffic from iperf client %s to iperf server %s.' 504 % (ip_client_identifier, server_address)) 505 506 self.log.info( 507 'Passed traffic from iperf client %s to iperf server %s with avg ' 508 'rate of %s MB/s.' % (ip_client_identifier, server_address, 509 client_to_ap_result.avg_receive_rate)) 510 511 self.log.info( 512 'Running traffic from iperf server %s to iperf client %s.' % 513 (server_address, ip_client_identifier)) 514 ap_to_client_path = ip_client.start( 515 server_address, '-i 1 -t 10 -R -J -p %s' % server_port, 516 'soft_ap_to_client') 517 518 ap_to_client_result = iperf_server.IPerfResult(ap_to_client_path) 519 if (not ap_to_client_result.avg_receive_rate): 520 raise ConnectionError( 521 'Failed to pass traffic from iperf server %s to iperf client %s.' 522 % (server_address, ip_client_identifier)) 523 524 self.log.info( 525 'Passed traffic from iperf server %s to iperf client %s with avg ' 526 'rate of %s MB/s.' % (server_address, ip_client_identifier, 527 ap_to_client_result.avg_receive_rate)) 528 529 def run_iperf_traffic_parallel_process(self, 530 ip_client, 531 server_address, 532 error_queue, 533 server_port=5201): 534 """ Executes run_iperf_traffic using a queue to capture errors. Used 535 when running iperf in a parallel process. 536 537 Args: 538 ip_client: iperf client to use 539 server_address: ipv4 address of the iperf server to use 540 error_queue: multiprocessing queue to capture errors 541 server_port: port of the iperf server 542 """ 543 try: 544 self.run_iperf_traffic(ip_client, 545 server_address, 546 server_port=server_port) 547 except ConnectionError as err: 548 error_queue.put('In iperf process from %s to %s: %s' % 549 (self.get_iperf_client_identifier(ip_client), 550 server_address, err)) 551 552 def get_iperf_client_identifier(self, ip_client): 553 """ Retrieves an indentifer string from iperf client, for logging. 554 555 Args: 556 ip_client: iperf client to grab identifier from 557 """ 558 if type(ip_client) == iperf_client.IPerfClientOverAdb: 559 return ip_client._android_device_or_serial 560 return ip_client._ssh_settings.hostname 561 562 def dut_is_connected_as_client(self, 563 channel, 564 check_traffic=False, 565 wait_for_addr_timeout=DEFAULT_TIMEOUT): 566 """Checks if DUT is successfully connected to AP. 567 568 Args: 569 channel: int, channel of the AP network (to retrieve interfaces) 570 check_traffic: bool, if true, verifies traffic between DUT and AP, 571 else just checks ping. 572 wait_for_addr_timeout: int, time, in seconds, to wait when getting 573 DUT and AP addresses 574 575 Returns: 576 True, if connected correctly 577 False, otherwise 578 """ 579 try: 580 dut_client_interface = self.get_dut_interface_by_role( 581 INTERFACE_ROLE_CLIENT, 582 wait_for_addr_timeout=wait_for_addr_timeout) 583 ap_ipv4 = self.get_ap_ipv4_address(channel, 584 timeout=wait_for_addr_timeout) 585 except ConnectionError as err: 586 self.log.error( 587 'Failed to retrieve interfaces and addresses. Err: %s' % err) 588 return False 589 590 if not self.device_can_ping_addr(self.dut, ap_ipv4): 591 self.log.error('Failed to ping from DUT to AP.') 592 return False 593 594 if not self.device_can_ping_addr(self.access_point, 595 dut_client_interface.ipv4): 596 self.log.error('Failed to ping from AP to DUT.') 597 return False 598 599 if check_traffic: 600 try: 601 self.run_iperf_traffic(self.ap_iperf_client, 602 dut_client_interface.ipv4) 603 except ConnectionError as err: 604 self.log.error('Failed to run traffic between DUT and AP.') 605 return False 606 return True 607 608 def client_is_connected_to_soft_ap( 609 self, 610 client, 611 client_interface=ANDROID_DEFAULT_WLAN_INTERFACE, 612 check_traffic=False, 613 wait_for_addr_timeout=DEFAULT_TIMEOUT): 614 """Checks if client is successfully connected to DUT SoftAP. 615 616 Args: 617 client: SoftApClient to check 618 client_interface: string, wlan interface name of client 619 check_traffic: bool, if true, verifies traffic between client and 620 DUT, else just checks ping. 621 wait_for_addr_timeout: int, time, in seconds, to wait when getting 622 DUT and client addresses 623 624 Returns: 625 True, if connected correctly 626 False, otherwise 627 """ 628 629 try: 630 dut_ap_interface = self.get_dut_interface_by_role( 631 INTERFACE_ROLE_AP, wait_for_addr_timeout=wait_for_addr_timeout) 632 client_ipv4 = self.wait_for_ipv4_address( 633 client.w_device, 634 client_interface, 635 timeout=wait_for_addr_timeout) 636 except ConnectionError as err: 637 self.log.error( 638 'Failed to retrieve interfaces and addresses. Err: %s' % err) 639 return False 640 641 if not self.device_can_ping_addr(self.dut, client_ipv4): 642 self.log.error('Failed to ping client (%s) from DUT.' % 643 client_ipv4) 644 return False 645 if not self.device_can_ping_addr(client.w_device, 646 dut_ap_interface.ipv4): 647 self.log.error('Failed to ping DUT from client (%s)' % client_ipv4) 648 return False 649 650 if check_traffic: 651 try: 652 self.run_iperf_traffic(client.ip_client, dut_ap_interface.ipv4) 653 except ConnectionError as err: 654 self.log.error( 655 'Failed to pass traffic between client (%s) and DUT.' % 656 client_ipv4) 657 return False 658 return True 659 660 def verify_soft_ap_connectivity_from_state(self, state, client): 661 """Verifies SoftAP state based on a client connection. 662 663 Args: 664 state: bool, whether SoftAP should be up 665 client: SoftApClient, to verify connectivity (or lack therof) 666 """ 667 if state == STATE_UP: 668 return self.client_is_connected_to_soft_ap(client) 669 else: 670 with utils.SuppressLogOutput(): 671 try: 672 return not self.client_is_connected_to_soft_ap( 673 client, 674 wait_for_addr_timeout=DEFAULT_NO_ADDR_EXPECTED_TIMEOUT) 675 # Allow a failed to find ap interface error 676 except LookupError as err: 677 self.log.debug('Hit expected LookupError: %s' % err) 678 return True 679 680 def verify_client_mode_connectivity_from_state(self, state, channel): 681 """Verifies client mode state based on DUT-AP connection. 682 683 Args: 684 state: bool, whether client mode should be up 685 channel: int, channel of the APs network 686 """ 687 if state == STATE_UP: 688 return self.dut_is_connected_as_client(channel) 689 else: 690 with utils.SuppressLogOutput(): 691 try: 692 return not self.dut_is_connected_as_client( 693 channel, 694 wait_for_addr_timeout=DEFAULT_NO_ADDR_EXPECTED_TIMEOUT) 695 # Allow a failed to find client interface error 696 except LookupError as err: 697 self.log.debug('Hit expected LookupError: %s' % err) 698 return True 699 700# Test Types 701 702 def verify_soft_ap_associate_only(self, client, settings): 703 if not self.associate_with_soft_ap(client.w_device, settings): 704 asserts.fail('Failed to associate client with SoftAP.') 705 706 def verify_soft_ap_associate_and_ping(self, client, settings): 707 self.verify_soft_ap_associate_only(client, settings) 708 if not self.client_is_connected_to_soft_ap(client): 709 asserts.fail('Client and SoftAP could not ping eachother.') 710 711 def verify_soft_ap_associate_and_pass_traffic(self, client, settings): 712 self.verify_soft_ap_associate_only(client, settings) 713 if not self.client_is_connected_to_soft_ap(client, check_traffic=True): 714 asserts.fail( 715 'Client and SoftAP not responding to pings and passing traffic ' 716 'as expected.') 717 718# Runners for Generated Test Cases 719 720 def run_soft_ap_association_stress_test(self, settings): 721 """Sets up a SoftAP, and repeatedly associates and disassociates a 722 client. 723 724 Args: 725 settings: test configuration settings, see 726 test_soft_ap_association_stress for details 727 """ 728 client = settings['client'] 729 soft_ap_params = settings['soft_ap_params'] 730 test_type = settings['test_type'] 731 if not test_type in TEST_TYPES: 732 raise ValueError('Unrecognized test type %s' % test_type) 733 iterations = settings['iterations'] 734 self.log.info( 735 'Running association stress test type %s in iteration %s times' % 736 (test_type, iterations)) 737 738 self.start_soft_ap(soft_ap_params) 739 740 passed_count = 0 741 for run in range(iterations): 742 try: 743 self.log.info('Starting SoftAp association run %s' % 744 str(run + 1)) 745 746 if test_type == TEST_TYPE_ASSOCIATE_ONLY: 747 self.verify_soft_ap_associate_only(client, soft_ap_params) 748 749 elif test_type == TEST_TYPE_ASSOCIATE_AND_PING: 750 self.verify_soft_ap_associate_and_ping( 751 client, soft_ap_params) 752 753 elif test_type == TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC: 754 self.verify_soft_ap_associate_and_pass_traffic( 755 client, soft_ap_params) 756 757 else: 758 raise AttributeError('Invalid test type: %s' % test_type) 759 760 except signals.TestFailure as err: 761 self.log.error( 762 'SoftAp association stress run %s failed. Err: %s' % 763 (str(run + 1), err.details)) 764 else: 765 self.log.info('SoftAp association stress run %s successful.' % 766 str(run + 1)) 767 passed_count += 1 768 769 if passed_count < iterations: 770 asserts.fail( 771 'SoftAp association stress test passed on %s/%s runs.' % 772 (passed_count, iterations)) 773 774 asserts.explicit_pass( 775 'SoftAp association stress test passed on %s/%s runs.' % 776 (passed_count, iterations)) 777 778# Alternate SoftAP and Client mode test 779 780 def run_soft_ap_and_client_mode_alternating_test(self, settings): 781 """Runs a single soft_ap and client alternating stress test. 782 783 See test_soft_ap_and_client_mode_alternating_stress for details. 784 """ 785 iterations = settings['iterations'] 786 pass_count = 0 787 client = self.primary_client 788 current_soft_ap_state = STATE_DOWN 789 current_client_mode_state = STATE_DOWN 790 791 self.client_mode_toggle_pre_test(settings) 792 for iteration in range(iterations): 793 passes = True 794 795 # Attempt to toggle SoftAP on, then off 796 for _ in range(2): 797 (current_soft_ap_state, err) = self.run_toggle_iteration_func( 798 self.soft_ap_toggle_test_iteration, settings, 799 current_soft_ap_state) 800 if err: 801 self.log.error('Iteration %s failed. Err: %s' % 802 (str(iteration + 1), err)) 803 passes = False 804 if current_soft_ap_state == STATE_DOWN: 805 break 806 807 # Attempt to toggle Client mode on, then off 808 for _ in range(2): 809 (current_client_mode_state, 810 err) = self.run_toggle_iteration_func( 811 self.client_mode_toggle_test_iteration, settings, 812 current_client_mode_state) 813 if err: 814 self.log.error('Iteration %s failed. Err: %s' % 815 (str(iteration + 1), err)) 816 passes = False 817 if current_client_mode_state == STATE_DOWN: 818 break 819 820 if passes: 821 pass_count += 1 822 823 if pass_count == iterations: 824 asserts.explicit_pass( 825 'Toggle SoftAP and client mode stress test passed %s/%s times.' 826 % (pass_count, iterations)) 827 else: 828 asserts.fail( 829 'Toggle SoftAP and client mode stress test only passed %s/%s ' 830 'times.' % (pass_count, iterations)) 831 832# Toggle Stress Test Helper Functions 833 834 def run_toggle_stress_test(self, settings): 835 """Runner function for toggle stress tests. 836 837 Repeats some test function through stress test iterations, logging 838 failures, tracking pass rate, managing states, etc. 839 840 Args: 841 settings: dict, stress test settings 842 843 Asserts: 844 PASS: if all iterations of the test function pass 845 FAIL: if any iteration of the test function fails 846 """ 847 test_runner_func = settings['test_runner_func'] 848 pre_test_func = settings.get('pre_test_func', None) 849 iterations = settings['iterations'] 850 if pre_test_func: 851 pre_test_func(settings) 852 853 pass_count = 0 854 current_state = STATE_DOWN 855 for iteration in range(iterations): 856 (current_state, 857 err) = self.run_toggle_iteration_func(test_runner_func, settings, 858 current_state) 859 if err: 860 self.log.error('Iteration %s failed. Err: %s' % 861 (str(iteration + 1), err)) 862 else: 863 pass_count += 1 864 865 if pass_count == iterations: 866 asserts.explicit_pass('Stress test passed %s/%s times.' % 867 (pass_count, iterations)) 868 else: 869 asserts.fail('Stress test only passed %s/%s ' 870 'times.' % (pass_count, iterations)) 871 872 def run_toggle_iteration_func(self, func, settings, current_state): 873 """Runs a toggle iteration function, updating the current state 874 based on what the toggle iteration function raises. 875 876 Used for toggle stress tests. 877 878 Note on EnvironmentError vs StressTestIterationFailure: 879 StressTestIterationFailure is raised by func when the toggle occurs 880 but connectivty or some other post-toggle check fails (i.e. the 881 next iteration should toggle to the next state.) 882 883 EnvironmentError is raise by func when the toggle itself fails (i.e 884 the next iteration should retry the same toggle again.) 885 886 Args: 887 func: toggle iteration func to run (e.g soft_ap_toggle_iteration) 888 settings: dict, stress test settings 889 current_state: bool, the current state of the mode being toggled 890 891 Returns: 892 (new_state, err): 893 new_state: bool, state of the mode after toggle attempt 894 err: exception, if any are raise, else None 895 """ 896 try: 897 func(settings, current_state) 898 except EnvironmentError as err: 899 return (current_state, err) 900 except StressTestIterationFailure as err: 901 return (not current_state, err) 902 else: 903 return (not current_state, None) 904 905# Stress Test Toggle Functions 906 907 def start_soft_ap_and_verify_connected(self, client, soft_ap_params): 908 """Sets up SoftAP, associates a client, then verifies connection. 909 910 Args: 911 client: SoftApClient, client to use to verify SoftAP 912 soft_ap_params: dict, containing parameters to setup softap 913 914 Raises: 915 StressTestIterationFailure, if toggle occurs, but connection 916 is not functioning as expected 917 """ 918 # Change SSID every time, to avoid client connection issues. 919 soft_ap_params['ssid'] = utils.rand_ascii_str( 920 hostapd_constants.AP_SSID_LENGTH_2G) 921 self.start_soft_ap(soft_ap_params) 922 associated = self.associate_with_soft_ap(client.w_device, 923 soft_ap_params) 924 if not associated: 925 raise StressTestIterationFailure( 926 'Failed to associated client to DUT SoftAP. ' 927 'Continuing with iterations.') 928 929 if not self.verify_soft_ap_connectivity_from_state(STATE_UP, client): 930 raise StressTestIterationFailure( 931 'Failed to ping between client and DUT. Continuing ' 932 'with iterations.') 933 934 def stop_soft_ap_and_verify_disconnected(self, client, soft_ap_params): 935 """Tears down SoftAP, and verifies connection is down. 936 937 Args: 938 client: SoftApClient, client to use to verify SoftAP 939 soft_ap_params: dict, containing parameters of SoftAP to teardown 940 941 Raise: 942 EnvironmentError, if client and AP can still communicate 943 """ 944 self.log.info('Stopping SoftAP on DUT.') 945 self.stop_soft_ap(soft_ap_params) 946 947 if not self.verify_soft_ap_connectivity_from_state(STATE_DOWN, client): 948 raise EnvironmentError( 949 'Client can still ping DUT. Continuing with ' 950 'iterations.') 951 952 def start_client_mode_and_verify_connected(self, ap_params): 953 """Connects DUT to AP in client mode and verifies connection 954 955 Args: 956 ap_params: dict, containing parameters of the AP network 957 958 Raises: 959 EnvironmentError, if DUT fails to associate altogether 960 StressTestIterationFailure, if DUT associates but connection is not 961 functioning as expected. 962 """ 963 ap_ssid = ap_params['ssid'] 964 ap_password = ap_params['password'] 965 ap_channel = ap_params['channel'] 966 ap_security = ap_params.get('security') 967 968 if ap_security: 969 ap_security_mode = ap_security.security_mode_string 970 else: 971 ap_security_mode = None 972 973 self.log.info('Associating DUT with AP network: %s' % ap_ssid) 974 associated = self.dut.associate( 975 target_ssid=ap_ssid, 976 target_pwd=ap_password, 977 target_security=hostapd_constants. 978 SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get( 979 ap_security_mode, None)) 980 if not associated: 981 raise EnvironmentError('Failed to associate DUT in client mode.') 982 else: 983 self.log.info('Association successful.') 984 985 if not self.verify_client_mode_connectivity_from_state( 986 STATE_UP, ap_channel): 987 raise StressTestIterationFailure('Failed to ping AP from DUT.') 988 989 def stop_client_mode_and_verify_disconnected(self, ap_params): 990 """Disconnects DUT from AP and verifies connection is down. 991 992 Args: 993 ap_params: dict, containing parameters of the AP network 994 995 Raises: 996 EnvironmentError, if DUT and AP can still communicate 997 """ 998 self.log.info('Disconnecting DUT from AP.') 999 self.dut.disconnect() 1000 if not self.verify_client_mode_connectivity_from_state( 1001 STATE_DOWN, ap_params['channel']): 1002 raise EnvironmentError('DUT can still ping AP.') 1003 1004# Toggle Stress Test Iteration and Pre-Test Functions 1005 1006# SoftAP Toggle Stress Test Helper Functions 1007 1008 def soft_ap_toggle_test_iteration(self, settings, current_state): 1009 """Runs a single iteration of SoftAP toggle stress test 1010 1011 Args: 1012 settings: dict, containing test settings 1013 current_state: bool, current state of SoftAP (True if up, 1014 else False) 1015 1016 Raises: 1017 StressTestIterationFailure, if toggle occurs but mode isn't 1018 functioning correctly. 1019 EnvironmentError, if toggle fails to occur at all 1020 """ 1021 soft_ap_params = settings['soft_ap_params'] 1022 self.log.info('Toggling SoftAP %s.' % 1023 ('down' if current_state else 'up')) 1024 1025 if current_state == STATE_DOWN: 1026 self.start_soft_ap_and_verify_connected(self.primary_client, 1027 soft_ap_params) 1028 1029 else: 1030 self.stop_soft_ap_and_verify_disconnected(self.primary_client, 1031 soft_ap_params) 1032 1033# Client Mode Toggle Stress Test Helper Functions 1034 1035 def client_mode_toggle_pre_test(self, settings): 1036 """Prepares the AP before client mode toggle tests 1037 1038 Args: 1039 settings: dict, stress test settings 1040 1041 Raises: 1042 ConnectionError, if AP setup fails 1043 """ 1044 ap_params = settings['ap_params'] 1045 ap_channel = ap_params['channel'] 1046 ap_profile = ap_params.pop('profile') 1047 self.log.info('Setting up AP with params: %s' % ap_params) 1048 setup_ap(access_point=self.access_point, 1049 profile_name=ap_profile, 1050 **ap_params) 1051 # Confirms AP assigned itself an address 1052 self.get_ap_ipv4_address(ap_channel) 1053 1054 def client_mode_toggle_test_iteration(self, settings, current_state): 1055 """Runs a single iteration of client mode toggle stress test 1056 1057 Args: 1058 settings: dict, containing test settings 1059 current_state: bool, current state of client mode (True if up, 1060 else False) 1061 1062 Raises: 1063 StressTestIterationFailure, if toggle occurs but mode isn't 1064 functioning correctly. 1065 EnvironmentError, if toggle fails to occur at all 1066 """ 1067 # TODO(b/168054673): Use client connections and policy connect 1068 ap_params = settings['ap_params'] 1069 self.log.info('Toggling client mode %s' % 1070 ('off' if current_state else 'on')) 1071 1072 if current_state == STATE_DOWN: 1073 self.start_client_mode_and_verify_connected(ap_params) 1074 1075 else: 1076 self.stop_client_mode_and_verify_disconnected(ap_params) 1077 1078# Toggle SoftAP with Client Mode Up Test Helper Functions 1079 1080 def soft_ap_toggle_with_client_mode_pre_test(self, settings): 1081 """Sets up and verifies client mode before SoftAP toggle test. 1082 Args: 1083 settings: dict, stress test settings 1084 1085 Raises: 1086 ConnectionError, if client mode setup fails 1087 """ 1088 self.client_mode_toggle_pre_test(settings) 1089 try: 1090 self.start_client_mode_and_verify_connected(settings['ap_params']) 1091 except StressTestIterationFailure as err: 1092 # This prevents it being treated as a routine error 1093 raise ConnectionError( 1094 'Failed to set up DUT client mode before SoftAP toggle test.' 1095 'Err: %s' % err) 1096 1097 def soft_ap_toggle_with_client_mode_iteration( 1098 self, 1099 settings, 1100 current_state, 1101 ): 1102 """Runs single iteration of SoftAP toggle stress with client mode test. 1103 1104 Args: 1105 settings: dict, containing test settings 1106 current_state: bool, current state of SoftAP (True if up, 1107 else False) 1108 1109 Raises: 1110 StressTestIterationFailure, if toggle occurs but mode isn't 1111 functioning correctly. 1112 EnvironmentError, if toggle fails to occur at all 1113 """ 1114 ap_params = settings['ap_params'] 1115 ap_channel = ap_params['channel'] 1116 self.soft_ap_toggle_test_iteration(settings, current_state) 1117 if not self.dut_is_connected_as_client(ap_channel): 1118 raise StressTestIterationFailure( 1119 'DUT client mode is no longer functional after SoftAP toggle.') 1120 1121# Toggle Client Mode with SoftAP Up Test Helper Functions 1122 1123 def client_mode_toggle_with_soft_ap_pre_test(self, settings): 1124 """Sets up and verifies softap before client mode toggle test. 1125 Args: 1126 settings: dict, stress test settings 1127 1128 Raises: 1129 ConnectionError, if softap setup fails 1130 """ 1131 self.client_mode_toggle_pre_test(settings) 1132 try: 1133 self.start_soft_ap_and_verify_connected(self.primary_client, 1134 settings['soft_ap_params']) 1135 except StressTestIterationFailure as err: 1136 # This prevents it being treated as a routine error 1137 raise ConnectionError( 1138 'Failed to set up SoftAP before client mode toggle test. Err: %s' 1139 % err) 1140 1141 def client_mode_toggle_with_soft_ap_iteration(self, settings, 1142 current_state): 1143 """Runs single iteration of client mode toggle stress with SoftAP test. 1144 1145 Args: 1146 settings: dict, containing test settings 1147 current_state: bool, current state of client mode (True if up, 1148 else False) 1149 1150 Raises: 1151 StressTestIterationFailure, if toggle occurs but mode isn't 1152 functioning correctly. 1153 EnvironmentError, if toggle fails to occur at all 1154 """ 1155 self.client_mode_toggle_test_iteration(settings, current_state) 1156 if not self.client_is_connected_to_soft_ap(self.primary_client): 1157 raise StressTestIterationFailure( 1158 'SoftAP is no longer functional after client mode toggle.') 1159 1160# Toggle SoftAP and Client Mode Randomly 1161 1162 def run_soft_ap_and_client_mode_random_toggle_stress_test(self, settings): 1163 """Runner function for SoftAP and client mode random toggle tests. 1164 1165 Each iteration, randomly chooses if a mode will be toggled or not. 1166 1167 Args: 1168 settings: dict, containing test settings 1169 """ 1170 iterations = settings['iterations'] 1171 pass_count = 0 1172 current_soft_ap_state = STATE_DOWN 1173 current_client_mode_state = STATE_DOWN 1174 ap_channel = settings['ap_params']['channel'] 1175 1176 self.client_mode_toggle_pre_test(settings) 1177 for iteration in range(iterations): 1178 self.log.info('Starting iteration %s out of %s.' % 1179 (str(iteration + 1), iterations)) 1180 passes = True 1181 1182 # Randomly determine if softap, client mode, or both should 1183 # be toggled. 1184 rand_toggle_choice = random.randrange(0, 3) 1185 if rand_toggle_choice <= 1: 1186 (current_soft_ap_state, err) = self.run_toggle_iteration_func( 1187 self.soft_ap_toggle_test_iteration, settings, 1188 current_soft_ap_state) 1189 if err: 1190 self.log.error( 1191 'Iteration %s failed toggling SoftAP. Err: %s' % 1192 (str(iteration + 1), err)) 1193 passes = False 1194 if rand_toggle_choice >= 1: 1195 (current_client_mode_state, 1196 err) = self.run_toggle_iteration_func( 1197 self.client_mode_toggle_test_iteration, settings, 1198 current_client_mode_state) 1199 if err: 1200 self.log.error( 1201 'Iteration %s failed toggling client mode. Err: %s' % 1202 (str(iteration + 1), err)) 1203 passes = False 1204 1205 soft_ap_verified = self.verify_soft_ap_connectivity_from_state( 1206 current_soft_ap_state, self.primary_client) 1207 client_mode_verified = self.verify_client_mode_connectivity_from_state( 1208 current_client_mode_state, ap_channel) 1209 1210 if not soft_ap_verified or not client_mode_verified: 1211 passes = False 1212 if passes: 1213 pass_count += 1 1214 1215 if pass_count == iterations: 1216 asserts.explicit_pass('Stress test passed %s/%s times.' % 1217 (pass_count, iterations)) 1218 else: 1219 asserts.fail('Stress test only passed %s/%s ' 1220 'times.' % (pass_count, iterations)) 1221 1222 1223# Test Cases 1224 1225 def test_soft_ap_2g_open_local(self): 1226 soft_ap_params = { 1227 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1228 'security_type': SECURITY_OPEN, 1229 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1230 'operating_band': OPERATING_BAND_2G 1231 } 1232 self.start_soft_ap(soft_ap_params) 1233 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1234 soft_ap_params) 1235 1236 def test_soft_ap_5g_open_local(self): 1237 soft_ap_params = { 1238 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1239 'security_type': SECURITY_OPEN, 1240 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1241 'operating_band': OPERATING_BAND_5G 1242 } 1243 self.start_soft_ap(soft_ap_params) 1244 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1245 soft_ap_params) 1246 1247 def test_soft_ap_any_open_local(self): 1248 soft_ap_params = { 1249 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1250 'security_type': SECURITY_OPEN, 1251 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1252 'operating_band': OPERATING_BAND_ANY 1253 } 1254 self.start_soft_ap(soft_ap_params) 1255 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1256 soft_ap_params) 1257 1258 def test_soft_ap_2g_wep_local(self): 1259 soft_ap_params = { 1260 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1261 'security_type': SECURITY_WEP, 1262 'password': generate_random_password(security_mode=SECURITY_WEP), 1263 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1264 'operating_band': OPERATING_BAND_2G 1265 } 1266 self.start_soft_ap(soft_ap_params) 1267 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1268 soft_ap_params) 1269 1270 def test_soft_ap_5g_wep_local(self): 1271 soft_ap_params = { 1272 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1273 'security_type': SECURITY_WEP, 1274 'password': generate_random_password(security_mode=SECURITY_WEP), 1275 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1276 'operating_band': OPERATING_BAND_5G 1277 } 1278 self.start_soft_ap(soft_ap_params) 1279 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1280 soft_ap_params) 1281 1282 def test_soft_ap_any_wep_local(self): 1283 soft_ap_params = { 1284 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1285 'security_type': SECURITY_WEP, 1286 'password': generate_random_password(security_mode=SECURITY_WEP), 1287 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1288 'operating_band': OPERATING_BAND_ANY 1289 } 1290 self.start_soft_ap(soft_ap_params) 1291 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, ) 1292 1293 def test_soft_ap_2g_wpa_local(self): 1294 soft_ap_params = { 1295 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1296 'security_type': SECURITY_WPA, 1297 'password': generate_random_password(), 1298 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1299 'operating_band': OPERATING_BAND_2G 1300 } 1301 self.start_soft_ap(soft_ap_params) 1302 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1303 soft_ap_params) 1304 1305 def test_soft_ap_5g_wpa_local(self): 1306 soft_ap_params = { 1307 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1308 'security_type': SECURITY_WPA, 1309 'password': generate_random_password(), 1310 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1311 'operating_band': OPERATING_BAND_5G 1312 } 1313 self.start_soft_ap(soft_ap_params) 1314 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1315 soft_ap_params) 1316 1317 def test_soft_ap_any_wpa_local(self): 1318 soft_ap_params = { 1319 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1320 'security_type': SECURITY_WPA, 1321 'password': generate_random_password(), 1322 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1323 'operating_band': OPERATING_BAND_ANY 1324 } 1325 self.start_soft_ap(soft_ap_params) 1326 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1327 soft_ap_params) 1328 1329 def test_soft_ap_2g_wpa2_local(self): 1330 soft_ap_params = { 1331 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1332 'security_type': SECURITY_WPA2, 1333 'password': generate_random_password(), 1334 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1335 'operating_band': OPERATING_BAND_2G 1336 } 1337 self.start_soft_ap(soft_ap_params) 1338 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1339 soft_ap_params) 1340 1341 def test_soft_ap_5g_wpa2_local(self): 1342 soft_ap_params = { 1343 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1344 'security_type': SECURITY_WPA2, 1345 'password': generate_random_password(), 1346 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1347 'operating_band': OPERATING_BAND_5G 1348 } 1349 self.start_soft_ap(soft_ap_params) 1350 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1351 soft_ap_params) 1352 1353 def test_soft_ap_any_wpa2_local(self): 1354 soft_ap_params = { 1355 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1356 'security_type': SECURITY_WPA2, 1357 'password': generate_random_password(), 1358 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1359 'operating_band': OPERATING_BAND_ANY 1360 } 1361 self.start_soft_ap(soft_ap_params) 1362 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1363 soft_ap_params) 1364 1365 def test_soft_ap_2g_wpa3_local(self): 1366 soft_ap_params = { 1367 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1368 'security_type': SECURITY_WPA3, 1369 'password': generate_random_password(), 1370 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1371 'operating_band': OPERATING_BAND_2G 1372 } 1373 self.start_soft_ap(soft_ap_params) 1374 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1375 soft_ap_params) 1376 1377 def test_soft_ap_5g_wpa3_local(self): 1378 soft_ap_params = { 1379 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1380 'security_type': SECURITY_WPA3, 1381 'password': generate_random_password(), 1382 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1383 'operating_band': OPERATING_BAND_ANY 1384 } 1385 self.start_soft_ap(soft_ap_params) 1386 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1387 soft_ap_params) 1388 1389 def test_soft_ap_any_wpa3_local(self): 1390 soft_ap_params = { 1391 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1392 'security_type': SECURITY_WPA3, 1393 'password': generate_random_password(), 1394 'connectivity_mode': CONNECTIVITY_MODE_LOCAL, 1395 'operating_band': OPERATING_BAND_ANY 1396 } 1397 self.start_soft_ap(soft_ap_params) 1398 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1399 soft_ap_params) 1400 1401 def test_soft_ap_2g_open_unrestricted(self): 1402 soft_ap_params = { 1403 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1404 'security_type': SECURITY_OPEN, 1405 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1406 'operating_band': OPERATING_BAND_2G 1407 } 1408 self.start_soft_ap(soft_ap_params) 1409 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1410 soft_ap_params) 1411 1412 def test_soft_ap_5g_open_unrestricted(self): 1413 soft_ap_params = { 1414 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1415 'security_type': SECURITY_OPEN, 1416 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1417 'operating_band': OPERATING_BAND_5G 1418 } 1419 self.start_soft_ap(soft_ap_params) 1420 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1421 soft_ap_params) 1422 1423 def test_soft_ap_any_open_unrestricted(self): 1424 soft_ap_params = { 1425 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1426 'security_type': SECURITY_OPEN, 1427 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1428 'operating_band': OPERATING_BAND_ANY 1429 } 1430 self.start_soft_ap(soft_ap_params) 1431 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1432 soft_ap_params) 1433 1434 def test_soft_ap_2g_wep_unrestricted(self): 1435 soft_ap_params = { 1436 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1437 'security_type': SECURITY_WEP, 1438 'password': generate_random_password(security_mode=SECURITY_WEP), 1439 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1440 'operating_band': OPERATING_BAND_2G 1441 } 1442 self.start_soft_ap(soft_ap_params) 1443 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1444 soft_ap_params) 1445 1446 def test_soft_ap_5g_wep_unrestricted(self): 1447 soft_ap_params = { 1448 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1449 'security_type': SECURITY_WEP, 1450 'password': generate_random_password(security_mode=SECURITY_WEP), 1451 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1452 'operating_band': OPERATING_BAND_5G 1453 } 1454 self.start_soft_ap(soft_ap_params) 1455 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1456 soft_ap_params) 1457 1458 def test_soft_ap_any_wep_unrestricted(self): 1459 soft_ap_params = { 1460 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1461 'security_type': SECURITY_WEP, 1462 'password': generate_random_password(security_mode=SECURITY_WEP), 1463 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1464 'operating_band': OPERATING_BAND_ANY 1465 } 1466 self.start_soft_ap(soft_ap_params) 1467 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1468 soft_ap_params) 1469 1470 def test_soft_ap_2g_wpa_unrestricted(self): 1471 soft_ap_params = { 1472 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1473 'security_type': SECURITY_WPA, 1474 'password': generate_random_password(), 1475 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1476 'operating_band': OPERATING_BAND_2G 1477 } 1478 self.start_soft_ap(soft_ap_params) 1479 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1480 soft_ap_params) 1481 1482 def test_soft_ap_5g_wpa_unrestricted(self): 1483 soft_ap_params = { 1484 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1485 'security_type': SECURITY_WPA, 1486 'password': generate_random_password(), 1487 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1488 'operating_band': OPERATING_BAND_5G 1489 } 1490 self.start_soft_ap(soft_ap_params) 1491 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1492 soft_ap_params) 1493 1494 def test_soft_ap_any_wpa_unrestricted(self): 1495 soft_ap_params = { 1496 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1497 'security_type': SECURITY_WPA, 1498 'password': generate_random_password(), 1499 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1500 'operating_band': OPERATING_BAND_ANY 1501 } 1502 self.start_soft_ap(soft_ap_params) 1503 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1504 soft_ap_params) 1505 1506 def test_soft_ap_2g_wpa2_unrestricted(self): 1507 soft_ap_params = { 1508 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1509 'security_type': SECURITY_WPA2, 1510 'password': generate_random_password(), 1511 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1512 'operating_band': OPERATING_BAND_2G 1513 } 1514 self.start_soft_ap(soft_ap_params) 1515 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1516 soft_ap_params) 1517 1518 def test_soft_ap_5g_wpa2_unrestricted(self): 1519 soft_ap_params = { 1520 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1521 'security_type': SECURITY_WPA2, 1522 'password': generate_random_password(), 1523 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1524 'operating_band': OPERATING_BAND_5G 1525 } 1526 self.start_soft_ap(soft_ap_params) 1527 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1528 soft_ap_params) 1529 1530 def test_soft_ap_any_wpa2_unrestricted(self): 1531 soft_ap_params = { 1532 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1533 'security_type': SECURITY_WPA2, 1534 'password': generate_random_password(), 1535 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1536 'operating_band': OPERATING_BAND_ANY 1537 } 1538 self.start_soft_ap(soft_ap_params) 1539 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1540 soft_ap_params) 1541 1542 def test_soft_ap_2g_wpa3_unrestricted(self): 1543 soft_ap_params = { 1544 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G), 1545 'security_type': SECURITY_WPA3, 1546 'password': generate_random_password(), 1547 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1548 'operating_band': OPERATING_BAND_2G 1549 } 1550 self.start_soft_ap(soft_ap_params) 1551 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1552 soft_ap_params) 1553 1554 def test_soft_ap_5g_wpa3_unrestricted(self): 1555 soft_ap_params = { 1556 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1557 'security_type': SECURITY_WPA3, 1558 'password': generate_random_password(), 1559 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1560 'operating_band': OPERATING_BAND_ANY 1561 } 1562 self.start_soft_ap(soft_ap_params) 1563 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1564 soft_ap_params) 1565 1566 def test_soft_ap_any_wpa3_unrestricted(self): 1567 soft_ap_params = { 1568 'ssid': utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G), 1569 'security_type': SECURITY_WPA3, 1570 'password': generate_random_password(), 1571 'connectivity_mode': CONNECTIVITY_MODE_UNRESTRICTED, 1572 'operating_band': OPERATING_BAND_ANY 1573 } 1574 self.start_soft_ap(soft_ap_params) 1575 self.verify_soft_ap_associate_and_pass_traffic(self.primary_client, 1576 soft_ap_params) 1577 1578 def test_multi_client(self): 1579 """Tests multi-client association with a single soft AP network. 1580 1581 This tests associates a variable length list of clients, verfying it can 1582 can ping the SoftAP and pass traffic, and then verfies all previously 1583 associated clients can still ping and pass traffic. 1584 1585 The same occurs in reverse for disassocations. 1586 1587 SoftAP parameters can be changed from default via ACTS config: 1588 Example Config 1589 "soft_ap_test_params" : { 1590 "multi_client_test_params": { 1591 "ssid": "testssid", 1592 "security_type": "wpa2", 1593 "password": "password", 1594 "connectivity_mode": "local_only", 1595 "operating_band": "only_2_4_ghz" 1596 } 1597 } 1598 """ 1599 # TODO(fxb/59335): Validate clients on network can reach eachother. 1600 asserts.skip_if( 1601 len(self.clients) < 2, 'Test requires at least 2 SoftAPClients') 1602 1603 test_params = self.soft_ap_test_params.get('multi_client_test_params', 1604 {}) 1605 soft_ap_params = get_soft_ap_params_from_config_or_default( 1606 test_params.get('soft_ap_params', {})) 1607 1608 self.start_soft_ap(soft_ap_params) 1609 1610 dut_ap_interface = self.get_dut_interface_by_role(INTERFACE_ROLE_AP) 1611 associated = [] 1612 1613 for client in self.clients: 1614 # Associate new client 1615 self.verify_soft_ap_associate_and_ping(client, soft_ap_params) 1616 client_ipv4 = self.wait_for_ipv4_address( 1617 client.w_device, ANDROID_DEFAULT_WLAN_INTERFACE) 1618 1619 # Verify previously associated clients still behave as expected 1620 for client_map in associated: 1621 associated_client = client_map['client'] 1622 associated_client_ipv4 = client_map['ipv4'] 1623 self.log.info( 1624 'Verifying previously associated client %s still functions correctly.' 1625 % associated_client.w_device.device.serial) 1626 if not self.client_is_connected_to_soft_ap(associated_client, 1627 check_traffic=True): 1628 asserts.fail( 1629 'Previously associated client %s failed checks after ' 1630 'client %s associated.' % 1631 (associated_client.w_device.device.serial, 1632 client.w_device.device.serial)) 1633 1634 associated.append({'client': client, 'ipv4': client_ipv4}) 1635 1636 self.log.info( 1637 'All devices successfully associated. Beginning disassociations.') 1638 1639 while len(associated) > 0: 1640 # Disassociate client 1641 client = associated.pop()['client'] 1642 self.disconnect_from_soft_ap(client.w_device) 1643 1644 # Verify still connected clients still behave as expected 1645 for client_map in associated: 1646 associated_client = client_map['client'] 1647 associated_client_ipv4 = client_map['ipv4'] 1648 1649 self.log.info( 1650 'Verifying still associated client %s still functions ' 1651 'correctly.' % associated_client.w_device.device.serial) 1652 if not self.client_is_connected_to_soft_ap(associated_client, 1653 check_traffic=True): 1654 asserts.fail( 1655 'Previously associated client %s failed checks after' 1656 ' client %s disassociated.' % 1657 (associated_client.w_device.device.serial, 1658 client.w_device.device.serial)) 1659 1660 self.log.info('All disassociations occurred smoothly.') 1661 1662 def test_simultaneous_soft_ap_and_client(self): 1663 """ Tests FuchsiaDevice DUT can act as a client and a SoftAP 1664 simultaneously. 1665 1666 Raises: 1667 ConnectionError: if DUT fails to connect as client 1668 RuntimeError: if parallel processes fail to join 1669 TestFailure: if DUT fails to pass traffic as either a client or an 1670 AP 1671 """ 1672 # TODO(fxb/59306): Fix flakey parallel streams. 1673 asserts.skip_if(not self.access_point, 'No access point provided.') 1674 1675 self.log.info('Setting up AP using hostapd.') 1676 test_params = self.soft_ap_test_params.get( 1677 'soft_ap_and_client_test_params', {}) 1678 1679 # Configure AP 1680 ap_params = get_ap_params_from_config_or_default( 1681 test_params.get('ap_params', {})) 1682 1683 # Setup AP and associate DUT 1684 ap_profile = ap_params.pop('profile') 1685 setup_ap(access_point=self.access_point, 1686 profile_name=ap_profile, 1687 **ap_params) 1688 try: 1689 self.start_client_mode_and_verify_connected(ap_params) 1690 except Exception as err: 1691 asserts.fail('Failed to set up client mode. Err: %s' % err) 1692 1693 # Setup SoftAP 1694 soft_ap_params = get_soft_ap_params_from_config_or_default( 1695 test_params.get('soft_ap_params', {})) 1696 self.start_soft_ap_and_verify_connected(self.primary_client, 1697 soft_ap_params) 1698 1699 # Get FuchsiaDevice's AP interface info 1700 dut_ap_interface = self.get_dut_interface_by_role(INTERFACE_ROLE_AP) 1701 dut_client_interface = self.get_dut_interface_by_role( 1702 INTERFACE_ROLE_CLIENT) 1703 1704 # Set up secondary iperf server of FuchsiaDevice 1705 self.log.info('Setting up second iperf server on FuchsiaDevice DUT.') 1706 secondary_iperf_server = iperf_server.IPerfServerOverSsh( 1707 self.iperf_server_config, DEFAULT_IPERF_PORT + 1, use_killall=True) 1708 secondary_iperf_server.start() 1709 1710 # Set up iperf client on AP 1711 self.log.info('Setting up iperf client on AP.') 1712 ap_iperf_client = iperf_client.IPerfClientOverSsh( 1713 self.user_params['AccessPoint'][0]['ssh_config']) 1714 1715 # Setup iperf processes: 1716 # Primary client <-> SoftAP interface on FuchsiaDevice 1717 # AP <-> Client interface on FuchsiaDevice 1718 process_errors = mp.Queue() 1719 iperf_soft_ap = mp.Process( 1720 target=self.run_iperf_traffic_parallel_process, 1721 args=[ 1722 self.primary_client.ip_client, dut_ap_interface.ipv4, 1723 process_errors 1724 ]) 1725 1726 iperf_fuchsia_client = mp.Process( 1727 target=self.run_iperf_traffic_parallel_process, 1728 args=[ap_iperf_client, dut_client_interface.ipv4, process_errors], 1729 kwargs={'server_port': 5202}) 1730 1731 # Run iperf processes simultaneously 1732 self.log.info('Running simultaneous iperf traffic: between AP and DUT ' 1733 'client interface, and DUT AP interface and client.') 1734 1735 iperf_soft_ap.start() 1736 iperf_fuchsia_client.start() 1737 1738 # Block until processes can join or timeout 1739 for proc in [iperf_soft_ap, iperf_fuchsia_client]: 1740 proc.join(timeout=DEFAULT_IPERF_TIMEOUT) 1741 if proc.is_alive(): 1742 proc.terminate() 1743 proc.join() 1744 raise RuntimeError('Failed to join process %s' % proc) 1745 1746 # Stop iperf server (also stopped in teardown class as failsafe) 1747 secondary_iperf_server.stop() 1748 1749 # Check errors from parallel processes 1750 if process_errors.empty(): 1751 asserts.explicit_pass( 1752 'FuchsiaDevice was successfully able to pass traffic as a ' 1753 'client and an AP simultaneously.') 1754 else: 1755 while not process_errors.empty(): 1756 self.log.error('Error in iperf process: %s' % 1757 process_errors.get()) 1758 asserts.fail( 1759 'FuchsiaDevice failed to pass traffic as a client and an AP ' 1760 'simultaneously.') 1761 1762 def test_soft_ap_association_stress(self): 1763 """ Sets up a single AP and repeatedly associate/disassociate 1764 a client, verifying connection every time 1765 1766 Each test creates 1 SoftAP and repeatedly associates/disassociates 1767 client. 1768 1769 Example Config 1770 "soft_ap_test_params" : { 1771 "soft_ap_association_stress_tests": [ 1772 { 1773 "ssid": "test_network", 1774 "security_type": "wpa2", 1775 "password": "password", 1776 "connectivity_mode": "local_only", 1777 "operating_band": "only_2_4_ghz", 1778 "iterations": 10 1779 } 1780 ] 1781 } 1782 """ 1783 tests = self.soft_ap_test_params.get( 1784 'test_soft_ap_association_stress', 1785 [dict(test_name='test_soft_ap_association_stress_default')]) 1786 1787 test_settings_list = [] 1788 for config_settings in tests: 1789 soft_ap_params = get_soft_ap_params_from_config_or_default( 1790 config_settings.get('soft_ap_params', {})) 1791 test_type = config_settings.get('test_type', 1792 'associate_and_pass_traffic') 1793 iterations = config_settings.get('iterations', 1794 DEFAULT_STRESS_TEST_ITERATIONS) 1795 test_settings = { 1796 'test_name': config_settings['test_name'], 1797 'client': self.primary_client, 1798 'soft_ap_params': soft_ap_params, 1799 'test_type': test_type, 1800 'iterations': iterations 1801 } 1802 test_settings_list.append(test_settings) 1803 1804 self.run_generated_testcases(self.run_soft_ap_association_stress_test, 1805 test_settings_list, 1806 name_func=get_test_name_from_settings) 1807 1808 def test_soft_ap_and_client_mode_alternating_stress(self): 1809 """ Runs tests that alternate between SoftAP and Client modes. 1810 1811 Each tests sets up an AP. Then, for each iteration: 1812 - DUT starts up SoftAP, client associates with SoftAP, 1813 connection is verified, then disassociates 1814 - DUT associates to the AP, connection is verified, then 1815 disassociates 1816 1817 Example Config: 1818 "soft_ap_test_params": { 1819 "toggle_soft_ap_and_client_tests": [ 1820 { 1821 "test_name": "test_wpa2_client_ap_toggle", 1822 "ap_params": { 1823 "channel": 6, 1824 "ssid": "test-ap-network", 1825 "security_mode": "wpa2", 1826 "password": "password" 1827 }, 1828 "soft_ap_params": { 1829 "ssid": "test-soft-ap-network", 1830 "security_type": "wpa2", 1831 "password": "other-password", 1832 "connectivity_mode": "local_only", 1833 "operating_band": "only_2_4_ghz" 1834 }, 1835 "iterations": 5 1836 } 1837 ] 1838 } 1839 """ 1840 asserts.skip_if(not self.access_point, 'No access point provided.') 1841 tests = self.soft_ap_test_params.get( 1842 'test_soft_ap_and_client_mode_alternating_stress', [ 1843 dict(test_name= 1844 'test_soft_ap_and_client_mode_alternating_stress_default') 1845 ]) 1846 1847 test_settings_list = [] 1848 for config_settings in tests: 1849 ap_params = get_ap_params_from_config_or_default( 1850 config_settings.get('ap_params', {})) 1851 soft_ap_params = get_soft_ap_params_from_config_or_default( 1852 config_settings.get('soft_ap_params', {})) 1853 iterations = config_settings.get('iterations', 1854 DEFAULT_STRESS_TEST_ITERATIONS) 1855 1856 test_settings = { 1857 'test_name': config_settings['test_name'], 1858 'iterations': iterations, 1859 'soft_ap_params': soft_ap_params, 1860 'ap_params': ap_params, 1861 } 1862 1863 test_settings_list.append(test_settings) 1864 self.run_generated_testcases( 1865 test_func=self.run_soft_ap_and_client_mode_alternating_test, 1866 settings=test_settings_list, 1867 name_func=get_test_name_from_settings) 1868 1869 def test_soft_ap_toggle_stress(self): 1870 """ Runs SoftAP toggling stress test. 1871 1872 Each iteration toggles SoftAP to the opposite state (up or down). 1873 1874 If toggled up, a client is associated and connection is verified 1875 If toggled down, test verifies client is not connected 1876 1877 Will run with default params, but custom tests can be provided in the 1878 ACTS config. 1879 1880 Example Config 1881 "soft_ap_test_params" : { 1882 "test_soft_ap_toggle_stress": [ 1883 "soft_ap_params": { 1884 "security_type": "wpa2", 1885 "password": "password", 1886 "connectivity_mode": "local_only", 1887 "operating_band": "only_2_4_ghz", 1888 }, 1889 "iterations": 10 1890 ] 1891 } 1892 """ 1893 tests = self.soft_ap_test_params.get( 1894 'test_soft_ap_toggle_stress', 1895 [dict(test_name='test_soft_ap_toggle_stress_default')]) 1896 1897 test_settings_list = [] 1898 for config_settings in tests: 1899 soft_ap_params = get_soft_ap_params_from_config_or_default( 1900 config_settings.get('soft_ap_params', {})) 1901 iterations = config_settings.get('iterations', 1902 DEFAULT_STRESS_TEST_ITERATIONS) 1903 test_settings = { 1904 'test_name': config_settings['test_name'], 1905 'test_runner_func': self.soft_ap_toggle_test_iteration, 1906 'soft_ap_params': soft_ap_params, 1907 'iterations': iterations 1908 } 1909 test_settings_list.append(test_settings) 1910 1911 self.run_generated_testcases(self.run_toggle_stress_test, 1912 test_settings_list, 1913 name_func=get_test_name_from_settings) 1914 1915 def test_client_mode_toggle_stress(self): 1916 """ Runs client mode toggling stress test. 1917 1918 Each iteration toggles client mode to the opposite state (up or down). 1919 1920 If toggled up, DUT associates to AP, and connection is verified 1921 If toggled down, test verifies DUT is not connected to AP 1922 1923 Will run with default params, but custom tests can be provided in the 1924 ACTS config. 1925 1926 Example Config 1927 "soft_ap_test_params" : { 1928 "test_client_mode_toggle_stress": [ 1929 "soft_ap_params": { 1930 'ssid': ssid, 1931 'channel': channel, 1932 'security_mode': security, 1933 'password': password 1934 }, 1935 "iterations": 10 1936 ] 1937 } 1938 """ 1939 asserts.skip_if(not self.access_point, 'No access point provided.') 1940 tests = self.soft_ap_test_params.get( 1941 'test_client_mode_toggle_stress', 1942 [dict(test_name='test_client_mode_toggle_stress_default')]) 1943 1944 test_settings_list = [] 1945 for config_settings in tests: 1946 ap_params = get_ap_params_from_config_or_default( 1947 config_settings.get('ap_params', {})) 1948 iterations = config_settings.get('iterations', 1949 DEFAULT_STRESS_TEST_ITERATIONS) 1950 test_settings = { 1951 'test_name': config_settings['test_name'], 1952 'test_runner_func': self.client_mode_toggle_test_iteration, 1953 'pre_test_func': self.client_mode_toggle_pre_test, 1954 'ap_params': ap_params, 1955 'iterations': iterations 1956 } 1957 test_settings_list.append(test_settings) 1958 self.run_generated_testcases(self.run_toggle_stress_test, 1959 test_settings_list, 1960 name_func=get_test_name_from_settings) 1961 1962 def test_soft_ap_toggle_stress_with_client_mode(self): 1963 """Same as test_soft_ap_toggle_stress, but client mode is set up 1964 at test start and verified after every toggle.""" 1965 asserts.skip_if(not self.access_point, 'No access point provided.') 1966 tests = self.soft_ap_test_params.get( 1967 'test_soft_ap_toggle_stress_with_client_mode', [ 1968 dict(test_name= 1969 'test_soft_ap_toggle_stress_with_client_mode_default') 1970 ]) 1971 1972 test_settings_list = [] 1973 for config_settings in tests: 1974 soft_ap_params = get_soft_ap_params_from_config_or_default( 1975 config_settings.get('soft_ap_params', {})) 1976 ap_params = get_ap_params_from_config_or_default( 1977 config_settings.get('ap_params', {})) 1978 iterations = config_settings.get('iterations', 1979 DEFAULT_STRESS_TEST_ITERATIONS) 1980 test_settings = { 1981 'test_name': config_settings['test_name'], 1982 'test_runner_func': 1983 self.soft_ap_toggle_with_client_mode_iteration, 1984 'pre_test_func': self.soft_ap_toggle_with_client_mode_pre_test, 1985 'soft_ap_params': soft_ap_params, 1986 'ap_params': ap_params, 1987 'iterations': iterations 1988 } 1989 test_settings_list.append(test_settings) 1990 self.run_generated_testcases(self.run_toggle_stress_test, 1991 test_settings_list, 1992 name_func=get_test_name_from_settings) 1993 1994 def test_client_mode_toggle_stress_with_soft_ap(self): 1995 """Same as test_client_mode_toggle_stress, but softap is set up at 1996 test start and verified after every toggle.""" 1997 asserts.skip_if(not self.access_point, 'No access point provided.') 1998 tests = self.soft_ap_test_params.get( 1999 'test_client_mode_toggle_stress_with_soft_ap', [ 2000 dict(test_name= 2001 'test_client_mode_toggle_stress_with_soft_ap_default') 2002 ]) 2003 2004 test_settings_list = [] 2005 for config_settings in tests: 2006 soft_ap_params = get_soft_ap_params_from_config_or_default( 2007 config_settings.get('soft_ap_params', {})) 2008 ap_params = get_ap_params_from_config_or_default( 2009 config_settings.get('ap_params', {})) 2010 iterations = config_settings.get('iterations', 2011 DEFAULT_STRESS_TEST_ITERATIONS) 2012 test_settings = { 2013 'test_name': config_settings['test_name'], 2014 'test_runner_func': 2015 self.client_mode_toggle_with_soft_ap_iteration, 2016 'pre_test_func': self.client_mode_toggle_with_soft_ap_pre_test, 2017 'soft_ap_params': soft_ap_params, 2018 'ap_params': ap_params, 2019 'iterations': iterations 2020 } 2021 test_settings_list.append(test_settings) 2022 self.run_generated_testcases(self.run_toggle_stress_test, 2023 test_settings_list, 2024 name_func=get_test_name_from_settings) 2025 2026 def test_soft_ap_and_client_mode_random_toggle_stress(self): 2027 """Same as above toggle stres tests, but each iteration, either softap, 2028 client mode, or both are toggled, then states are verified.""" 2029 asserts.skip_if(not self.access_point, 'No access point provided.') 2030 tests = self.soft_ap_test_params.get( 2031 'test_soft_ap_and_client_mode_random_toggle_stress', [ 2032 dict( 2033 test_name= 2034 'test_soft_ap_and_client_mode_random_toggle_stress_default' 2035 ) 2036 ]) 2037 2038 test_settings_list = [] 2039 for config_settings in tests: 2040 soft_ap_params = get_soft_ap_params_from_config_or_default( 2041 config_settings.get('soft_ap_params', {})) 2042 ap_params = get_ap_params_from_config_or_default( 2043 config_settings.get('ap_params', {})) 2044 iterations = config_settings.get('iterations', 2045 DEFAULT_STRESS_TEST_ITERATIONS) 2046 test_settings = { 2047 'test_name': config_settings['test_name'], 2048 'soft_ap_params': soft_ap_params, 2049 'ap_params': ap_params, 2050 'iterations': iterations 2051 } 2052 test_settings_list.append(test_settings) 2053 self.run_generated_testcases( 2054 self.run_soft_ap_and_client_mode_random_toggle_stress_test, 2055 test_settings_list, 2056 name_func=get_test_name_from_settings) 2057