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