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