1#!/usr/bin/env python3 2# 3# Copyright 2019 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import inspect 18import logging 19 20import acts_contrib.test_utils.wifi.wifi_test_utils as awutils 21from acts.utils import get_interface_ip_addresses 22from acts.utils import adb_shell_ping 23 24from acts import asserts 25from acts.controllers import iperf_client 26from acts.controllers.fuchsia_device import FuchsiaDevice 27from acts.controllers.android_device import AndroidDevice 28 29 30def create_wlan_device(hardware_device): 31 """Creates a generic WLAN device based on type of device that is sent to 32 the functions. 33 34 Args: 35 hardware_device: A WLAN hardware device that is supported by ACTS. 36 """ 37 if isinstance(hardware_device, FuchsiaDevice): 38 return FuchsiaWlanDevice(hardware_device) 39 elif isinstance(hardware_device, AndroidDevice): 40 return AndroidWlanDevice(hardware_device) 41 else: 42 raise ValueError('Unable to create WlanDevice for type %s' % 43 type(hardware_device)) 44 45 46FUCHSIA_VALID_SECURITY_TYPES = {"none", "wep", "wpa", "wpa2", "wpa3"} 47 48 49class WlanDevice(object): 50 """Class representing a generic WLAN device. 51 52 Each object of this class represents a generic WLAN device. 53 Android device and Fuchsia devices are the currently supported devices/ 54 55 Attributes: 56 device: A generic WLAN device. 57 """ 58 def __init__(self, device): 59 self.device = device 60 self.log = logging 61 self.identifier = None 62 63 def wifi_toggle_state(self, state): 64 """Base generic WLAN interface. Only called if not overridden by 65 another supported device. 66 """ 67 raise NotImplementedError("{} must be defined.".format( 68 inspect.currentframe().f_code.co_name)) 69 70 def reset_wifi(self): 71 """Base generic WLAN interface. Only called if not overridden by 72 another supported device. 73 """ 74 raise NotImplementedError("{} must be defined.".format( 75 inspect.currentframe().f_code.co_name)) 76 77 def take_bug_report(self, test_name, begin_time): 78 """Base generic WLAN interface. Only called if not overridden by 79 another supported device. 80 """ 81 raise NotImplementedError("{} must be defined.".format( 82 inspect.currentframe().f_code.co_name)) 83 84 def get_log(self, test_name, begin_time): 85 """Base generic WLAN interface. Only called if not overridden by 86 another supported device. 87 """ 88 raise NotImplementedError("{} must be defined.".format( 89 inspect.currentframe().f_code.co_name)) 90 91 def turn_location_off_and_scan_toggle_off(self): 92 """Base generic WLAN interface. Only called if not overridden by 93 another supported device. 94 """ 95 raise NotImplementedError("{} must be defined.".format( 96 inspect.currentframe().f_code.co_name)) 97 98 def associate(self, 99 target_ssid, 100 target_pwd=None, 101 check_connectivity=True, 102 hidden=False, 103 target_security=None): 104 """Base generic WLAN interface. Only called if not overriden by 105 another supported device. 106 """ 107 raise NotImplementedError("{} must be defined.".format( 108 inspect.currentframe().f_code.co_name)) 109 110 def disconnect(self): 111 """Base generic WLAN interface. Only called if not overridden by 112 another supported device. 113 """ 114 raise NotImplementedError("{} must be defined.".format( 115 inspect.currentframe().f_code.co_name)) 116 117 def get_wlan_interface_id_list(self): 118 """Base generic WLAN interface. Only called if not overridden by 119 another supported device. 120 """ 121 raise NotImplementedError("{} must be defined.".format( 122 inspect.currentframe().f_code.co_name)) 123 124 def get_default_wlan_test_interface(self): 125 raise NotImplementedError("{} must be defined.".format( 126 inspect.currentframe().f_code.co_name)) 127 128 def destroy_wlan_interface(self, iface_id): 129 """Base generic WLAN interface. Only called if not overridden by 130 another supported device. 131 """ 132 raise NotImplementedError("{} must be defined.".format( 133 inspect.currentframe().f_code.co_name)) 134 135 def send_command(self, command): 136 raise NotImplementedError("{} must be defined.".format( 137 inspect.currentframe().f_code.co_name)) 138 139 def get_interface_ip_addresses(self, interface): 140 raise NotImplementedError("{} must be defined.".format( 141 inspect.currentframe().f_code.co_name)) 142 143 def is_connected(self, ssid=None): 144 raise NotImplementedError("{} must be defined.".format( 145 inspect.currentframe().f_code.co_name)) 146 147 def can_ping(self, 148 dest_ip, 149 count=3, 150 interval=1000, 151 timeout=1000, 152 size=25, 153 additional_ping_params=None): 154 raise NotImplementedError("{} must be defined.".format( 155 inspect.currentframe().f_code.co_name)) 156 157 def ping(self, 158 dest_ip, 159 count=3, 160 interval=1000, 161 timeout=1000, 162 size=25, 163 additional_ping_params=None): 164 raise NotImplementedError("{} must be defined.".format( 165 inspect.currentframe().f_code.co_name)) 166 167 def hard_power_cycle(self, pdus=None): 168 raise NotImplementedError("{} must be defined.".format( 169 inspect.currentframe().f_code.co_name)) 170 171 def save_network(self, ssid): 172 raise NotImplementedError("{} must be defined.".format( 173 inspect.currentframe().f_code.co_name)) 174 175 def clear_saved_networks(self): 176 raise NotImplementedError("{} must be defined.".format( 177 inspect.currentframe().f_code.co_name)) 178 179 def create_iperf_client(self, test_interface=None): 180 raise NotImplementedError("{} must be defined.".format( 181 inspect.currentframe().f_code.co_name)) 182 183 184class AndroidWlanDevice(WlanDevice): 185 """Class wrapper for an Android WLAN device. 186 187 Each object of this class represents a generic WLAN device. 188 Android device and Fuchsia devices are the currently supported devices/ 189 190 Attributes: 191 android_device: An Android WLAN device. 192 """ 193 def __init__(self, android_device): 194 super().__init__(android_device) 195 self.identifier = android_device.serial 196 197 def wifi_toggle_state(self, state): 198 awutils.wifi_toggle_state(self.device, state) 199 200 def reset_wifi(self): 201 awutils.reset_wifi(self.device) 202 203 def take_bug_report(self, test_name, begin_time): 204 self.device.take_bug_report(test_name, begin_time) 205 206 def get_log(self, test_name, begin_time): 207 self.device.cat_adb_log(test_name, begin_time) 208 209 def turn_location_off_and_scan_toggle_off(self): 210 awutils.turn_location_off_and_scan_toggle_off(self.device) 211 212 def associate(self, 213 target_ssid, 214 target_pwd=None, 215 key_mgmt=None, 216 check_connectivity=True, 217 hidden=False, 218 target_security=None): 219 """Function to associate an Android WLAN device. 220 221 Args: 222 target_ssid: SSID to associate to. 223 target_pwd: Password for the SSID, if necessary. 224 key_mgmt: The hostapd wpa_key_mgmt value, distinguishes wpa3 from 225 wpa2 for android tests. 226 check_connectivity: Whether to check for internet connectivity. 227 hidden: Whether the network is hidden. 228 Returns: 229 True if successfully connected to WLAN, False if not. 230 """ 231 network = {'SSID': target_ssid, 'hiddenSSID': hidden} 232 if target_pwd: 233 network['password'] = target_pwd 234 if key_mgmt: 235 network['security'] = key_mgmt 236 try: 237 awutils.connect_to_wifi_network( 238 self.device, 239 network, 240 check_connectivity=check_connectivity, 241 hidden=hidden) 242 return True 243 except Exception as e: 244 self.device.log.info('Failed to associated (%s)' % e) 245 return False 246 247 def disconnect(self): 248 awutils.turn_location_off_and_scan_toggle_off(self.device) 249 250 def get_wlan_interface_id_list(self): 251 pass 252 253 def get_default_wlan_test_interface(self): 254 return 'wlan0' 255 256 def destroy_wlan_interface(self, iface_id): 257 pass 258 259 def send_command(self, command): 260 return self.device.adb.shell(str(command)) 261 262 def get_interface_ip_addresses(self, interface): 263 return get_interface_ip_addresses(self.device, interface) 264 265 def is_connected(self, ssid=None): 266 wifi_info = self.device.droid.wifiGetConnectionInfo() 267 if ssid: 268 return 'BSSID' in wifi_info and wifi_info['SSID'] == ssid 269 return 'BSSID' in wifi_info 270 271 def can_ping(self, 272 dest_ip, 273 count=3, 274 interval=1000, 275 timeout=1000, 276 size=25, 277 additional_ping_params=None): 278 return adb_shell_ping(self.device, 279 dest_ip=dest_ip, 280 count=count, 281 timeout=timeout) 282 283 def ping(self, dest_ip, count=3, interval=1000, timeout=1000, size=25): 284 pass 285 286 def hard_power_cycle(self, pdus): 287 pass 288 289 def save_network(self, ssid): 290 pass 291 292 def clear_saved_networks(self): 293 pass 294 295 def create_iperf_client(self, test_interface=None): 296 """ Returns an iperf client on the Android, without requiring a 297 specific config. 298 299 Args: 300 test_interface: optional, string, name of test interface. 301 302 Returns: 303 IPerfClient object 304 """ 305 if not test_interface: 306 test_interface = self.get_default_wlan_test_interface() 307 308 return iperf_client.IPerfClientOverAdb( 309 android_device_or_serial=self.device, 310 test_interface=test_interface) 311 312 313class FuchsiaWlanDevice(WlanDevice): 314 """Class wrapper for an Fuchsia WLAN device. 315 316 Each object of this class represents a generic WLAN device. 317 Android device and Fuchsia devices are the currently supported devices/ 318 319 Attributes: 320 fuchsia_device: A Fuchsia WLAN device. 321 """ 322 def __init__(self, fuchsia_device): 323 super().__init__(fuchsia_device) 324 self.identifier = fuchsia_device.ip 325 self.device.configure_wlan() 326 327 def wifi_toggle_state(self, state): 328 """Stub for Fuchsia implementation.""" 329 pass 330 331 def reset_wifi(self): 332 """Stub for Fuchsia implementation.""" 333 pass 334 335 def take_bug_report(self, test_name, begin_time): 336 """Stub for Fuchsia implementation.""" 337 self.device.take_bug_report(test_name, begin_time) 338 339 def get_log(self, test_name, begin_time): 340 """Stub for Fuchsia implementation.""" 341 pass 342 343 def turn_location_off_and_scan_toggle_off(self): 344 """Stub for Fuchsia implementation.""" 345 pass 346 347 def associate(self, 348 target_ssid, 349 target_pwd=None, 350 key_mgmt=None, 351 check_connectivity=True, 352 hidden=False, 353 target_security=None): 354 """Function to associate a Fuchsia WLAN device. 355 356 Args: 357 target_ssid: SSID to associate to. 358 target_pwd: Password for the SSID, if necessary. 359 key_mgmt: the hostapd wpa_key_mgmt, if specified. 360 check_connectivity: Whether to check for internet connectivity. 361 hidden: Whether the network is hidden. 362 target_security: string, target security for network, used to 363 save the network in policy connects (see wlan_policy_lib) 364 Returns: 365 True if successfully connected to WLAN, False if not. 366 """ 367 if self.device.association_mechanism == 'drivers': 368 bss_scan_response = self.device.wlan_lib.wlanScanForBSSInfo() 369 if bss_scan_response.get('error'): 370 self.log.error('Scan for BSS info failed. Err: %s' % 371 bss_scan_response['error']) 372 return False 373 374 bss_descs_for_ssid = bss_scan_response['result'].get( 375 target_ssid, None) 376 if not bss_descs_for_ssid or len(bss_descs_for_ssid) < 1: 377 self.log.error( 378 'Scan failed to find a BSS description for target_ssid %s' 379 % target_ssid) 380 return False 381 382 connection_response = self.device.wlan_lib.wlanConnectToNetwork( 383 target_ssid, bss_descs_for_ssid[0], target_pwd=target_pwd) 384 return self.device.check_connect_response(connection_response) 385 else: 386 return self.device.wlan_policy_controller.save_and_connect( 387 target_ssid, target_security, password=target_pwd) 388 389 def disconnect(self): 390 """Function to disconnect from a Fuchsia WLAN device. 391 Asserts if disconnect was not successful. 392 """ 393 if self.device.association_mechanism == 'drivers': 394 disconnect_response = self.device.wlan_lib.wlanDisconnect() 395 return self.device.check_disconnect_response(disconnect_response) 396 else: 397 return self.device.wlan_policy_controller.remove_all_networks_and_wait_for_no_connections( 398 ) 399 400 def status(self): 401 return self.device.wlan_lib.wlanStatus() 402 403 def can_ping(self, 404 dest_ip, 405 count=3, 406 interval=1000, 407 timeout=1000, 408 size=25, 409 additional_ping_params=None): 410 return self.device.can_ping( 411 dest_ip, 412 count=count, 413 interval=interval, 414 timeout=timeout, 415 size=size, 416 additional_ping_params=additional_ping_params) 417 418 def ping(self, 419 dest_ip, 420 count=3, 421 interval=1000, 422 timeout=1000, 423 size=25, 424 additional_ping_params=None): 425 return self.device.ping(dest_ip, 426 count=count, 427 interval=interval, 428 timeout=timeout, 429 size=size, 430 additional_ping_params=additional_ping_params) 431 432 def get_wlan_interface_id_list(self): 433 """Function to list available WLAN interfaces. 434 435 Returns: 436 A list of wlan interface IDs. 437 """ 438 return self.device.wlan_lib.wlanGetIfaceIdList().get('result') 439 440 def get_default_wlan_test_interface(self): 441 """Returns name of the WLAN client interface""" 442 return self.device.wlan_controller.get_wlan_interface_name() 443 444 def destroy_wlan_interface(self, iface_id): 445 """Function to associate a Fuchsia WLAN device. 446 447 Args: 448 target_ssid: SSID to associate to. 449 target_pwd: Password for the SSID, if necessary. 450 check_connectivity: Whether to check for internet connectivity. 451 hidden: Whether the network is hidden. 452 Returns: 453 True if successfully destroyed wlan interface, False if not. 454 """ 455 result = self.device.wlan_lib.wlanDestroyIface(iface_id) 456 if result.get('error') is None: 457 return True 458 else: 459 self.log.error("Failed to destroy interface with: {}".format( 460 result.get('error'))) 461 return False 462 463 def send_command(self, command): 464 return self.device.send_command_ssh(str(command)).stdout 465 466 def get_interface_ip_addresses(self, interface): 467 return get_interface_ip_addresses(self.device, interface) 468 469 def is_connected(self, ssid=None): 470 """ Determines if wlan_device is connected to wlan network. 471 472 Args: 473 ssid (optional): string, to check if device is connect to a specific 474 network. 475 476 Returns: 477 True, if connected to a network or to the correct network when SSID 478 is provided. 479 False, if not connected or connect to incorrect network when SSID is 480 provided. 481 """ 482 response = self.status() 483 if response.get('error'): 484 raise ConnectionError( 485 'Failed to get client network connection status') 486 487 status = response.get('result') 488 if status and status.get('connected_to'): 489 if ssid: 490 connected_ssid = ''.join( 491 chr(i) for i in status['connected_to']['ssid']) 492 if ssid != connected_ssid: 493 return False 494 return True 495 return False 496 497 def hard_power_cycle(self, pdus): 498 self.device.reboot(reboot_type='hard', testbed_pdus=pdus) 499 500 def save_network(self, target_ssid, security_type=None, target_pwd=None): 501 if self.device.association_mechanism == 'drivers': 502 raise EnvironmentError( 503 'Cannot save network using the drivers. Saved networks are a ' 504 'policy layer concept.') 505 if security_type and security_type not in FUCHSIA_VALID_SECURITY_TYPES: 506 raise TypeError('Invalid security type: %s' % security_type) 507 if not self.device.wlan_policy_controller.save_network( 508 target_ssid, security_type, password=target_pwd): 509 raise EnvironmentError('Failed to save network: %s' % target_ssid) 510 511 def clear_saved_networks(self): 512 if self.device.association_mechanism == 'drivers': 513 raise EnvironmentError( 514 'Cannot clear saved network using the drivers. Saved networks ' 515 'are a policy layer concept.') 516 if not self.device.wlan_policy_controller.remove_all_networks(): 517 raise EnvironmentError('Failed to clear saved networks') 518 519 def create_iperf_client(self, test_interface=None): 520 """ Returns an iperf client on the FuchsiaDevice, without requiring a 521 specific config. 522 523 Args: 524 test_interface: optional, string, name of test interface. Defaults 525 to first found wlan client interface. 526 527 Returns: 528 IPerfClient object 529 """ 530 if not test_interface: 531 test_interface = self.get_default_wlan_test_interface() 532 return iperf_client.IPerfClientOverSsh( 533 { 534 'user': 'fuchsia', 535 'host': self.device.ip, 536 'ssh_config': self.device.ssh_config 537 }, 538 use_paramiko=True, 539 test_interface=test_interface) 540