1#!/usr/bin/env python3 2# 3# Copyright 2017 - 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 fcntl 18import selenium 19import splinter 20import time 21from acts import logger 22from acts.controllers import access_point 23from acts.controllers.ap_lib import bridge_interface 24from acts.controllers.ap_lib import hostapd_security 25from acts.controllers.ap_lib import hostapd_ap_preset 26 27BROWSER_WAIT_SHORT = 1 28BROWSER_WAIT_MED = 3 29BROWSER_WAIT_LONG = 30 30BROWSER_WAIT_EXTRA_LONG = 60 31 32 33def create(configs): 34 """Factory method for retail AP class. 35 36 Args: 37 configs: list of dicts containing ap settings. ap settings must contain 38 the following: brand, model, ip_address, username and password 39 """ 40 SUPPORTED_APS = { 41 ("Netgear", "R7000"): "NetgearR7000AP", 42 ("Netgear", "R7000NA"): "NetgearR7000NAAP", 43 ("Netgear", "R7500"): "NetgearR7500AP", 44 ("Netgear", "R7800"): "NetgearR7800AP", 45 ("Netgear", "R8000"): "NetgearR8000AP", 46 ("Netgear", "R8500"): "NetgearR8500AP", 47 ("Google", "Wifi"): "GoogleWifiAP" 48 } 49 objs = [] 50 for config in configs: 51 try: 52 ap_class_name = SUPPORTED_APS[(config["brand"], config["model"])] 53 ap_class = globals()[ap_class_name] 54 except KeyError: 55 raise KeyError("Invalid retail AP brand and model combination.") 56 objs.append(ap_class(config)) 57 return objs 58 59 60def detroy(objs): 61 return 62 63 64class BlockingBrowser(splinter.driver.webdriver.chrome.WebDriver): 65 """Class that implements a blocking browser session on top of selenium. 66 67 The class inherits from and builds upon splinter/selenium's webdriver class 68 and makes sure that only one such webdriver is active on a machine at any 69 single time. The class ensures single session operation using a lock file. 70 The class is to be used within context managers (e.g. with statements) to 71 ensure locks are always properly released. 72 """ 73 74 def __init__(self, headless, timeout): 75 """Constructor for BlockingBrowser class. 76 77 Args: 78 headless: boolean to control visible/headless browser operation 79 timeout: maximum time allowed to launch browser 80 """ 81 self.log = logger.create_tagged_trace_logger("ChromeDriver") 82 self.chrome_options = splinter.driver.webdriver.chrome.Options() 83 self.chrome_options.add_argument("--no-proxy-server") 84 self.chrome_options.add_argument("--no-sandbox") 85 self.chrome_options.add_argument("--allow-running-insecure-content") 86 self.chrome_options.add_argument("--ignore-certificate-errors") 87 self.chrome_capabilities = selenium.webdriver.common.desired_capabilities.DesiredCapabilities.CHROME.copy( 88 ) 89 self.chrome_capabilities["acceptSslCerts"] = True 90 self.chrome_capabilities["acceptInsecureCerts"] = True 91 if headless: 92 self.chrome_options.add_argument("--headless") 93 self.chrome_options.add_argument("--disable-gpu") 94 self.lock_file_path = "/usr/local/bin/chromedriver" 95 self.timeout = timeout 96 97 def __enter__(self): 98 self.lock_file = open(self.lock_file_path, "r") 99 start_time = time.time() 100 while time.time() < start_time + self.timeout: 101 try: 102 fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) 103 self.driver = selenium.webdriver.Chrome( 104 options=self.chrome_options, 105 desired_capabilities=self.chrome_capabilities) 106 self.element_class = splinter.driver.webdriver.WebDriverElement 107 self._cookie_manager = splinter.driver.webdriver.cookie_manager.CookieManager( 108 self.driver) 109 super(splinter.driver.webdriver.chrome.WebDriver, 110 self).__init__(2) 111 return super(BlockingBrowser, self).__enter__() 112 except BlockingIOError: 113 time.sleep(BROWSER_WAIT_SHORT) 114 raise TimeoutError("Could not start chrome browser in time.") 115 116 def __exit__(self, exc_type, exc_value, traceback): 117 try: 118 super(BlockingBrowser, self).__exit__(exc_type, exc_value, 119 traceback) 120 except: 121 raise RuntimeError("Failed to quit browser. Releasing lock file.") 122 finally: 123 fcntl.flock(self.lock_file, fcntl.LOCK_UN) 124 self.lock_file.close() 125 126 def restart(self): 127 """Method to restart browser session without releasing lock file.""" 128 self.quit() 129 self.__enter__() 130 131 def visit_persistent(self, 132 url, 133 page_load_timeout, 134 num_tries, 135 backup_url="about:blank", 136 check_for_element=None): 137 """Method to visit webpages and retry upon failure. 138 139 The function visits a web page and checks the the resulting URL matches 140 the intended URL, i.e. no redirects have happened 141 142 Args: 143 url: the intended url 144 page_load_timeout: timeout for page visits 145 num_tries: number of tries before url is declared unreachable 146 backup_url: url to visit if first url is not reachable. This can be 147 used to simply refresh the browser and try again or to re-login to 148 the AP 149 check_for_element: element id to check for existence on page 150 """ 151 self.driver.set_page_load_timeout(page_load_timeout) 152 for idx in range(num_tries): 153 try: 154 self.visit(url) 155 except: 156 self.restart() 157 158 page_reached = self.url.split("/")[-1] == url.split("/")[-1] 159 if check_for_element: 160 time.sleep(BROWSER_WAIT_MED) 161 element = self.find_by_id(check_for_element) 162 if not element: 163 page_reached = 0 164 if page_reached: 165 break 166 else: 167 try: 168 self.visit(backup_url) 169 except: 170 self.restart() 171 172 if idx == num_tries - 1: 173 self.log.error("URL unreachable. Current URL: {}".format( 174 self.url)) 175 raise RuntimeError("URL unreachable.") 176 177 178class WifiRetailAP(object): 179 """Base class implementation for retail ap. 180 181 Base class provides functions whose implementation is shared by all aps. 182 If some functions such as set_power not supported by ap, checks will raise 183 exceptions. 184 """ 185 186 def __init__(self, ap_settings): 187 raise NotImplementedError 188 189 def read_ap_settings(self): 190 """Function that reads current ap settings. 191 192 Function implementation is AP dependent and thus base class raises exception 193 if function not implemented in child class. 194 """ 195 raise NotImplementedError 196 197 def validate_ap_settings(self): 198 """Function to validate ap settings. 199 200 This function compares the actual ap settings read from the web GUI 201 with the assumed settings saved in the AP object. When called after AP 202 configuration, this method helps ensure that our configuration was 203 successful. 204 Note: Calling this function updates the stored ap_settings 205 206 Raises: 207 ValueError: If read AP settings do not match stored settings. 208 """ 209 assumed_ap_settings = self.ap_settings.copy() 210 actual_ap_settings = self.read_ap_settings() 211 if assumed_ap_settings != actual_ap_settings: 212 self.log.warning( 213 "Discrepancy in AP settings. Some settings may have been overwritten." 214 ) 215 216 def configure_ap(self, **config_flags): 217 """Function that configures ap based on values of ap_settings. 218 219 Function implementation is AP dependent and thus base class raises exception 220 if function not implemented in child class. 221 222 Args: 223 config_flags: optional configuration flags 224 """ 225 raise NotImplementedError 226 227 def set_region(self, region): 228 """Function that sets AP region. 229 230 This function sets the region for the AP. Note that this may overwrite 231 channel and bandwidth settings in cases where the new region does not 232 support the current wireless configuration. 233 234 Args: 235 region: string indicating AP region 236 """ 237 self.log.warning("Updating region may overwrite wireless settings.") 238 setting_to_update = {"region": region} 239 self.update_ap_settings(setting_to_update) 240 241 def set_radio_on_off(self, network, status): 242 """Function that turns the radio on or off. 243 244 Args: 245 network: string containing network identifier (2G, 5G_1, 5G_2) 246 status: boolean indicating on or off (0: off, 1: on) 247 """ 248 setting_to_update = {"status_{}".format(network): int(status)} 249 self.update_ap_settings(setting_to_update) 250 251 def set_ssid(self, network, ssid): 252 """Function that sets network SSID. 253 254 Args: 255 network: string containing network identifier (2G, 5G_1, 5G_2) 256 ssid: string containing ssid 257 """ 258 setting_to_update = {"ssid_{}".format(network): str(ssid)} 259 self.update_ap_settings(setting_to_update) 260 261 def set_channel(self, network, channel): 262 """Function that sets network channel. 263 264 Args: 265 network: string containing network identifier (2G, 5G_1, 5G_2) 266 channel: string or int containing channel 267 """ 268 setting_to_update = {"channel_{}".format(network): str(channel)} 269 self.update_ap_settings(setting_to_update) 270 271 def set_bandwidth(self, network, bandwidth): 272 """Function that sets network bandwidth/mode. 273 274 Args: 275 network: string containing network identifier (2G, 5G_1, 5G_2) 276 bandwidth: string containing mode, e.g. 11g, VHT20, VHT40, VHT80. 277 """ 278 setting_to_update = {"bandwidth_{}".format(network): str(bandwidth)} 279 self.update_ap_settings(setting_to_update) 280 281 def set_power(self, network, power): 282 """Function that sets network transmit power. 283 284 Args: 285 network: string containing network identifier (2G, 5G_1, 5G_2) 286 power: string containing power level, e.g., 25%, 100% 287 """ 288 setting_to_update = {"power_{}".format(network): str(power)} 289 self.update_ap_settings(setting_to_update) 290 291 def set_security(self, network, security_type, *password): 292 """Function that sets network security setting and password. 293 294 Args: 295 network: string containing network identifier (2G, 5G_1, 5G_2) 296 security: string containing security setting, e.g., WPA2-PSK 297 password: optional argument containing password 298 """ 299 if (len(password) == 1) and (type(password[0]) == str): 300 setting_to_update = { 301 "security_type_{}".format(network): str(security_type), 302 "password_{}".format(network): str(password[0]) 303 } 304 else: 305 setting_to_update = { 306 "security_type_{}".format(network): str(security_type) 307 } 308 self.update_ap_settings(setting_to_update) 309 310 def set_rate(self): 311 """Function that configures rate used by AP. 312 313 Function implementation is not supported by most APs and thus base 314 class raises exception if function not implemented in child class. 315 """ 316 raise NotImplementedError 317 318 def update_ap_settings(self, dict_settings={}, **named_settings): 319 """Function to update settings of existing AP. 320 321 Function copies arguments into ap_settings and calls configure_retail_ap 322 to apply them. 323 324 Args: 325 *dict_settings accepts single dictionary of settings to update 326 **named_settings accepts named settings to update 327 Note: dict and named_settings cannot contain the same settings. 328 """ 329 settings_to_update = dict(dict_settings, **named_settings) 330 if len(settings_to_update) != len(dict_settings) + len(named_settings): 331 raise KeyError("The following keys were passed twice: {}".format( 332 (set(dict_settings.keys()).intersection( 333 set(named_settings.keys()))))) 334 if not set(settings_to_update.keys()).issubset( 335 set(self.ap_settings.keys())): 336 raise KeyError( 337 "The following settings are invalid for this AP: {}".format( 338 set(settings_to_update.keys()).difference( 339 set(self.ap_settings.keys())))) 340 341 updates_requested = False 342 status_toggle_flag = False 343 for setting, value in settings_to_update.items(): 344 if self.ap_settings[setting] != value: 345 self.ap_settings[setting] = value 346 if "status" in setting: 347 status_toggle_flag = True 348 updates_requested = True 349 350 if updates_requested: 351 self.configure_ap(status_toggled=status_toggle_flag) 352 353 def band_lookup_by_channel(self, channel): 354 """Function that gives band name by channel number. 355 356 Args: 357 channel: channel number to lookup 358 Returns: 359 band: name of band which this channel belongs to on this ap 360 """ 361 for key, value in self.channel_band_map.items(): 362 if channel in value: 363 return key 364 raise ValueError("Invalid channel passed in argument.") 365 366 367class NetgearR7000AP(WifiRetailAP): 368 """Class that implements Netgear R7500 AP.""" 369 370 def __init__(self, ap_settings): 371 self.ap_settings = ap_settings.copy() 372 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 373 self.ap_settings["ip_address"])) 374 self.init_gui_data() 375 # Read and update AP settings 376 self.read_ap_settings() 377 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 378 self.update_ap_settings(ap_settings) 379 380 def init_gui_data(self): 381 """Function to initialize data used while interacting with web GUI""" 382 self.config_page = ( 383 "{protocol}://{username}:{password}@" 384 "{ip_address}:{port}/WLG_wireless_dual_band_r10.htm").format( 385 protocol=self.ap_settings["protocol"], 386 username=self.ap_settings["admin_username"], 387 password=self.ap_settings["admin_password"], 388 ip_address=self.ap_settings["ip_address"], 389 port=self.ap_settings["port"]) 390 self.config_page_nologin = ( 391 "{protocol}://{ip_address}:{port}/" 392 "WLG_wireless_dual_band_r10.htm").format( 393 protocol=self.ap_settings["protocol"], 394 ip_address=self.ap_settings["ip_address"], 395 port=self.ap_settings["port"]) 396 self.config_page_advanced = ( 397 "{protocol}://{username}:{password}@" 398 "{ip_address}:{port}/WLG_adv_dual_band2.htm").format( 399 protocol=self.ap_settings["protocol"], 400 username=self.ap_settings["admin_username"], 401 password=self.ap_settings["admin_password"], 402 ip_address=self.ap_settings["ip_address"], 403 port=self.ap_settings["port"]) 404 self.networks = ["2G", "5G_1"] 405 self.channel_band_map = { 406 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 407 "5G_1": [ 408 36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120, 409 124, 128, 132, 136, 140, 149, 153, 157, 161, 165 410 ] 411 } 412 self.region_map = { 413 "1": "Africa", 414 "2": "Asia", 415 "3": "Australia", 416 "4": "Canada", 417 "5": "Europe", 418 "6": "Israel", 419 "7": "Japan", 420 "8": "Korea", 421 "9": "Mexico", 422 "10": "South America", 423 "11": "United States", 424 "12": "Middle East(Algeria/Syria/Yemen)", 425 "14": "Russia", 426 "16": "China", 427 "17": "India", 428 "18": "Malaysia", 429 "19": "Middle East(Iran/Labanon/Qatar)", 430 "20": "Middle East(Turkey/Egypt/Tunisia/Kuwait)", 431 "21": "Middle East(Saudi Arabia)", 432 "22": "Middle East(United Arab Emirates)", 433 "23": "Singapore", 434 "24": "Taiwan" 435 } 436 self.config_page_fields = { 437 "region": "WRegion", 438 ("2G", "status"): "enable_ap", 439 ("5G_1", "status"): "enable_ap_an", 440 ("2G", "ssid"): "ssid", 441 ("5G_1", "ssid"): "ssid_an", 442 ("2G", "channel"): "w_channel", 443 ("5G_1", "channel"): "w_channel_an", 444 ("2G", "bandwidth"): "opmode", 445 ("5G_1", "bandwidth"): "opmode_an", 446 ("2G", "power"): "enable_tpc", 447 ("5G_1", "power"): "enable_tpc_an", 448 ("2G", "security_type"): "security_type", 449 ("5G_1", "security_type"): "security_type_an", 450 ("2G", "password"): "passphrase", 451 ("5G_1", "password"): "passphrase_an" 452 } 453 self.bw_mode_values = { 454 "g and b": "11g", 455 "145Mbps": "VHT20", 456 "300Mbps": "VHT40", 457 "HT80": "VHT80" 458 } 459 self.power_mode_values = { 460 "1": "100%", 461 "2": "75%", 462 "3": "50%", 463 "4": "25%" 464 } 465 self.bw_mode_text = { 466 "11g": "Up to 54 Mbps", 467 "VHT20": "Up to 289 Mbps", 468 "VHT40": "Up to 600 Mbps", 469 "VHT80": "Up to 1300 Mbps" 470 } 471 472 def read_ap_settings(self): 473 """Function to read ap settings.""" 474 with BlockingBrowser(self.ap_settings["headless_browser"], 475 900) as browser: 476 # Visit URL 477 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 478 479 for key, value in self.config_page_fields.items(): 480 if "status" in key: 481 browser.visit_persistent(self.config_page_advanced, 482 BROWSER_WAIT_MED, 10) 483 config_item = browser.find_by_name(value) 484 self.ap_settings["{}_{}".format(key[1], key[0])] = int( 485 config_item.first.checked) 486 browser.visit_persistent(self.config_page, 487 BROWSER_WAIT_MED, 10) 488 else: 489 config_item = browser.find_by_name(value) 490 if "bandwidth" in key: 491 self.ap_settings["{}_{}".format( 492 key[1], key[0])] = self.bw_mode_values[ 493 config_item.first.value] 494 elif "power" in key: 495 self.ap_settings["{}_{}".format( 496 key[1], key[0])] = self.power_mode_values[ 497 config_item.first.value] 498 elif "region" in key: 499 self.ap_settings["region"] = self.region_map[ 500 config_item.first.value] 501 elif "security_type" in key: 502 for item in config_item: 503 if item.checked: 504 self.ap_settings["{}_{}".format( 505 key[1], key[0])] = item.value 506 else: 507 config_item = browser.find_by_name(value) 508 self.ap_settings["{}_{}".format( 509 key[1], key[0])] = config_item.first.value 510 return self.ap_settings.copy() 511 512 def configure_ap(self, **config_flags): 513 """Function to configure ap wireless settings.""" 514 # Turn radios on or off 515 if config_flags["status_toggled"]: 516 self.configure_radio_on_off() 517 # Configure radios 518 with BlockingBrowser(self.ap_settings["headless_browser"], 519 900) as browser: 520 # Visit URL 521 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 522 browser.visit_persistent(self.config_page_nologin, 523 BROWSER_WAIT_MED, 10, self.config_page) 524 525 # Update region, and power/bandwidth for each network 526 config_item = browser.find_by_name( 527 self.config_page_fields["region"]).first 528 config_item.select_by_text(self.ap_settings["region"]) 529 for key, value in self.config_page_fields.items(): 530 if "power" in key: 531 config_item = browser.find_by_name(value).first 532 config_item.select_by_text(self.ap_settings["{}_{}".format( 533 key[1], key[0])]) 534 elif "bandwidth" in key: 535 config_item = browser.find_by_name(value).first 536 try: 537 config_item.select_by_text( 538 self.bw_mode_text[self.ap_settings["{}_{}".format( 539 key[1], key[0])]]) 540 except AttributeError: 541 self.log.warning( 542 "Cannot select bandwidth. Keeping AP default.") 543 544 # Update security settings (passwords updated only if applicable) 545 for key, value in self.config_page_fields.items(): 546 if "security_type" in key: 547 browser.choose( 548 value, self.ap_settings["{}_{}".format(key[1], 549 key[0])]) 550 if self.ap_settings["{}_{}".format(key[1], 551 key[0])] == "WPA2-PSK": 552 config_item = browser.find_by_name( 553 self.config_page_fields[(key[0], 554 "password")]).first 555 config_item.fill(self.ap_settings["{}_{}".format( 556 "password", key[0])]) 557 558 # Update SSID and channel for each network 559 # NOTE: Update ordering done as such as workaround for R8000 560 # wherein channel and SSID get overwritten when some other 561 # variables are changed. However, region does have to be set before 562 # channel in all cases. 563 for key, value in self.config_page_fields.items(): 564 if "ssid" in key: 565 config_item = browser.find_by_name(value).first 566 config_item.fill(self.ap_settings["{}_{}".format( 567 key[1], key[0])]) 568 elif "channel" in key: 569 config_item = browser.find_by_name(value).first 570 try: 571 config_item.select(self.ap_settings["{}_{}".format( 572 key[1], key[0])]) 573 time.sleep(BROWSER_WAIT_SHORT) 574 except AttributeError: 575 self.log.warning( 576 "Cannot select channel. Keeping AP default.") 577 try: 578 alert = browser.get_alert() 579 alert.accept() 580 except: 581 pass 582 583 time.sleep(BROWSER_WAIT_SHORT) 584 browser.find_by_name("Apply").first.click() 585 time.sleep(BROWSER_WAIT_SHORT) 586 try: 587 alert = browser.get_alert() 588 alert.accept() 589 time.sleep(BROWSER_WAIT_SHORT) 590 except: 591 time.sleep(BROWSER_WAIT_SHORT) 592 browser.visit_persistent(self.config_page, BROWSER_WAIT_EXTRA_LONG, 593 10) 594 595 def configure_radio_on_off(self): 596 """Helper configuration function to turn radios on/off.""" 597 with BlockingBrowser(self.ap_settings["headless_browser"], 598 900) as browser: 599 # Visit URL 600 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 601 browser.visit_persistent(self.config_page_advanced, 602 BROWSER_WAIT_MED, 10) 603 604 # Turn radios on or off 605 for key, value in self.config_page_fields.items(): 606 if "status" in key: 607 config_item = browser.find_by_name(value).first 608 if self.ap_settings["{}_{}".format(key[1], key[0])]: 609 config_item.check() 610 else: 611 config_item.uncheck() 612 613 time.sleep(BROWSER_WAIT_SHORT) 614 browser.find_by_name("Apply").first.click() 615 time.sleep(BROWSER_WAIT_EXTRA_LONG) 616 browser.visit_persistent(self.config_page, BROWSER_WAIT_EXTRA_LONG, 617 10) 618 619 620class NetgearR7000NAAP(NetgearR7000AP): 621 """Class that implements Netgear R7000 NA AP.""" 622 623 def init_gui_data(self): 624 """Function to initialize data used while interacting with web GUI""" 625 super.init_gui_data() 626 self.region_map["11"] = "North America" 627 628 629class NetgearR7500AP(WifiRetailAP): 630 """Class that implements Netgear R7500 AP.""" 631 632 def __init__(self, ap_settings): 633 self.ap_settings = ap_settings.copy() 634 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 635 self.ap_settings["ip_address"])) 636 self.init_gui_data() 637 # Read and update AP settings 638 self.read_ap_settings() 639 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 640 self.update_ap_settings(ap_settings) 641 642 def init_gui_data(self): 643 """Function to initialize data used while interacting with web GUI""" 644 self.config_page = ("{protocol}://{username}:{password}@" 645 "{ip_address}:{port}/index.htm").format( 646 protocol=self.ap_settings["protocol"], 647 username=self.ap_settings["admin_username"], 648 password=self.ap_settings["admin_password"], 649 ip_address=self.ap_settings["ip_address"], 650 port=self.ap_settings["port"]) 651 self.config_page_advanced = ( 652 "{protocol}://{username}:{password}@" 653 "{ip_address}:{port}/adv_index.htm").format( 654 protocol=self.ap_settings["protocol"], 655 username=self.ap_settings["admin_username"], 656 password=self.ap_settings["admin_password"], 657 ip_address=self.ap_settings["ip_address"], 658 port=self.ap_settings["port"]) 659 self.networks = ["2G", "5G_1"] 660 self.channel_band_map = { 661 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 662 "5G_1": [ 663 36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120, 664 124, 128, 132, 136, 140, 149, 153, 157, 161, 165 665 ] 666 } 667 self.config_page_fields = { 668 "region": "WRegion", 669 ("2G", "status"): "enable_ap", 670 ("5G_1", "status"): "enable_ap_an", 671 ("2G", "ssid"): "ssid", 672 ("5G_1", "ssid"): "ssid_an", 673 ("2G", "channel"): "w_channel", 674 ("5G_1", "channel"): "w_channel_an", 675 ("2G", "bandwidth"): "opmode", 676 ("5G_1", "bandwidth"): "opmode_an", 677 ("2G", "security_type"): "security_type", 678 ("5G_1", "security_type"): "security_type_an", 679 ("2G", "password"): "passphrase", 680 ("5G_1", "password"): "passphrase_an" 681 } 682 self.region_map = { 683 "0": "Africa", 684 "1": "Asia", 685 "2": "Australia", 686 "3": "Canada", 687 "4": "Europe", 688 "5": "Israel", 689 "6": "Japan", 690 "7": "Korea", 691 "8": "Mexico", 692 "9": "South America", 693 "10": "United States", 694 "11": "China", 695 "12": "India", 696 "13": "Malaysia", 697 "14": "Middle East(Algeria/Syria/Yemen)", 698 "15": "Middle East(Iran/Labanon/Qatar)", 699 "16": "Middle East(Turkey/Egypt/Tunisia/Kuwait)", 700 "17": "Middle East(Saudi Arabia)", 701 "18": "Middle East(United Arab Emirates)", 702 "19": "Russia", 703 "20": "Singapore", 704 "21": "Taiwan" 705 } 706 self.bw_mode_text_2g = { 707 "11g": "Up to 54 Mbps", 708 "VHT20": "Up to 289 Mbps", 709 "VHT40": "Up to 600 Mbps" 710 } 711 self.bw_mode_text_5g = { 712 "VHT20": "Up to 347 Mbps", 713 "VHT40": "Up to 800 Mbps", 714 "VHT80": "Up to 1733 Mbps" 715 } 716 self.bw_mode_values = { 717 "1": "11g", 718 "2": "VHT20", 719 "3": "VHT40", 720 "7": "VHT20", 721 "8": "VHT40", 722 "9": "VHT80" 723 } 724 725 def read_ap_settings(self): 726 """Function to read ap wireless settings.""" 727 # Get radio status (on/off) 728 self.read_radio_on_off() 729 # Get radio configuration. Note that if both radios are off, the below 730 # code will result in an error 731 with BlockingBrowser(self.ap_settings["headless_browser"], 732 900) as browser: 733 browser.visit_persistent( 734 self.config_page, 735 BROWSER_WAIT_MED, 736 10, 737 check_for_element="wireless") 738 wireless_button = browser.find_by_id("wireless").first 739 wireless_button.click() 740 time.sleep(BROWSER_WAIT_MED) 741 742 with browser.get_iframe("formframe") as iframe: 743 for key, value in self.config_page_fields.items(): 744 if "bandwidth" in key: 745 config_item = iframe.find_by_name(value).first 746 self.ap_settings["{}_{}".format( 747 key[1], 748 key[0])] = self.bw_mode_values[config_item.value] 749 elif "region" in key: 750 config_item = iframe.find_by_name(value).first 751 self.ap_settings["region"] = self.region_map[ 752 config_item.value] 753 elif "password" in key: 754 try: 755 config_item = iframe.find_by_name(value).first 756 self.ap_settings["{}_{}".format( 757 key[1], key[0])] = config_item.value 758 self.ap_settings["{}_{}".format( 759 "security_type", key[0])] = "WPA2-PSK" 760 except: 761 self.ap_settings["{}_{}".format( 762 key[1], key[0])] = "defaultpassword" 763 self.ap_settings["{}_{}".format( 764 "security_type", key[0])] = "Disable" 765 elif ("channel" in key) or ("ssid" in key): 766 config_item = iframe.find_by_name(value).first 767 self.ap_settings["{}_{}".format( 768 key[1], key[0])] = config_item.value 769 else: 770 pass 771 return self.ap_settings.copy() 772 773 def configure_ap(self, **config_flags): 774 """Function to configure ap wireless settings.""" 775 # Turn radios on or off 776 if config_flags["status_toggled"]: 777 self.configure_radio_on_off() 778 # Configure radios 779 with BlockingBrowser(self.ap_settings["headless_browser"], 780 900) as browser: 781 browser.visit_persistent( 782 self.config_page, 783 BROWSER_WAIT_MED, 784 10, 785 check_for_element="wireless") 786 wireless_button = browser.find_by_id("wireless").first 787 wireless_button.click() 788 time.sleep(BROWSER_WAIT_MED) 789 790 with browser.get_iframe("formframe") as iframe: 791 # Update AP region. Must be done before channel setting 792 config_item = iframe.find_by_name( 793 self.config_page_fields["region"]).first 794 config_item.select_by_text(self.ap_settings["region"]) 795 # Update wireless settings for each network 796 for key, value in self.config_page_fields.items(): 797 if "ssid" in key: 798 config_item = iframe.find_by_name(value).first 799 config_item.fill(self.ap_settings["{}_{}".format( 800 key[1], key[0])]) 801 elif "channel" in key: 802 channel_string = "0" * (int(self.ap_settings[ 803 "{}_{}".format(key[1], key[0])]) < 10) + str( 804 self.ap_settings["{}_{}".format( 805 key[1], key[0])]) + "(DFS)" * (48 < int( 806 self.ap_settings["{}_{}".format( 807 key[1], key[0])]) < 149) 808 config_item = iframe.find_by_name(value).first 809 try: 810 config_item.select_by_text(channel_string) 811 except AttributeError: 812 self.log.warning( 813 "Cannot select channel. Keeping AP default.") 814 elif key == ("2G", "bandwidth"): 815 config_item = iframe.find_by_name(value).first 816 try: 817 config_item.select_by_text( 818 str(self.bw_mode_text_2g[self.ap_settings[ 819 "{}_{}".format(key[1], key[0])]])) 820 except AttributeError: 821 self.log.warning( 822 "Cannot select bandwidth. Keeping AP default.") 823 elif key == ("5G_1", "bandwidth"): 824 config_item = iframe.find_by_name(value).first 825 try: 826 config_item.select_by_text( 827 str(self.bw_mode_text_5g[self.ap_settings[ 828 "{}_{}".format(key[1], key[0])]])) 829 except AttributeError: 830 self.log.warning( 831 "Cannot select bandwidth. Keeping AP default.") 832 # Update passwords for WPA2-PSK protected networks 833 # (Must be done after security type is selected) 834 for key, value in self.config_page_fields.items(): 835 if "security_type" in key: 836 iframe.choose( 837 value, self.ap_settings["{}_{}".format( 838 key[1], key[0])]) 839 if self.ap_settings["{}_{}".format( 840 key[1], key[0])] == "WPA2-PSK": 841 config_item = iframe.find_by_name( 842 self.config_page_fields[(key[0], 843 "password")]).first 844 config_item.fill(self.ap_settings["{}_{}".format( 845 "password", key[0])]) 846 847 apply_button = iframe.find_by_name("Apply") 848 apply_button[0].click() 849 time.sleep(BROWSER_WAIT_SHORT) 850 try: 851 alert = browser.get_alert() 852 alert.accept() 853 except: 854 pass 855 time.sleep(BROWSER_WAIT_SHORT) 856 try: 857 alert = browser.get_alert() 858 alert.accept() 859 except: 860 pass 861 time.sleep(BROWSER_WAIT_SHORT) 862 time.sleep(BROWSER_WAIT_EXTRA_LONG) 863 browser.visit_persistent(self.config_page, BROWSER_WAIT_EXTRA_LONG, 864 10) 865 866 def configure_radio_on_off(self): 867 """Helper configuration function to turn radios on/off.""" 868 with BlockingBrowser(self.ap_settings["headless_browser"], 869 900) as browser: 870 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 871 browser.visit_persistent( 872 self.config_page_advanced, 873 BROWSER_WAIT_MED, 874 10, 875 check_for_element="advanced_bt") 876 advanced_button = browser.find_by_id("advanced_bt").first 877 advanced_button.click() 878 time.sleep(BROWSER_WAIT_MED) 879 wireless_button = browser.find_by_id("wladv").first 880 wireless_button.click() 881 time.sleep(BROWSER_WAIT_MED) 882 883 with browser.get_iframe("formframe") as iframe: 884 # Turn radios on or off 885 for key, value in self.config_page_fields.items(): 886 if "status" in key: 887 config_item = iframe.find_by_name(value).first 888 if self.ap_settings["{}_{}".format(key[1], key[0])]: 889 config_item.check() 890 else: 891 config_item.uncheck() 892 893 time.sleep(BROWSER_WAIT_SHORT) 894 browser.find_by_name("Apply").first.click() 895 time.sleep(BROWSER_WAIT_EXTRA_LONG) 896 browser.visit_persistent(self.config_page, 897 BROWSER_WAIT_EXTRA_LONG, 10) 898 899 def read_radio_on_off(self): 900 """Helper configuration function to read radio status.""" 901 with BlockingBrowser(self.ap_settings["headless_browser"], 902 900) as browser: 903 browser.visit_persistent(self.config_page, BROWSER_WAIT_MED, 10) 904 browser.visit_persistent( 905 self.config_page_advanced, 906 BROWSER_WAIT_MED, 907 10, 908 check_for_element="advanced_bt") 909 advanced_button = browser.find_by_id("advanced_bt").first 910 advanced_button.click() 911 time.sleep(BROWSER_WAIT_SHORT) 912 wireless_button = browser.find_by_id("wladv").first 913 wireless_button.click() 914 time.sleep(BROWSER_WAIT_MED) 915 916 with browser.get_iframe("formframe") as iframe: 917 # Turn radios on or off 918 for key, value in self.config_page_fields.items(): 919 if "status" in key: 920 config_item = iframe.find_by_name(value).first 921 self.ap_settings["{}_{}".format(key[1], key[0])] = int( 922 config_item.checked) 923 924 925class NetgearR7800AP(NetgearR7500AP): 926 """Class that implements Netgear R7800 AP. 927 928 Since most of the class' implementation is shared with the R7500, this 929 class inherits from NetgearR7500AP and simply redifines config parameters 930 """ 931 932 def __init__(self, ap_settings): 933 self.ap_settings = ap_settings.copy() 934 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 935 self.ap_settings["ip_address"])) 936 self.init_gui_data() 937 # Overwrite minor differences from R7500 AP 938 self.bw_mode_text_2g["VHT20"] = "Up to 347 Mbps" 939 # Read and update AP settings 940 self.read_ap_settings() 941 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 942 self.update_ap_settings(ap_settings) 943 944 945class NetgearR8000AP(NetgearR7000AP): 946 """Class that implements Netgear R8000 AP. 947 948 Since most of the class' implementation is shared with the R7000, this 949 class inherits from NetgearR7000AP and simply redifines config parameters 950 """ 951 952 def __init__(self, ap_settings): 953 self.ap_settings = ap_settings.copy() 954 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 955 self.ap_settings["ip_address"])) 956 self.init_gui_data() 957 # Overwrite minor differences from R7000 AP 958 self.config_page = ( 959 "{protocol}://{username}:{password}@" 960 "{ip_address}:{port}/WLG_wireless_dual_band_r8000.htm").format( 961 protocol=self.ap_settings["protocol"], 962 username=self.ap_settings["admin_username"], 963 password=self.ap_settings["admin_password"], 964 ip_address=self.ap_settings["ip_address"], 965 port=self.ap_settings["port"]) 966 self.config_page_nologin = ( 967 "{protocol}://{ip_address}:{port}/" 968 "WLG_wireless_dual_band_r8000.htm").format( 969 protocol=self.ap_settings["protocol"], 970 ip_address=self.ap_settings["ip_address"], 971 port=self.ap_settings["port"]) 972 self.config_page_advanced = ( 973 "{protocol}://{username}:{password}@" 974 "{ip_address}:{port}/WLG_adv_dual_band2_r8000.htm").format( 975 protocol=self.ap_settings["protocol"], 976 username=self.ap_settings["admin_username"], 977 password=self.ap_settings["admin_password"], 978 ip_address=self.ap_settings["ip_address"], 979 port=self.ap_settings["port"]) 980 self.networks = ["2G", "5G_1", "5G_2"] 981 self.channel_band_map = { 982 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 983 "5G_1": [36, 40, 44, 48], 984 "5G_2": [149, 153, 157, 161, 165] 985 } 986 self.config_page_fields = { 987 "region": "WRegion", 988 ("2G", "status"): "enable_ap", 989 ("5G_1", "status"): "enable_ap_an", 990 ("5G_2", "status"): "enable_ap_an_2", 991 ("2G", "ssid"): "ssid", 992 ("5G_1", "ssid"): "ssid_an", 993 ("5G_2", "ssid"): "ssid_an_2", 994 ("2G", "channel"): "w_channel", 995 ("5G_1", "channel"): "w_channel_an", 996 ("5G_2", "channel"): "w_channel_an_2", 997 ("2G", "bandwidth"): "opmode", 998 ("5G_1", "bandwidth"): "opmode_an", 999 ("5G_2", "bandwidth"): "opmode_an_2", 1000 ("2G", "security_type"): "security_type", 1001 ("5G_1", "security_type"): "security_type_an", 1002 ("5G_2", "security_type"): "security_type_an_2", 1003 ("2G", "password"): "passphrase", 1004 ("5G_1", "password"): "passphrase_an", 1005 ("5G_2", "password"): "passphrase_an_2" 1006 } 1007 # Read and update AP settings 1008 self.read_ap_settings() 1009 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 1010 self.update_ap_settings(ap_settings) 1011 1012 1013class NetgearR8500AP(NetgearR7000AP): 1014 """Class that implements Netgear R8500 AP. 1015 1016 Since most of the class' implementation is shared with the R7000, this 1017 class inherits from NetgearR7000AP and simply redifines config parameters 1018 """ 1019 1020 def __init__(self, ap_settings): 1021 self.ap_settings = ap_settings.copy() 1022 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 1023 self.ap_settings["ip_address"])) 1024 self.init_gui_data() 1025 # Overwrite minor differences from R8000 AP 1026 self.config_page = ( 1027 "{protocol}://{username}:{password}@" 1028 "{ip_address}:{port}/WLG_wireless_tri_band.htm").format( 1029 protocol=self.ap_settings["protocol"], 1030 username=self.ap_settings["admin_username"], 1031 password=self.ap_settings["admin_password"], 1032 ip_address=self.ap_settings["ip_address"], 1033 port=self.ap_settings["port"]) 1034 self.config_page_nologin = ( 1035 "{protocol}://{ip_address}:{port}/" 1036 "WLG_wireless_tri_band.htm").format( 1037 protocol=self.ap_settings["protocol"], 1038 ip_address=self.ap_settings["ip_address"], 1039 port=self.ap_settings["port"]) 1040 self.config_page_advanced = ( 1041 "{protocol}://{username}:{password}@" 1042 "{ip_address}:{port}/WLG_adv_tri_band2.htm").format( 1043 protocol=self.ap_settings["protocol"], 1044 username=self.ap_settings["admin_username"], 1045 password=self.ap_settings["admin_password"], 1046 ip_address=self.ap_settings["ip_address"], 1047 port=self.ap_settings["port"]) 1048 self.networks = ["2G", "5G_1", "5G_2"] 1049 self.channel_band_map = { 1050 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 1051 "5G_1": [36, 40, 44, 48], 1052 "5G_2": [149, 153, 157, 161, 165] 1053 } 1054 self.config_page_fields = { 1055 "region": "WRegion", 1056 ("2G", "status"): "enable_ap", 1057 ("5G_1", "status"): "enable_ap_an", 1058 ("5G_2", "status"): "enable_ap_an_2", 1059 ("2G", "ssid"): "ssid", 1060 ("5G_1", "ssid"): "ssid_an", 1061 ("5G_2", "ssid"): "ssid_an_2", 1062 ("2G", "channel"): "w_channel", 1063 ("5G_1", "channel"): "w_channel_an", 1064 ("5G_2", "channel"): "w_channel_an_2", 1065 ("2G", "bandwidth"): "opmode", 1066 ("5G_1", "bandwidth"): "opmode_an", 1067 ("5G_2", "bandwidth"): "opmode_an_2", 1068 ("2G", "security_type"): "security_type", 1069 ("5G_1", "security_type"): "security_type_an", 1070 ("5G_2", "security_type"): "security_type_an_2", 1071 ("2G", "password"): "passphrase", 1072 ("5G_1", "password"): "passphrase_an", 1073 ("5G_2", "password"): "passphrase_an_2" 1074 } 1075 self.bw_mode_text = { 1076 "11g": "Up to 54 Mbps", 1077 "VHT20": "Up to 433 Mbps", 1078 "VHT40": "Up to 1000 Mbps", 1079 "VHT80": "Up to 2165 Mbps" 1080 } 1081 # Read and update AP settings 1082 self.read_ap_settings() 1083 if not set(ap_settings.items()).issubset(self.ap_settings.items()): 1084 self.update_ap_settings(ap_settings) 1085 1086 1087class GoogleWifiAP(WifiRetailAP): 1088 """ Class that implements Google Wifi AP. 1089 1090 This class is a work in progress 1091 """ 1092 1093 def __init__(self, ap_settings): 1094 self.ap_settings = ap_settings.copy() 1095 self.log = logger.create_tagged_trace_logger("AccessPoint|{}".format( 1096 self.ap_settings["ssh_config"]["host"])) 1097 if self.ap_settings["status_2G"] and self.ap_settings["status_5G_1"]: 1098 raise ValueError("Error initializing Google Wifi AP. " 1099 "Only one interface can be enabled at a time.") 1100 self.channel_band_map = { 1101 "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11], 1102 "5G_1": [36, 40, 44, 48, 149, 153, 157, 161, 165] 1103 } 1104 self.BW_MODE_MAP = { 1105 "legacy": 20, 1106 "VHT20": 20, 1107 "VHT40": 40, 1108 "VHT80": 80 1109 } 1110 self.default_settings = { 1111 "region": "United States", 1112 "brand": "Google", 1113 "model": "Wifi", 1114 "status_2G": 0, 1115 "status_5G_1": 0, 1116 "ssid_2G": "GoogleWifi_2G", 1117 "ssid_5G_1": "GoogleWifi_5G", 1118 "channel_2G": 11, 1119 "channel_5G_1": 149, 1120 "bandwidth_2G": "VHT20", 1121 "bandwidth_5G_1": "VHT20", 1122 "power_2G": "auto", 1123 "power_5G_1": "auto", 1124 "security_type_2G": "Open", 1125 "security_type_5G_1": "Open", 1126 "subnet_2G": "192.168.1.0/24", 1127 "subnet_5G_1": "192.168.9.0/24", 1128 "password_2G": "password", 1129 "password_5G_1": "password" 1130 } 1131 1132 for setting in self.default_settings.keys(): 1133 if setting not in self.ap_settings: 1134 self.log.warning( 1135 "{0} not found during init. Setting {0} = {1}".format( 1136 setting, self.default_settings[setting])) 1137 self.ap_settings[setting] = self.default_settings[setting] 1138 init_settings = self.ap_settings.copy() 1139 init_settings["ap_subnet"] = { 1140 "2g": self.ap_settings["subnet_2G"], 1141 "5g": self.ap_settings["subnet_5G_1"] 1142 } 1143 self.access_point = access_point.AccessPoint(init_settings) 1144 self.configure_ap() 1145 1146 def read_ap_settings(self): 1147 """Function that reads current ap settings.""" 1148 return self.ap_settings.copy() 1149 1150 def update_ap_settings(self, dict_settings={}, **named_settings): 1151 """Function to update settings of existing AP. 1152 1153 Function copies arguments into ap_settings and calls configure_ap 1154 to apply them. 1155 1156 Args: 1157 dict_settings: single dictionary of settings to update 1158 **named_settings: named settings to update 1159 Note: dict and named_settings cannot contain the same settings. 1160 """ 1161 settings_to_update = dict(dict_settings, **named_settings) 1162 if len(settings_to_update) != len(dict_settings) + len(named_settings): 1163 raise KeyError("The following keys were passed twice: {}".format( 1164 (set(dict_settings.keys()).intersection( 1165 set(named_settings.keys()))))) 1166 if not set(settings_to_update.keys()).issubset( 1167 set(self.ap_settings.keys())): 1168 raise KeyError( 1169 "The following settings are invalid for this AP: {}".format( 1170 set(settings_to_update.keys()).difference( 1171 set(self.ap_settings.keys())))) 1172 1173 updating_2G = any(["2G" in x for x in settings_to_update.keys()]) 1174 updating_5G_1 = any(["5G_1" in x for x in settings_to_update.keys()]) 1175 if updating_2G and updating_5G_1: 1176 raise ValueError( 1177 "Error updating Google WiFi AP. " 1178 "One interface can be activated and updated at a time") 1179 elif updating_2G: 1180 # If updating an interface and not explicitly setting its status, 1181 # it is assumed that the interface is to be ENABLED and updated 1182 if "status_2G" not in settings_to_update: 1183 settings_to_update["status_2G"] = 1 1184 settings_to_update["status_5G_1"] = 0 1185 elif updating_5G_1: 1186 if "status_5G_1" not in settings_to_update: 1187 settings_to_update["status_2G"] = 0 1188 settings_to_update["status_5G_1"] = 1 1189 1190 updates_requested = False 1191 for setting, value in settings_to_update.items(): 1192 if self.ap_settings[setting] != value: 1193 self.ap_settings[setting] = value 1194 updates_requested = True 1195 1196 if updates_requested: 1197 self.configure_ap() 1198 1199 def configure_ap(self): 1200 """Function to configure Google Wifi.""" 1201 self.log.info("Stopping Google Wifi interfaces.") 1202 self.access_point.stop_all_aps() 1203 1204 if self.ap_settings["status_2G"] == 1: 1205 network = "2G" 1206 self.log.info("Bringing up 2.4 GHz network.") 1207 elif self.ap_settings["status_5G_1"] == 1: 1208 network = "5G_1" 1209 self.log.info("Bringing up 5 GHz network.") 1210 else: 1211 return 1212 1213 bss_settings = [] 1214 ssid = self.ap_settings["ssid_{}".format(network)] 1215 if "WPA" in self.ap_settings["security_type_{}".format(network)]: 1216 password = self.ap_settings["password_{}".format(network)] 1217 security = hostapd_security.Security( 1218 security_mode="wpa", password=password) 1219 else: 1220 security = hostapd_security.Security( 1221 security_mode=None, password=None) 1222 channel = int(self.ap_settings["channel_{}".format(network)]) 1223 bandwidth = self.BW_MODE_MAP[self.ap_settings["bandwidth_{}".format( 1224 network)]] 1225 config = hostapd_ap_preset.create_ap_preset( 1226 channel=channel, 1227 ssid=ssid, 1228 security=security, 1229 bss_settings=bss_settings, 1230 vht_bandwidth=bandwidth, 1231 profile_name='whirlwind', 1232 iface_wlan_2g=self.access_point.wlan_2g, 1233 iface_wlan_5g=self.access_point.wlan_5g) 1234 config_bridge = self.access_point.generate_bridge_configs(channel) 1235 brconfigs = bridge_interface.BridgeInterfaceConfigs( 1236 config_bridge[0], config_bridge[1], config_bridge[2]) 1237 self.access_point.bridge.startup(brconfigs) 1238 self.access_point.start_ap(config) 1239 self.set_power(network, self.ap_settings["power_{}".format(network)]) 1240 self.log.info("AP started on channel {} with SSID {}".format( 1241 channel, ssid)) 1242 1243 def set_power(self, network, power): 1244 """Function that sets network transmit power. 1245 1246 Args: 1247 network: string containing network identifier (2G, 5G_1, 5G_2) 1248 power: power level in dBm 1249 """ 1250 if power == "auto": 1251 power_string = "auto" 1252 else: 1253 if not float(power).is_integer(): 1254 self.log.info( 1255 "Power in dBm must be an integer. Setting to {}".format( 1256 int(power))) 1257 power = int(power) 1258 power_string = "fixed {}".format(int(power) * 100) 1259 1260 if "2G" in network: 1261 interface = self.access_point.wlan_2g 1262 self.ap_settings["power_2G"] = power 1263 elif "5G_1" in network: 1264 interface = self.access_point.wlan_5g 1265 self.ap_settings["power_5G_1"] = power 1266 self.access_point.ssh.run("iw dev {} set txpower {}".format( 1267 interface, power_string)) 1268 1269 def set_rate(self, 1270 network, 1271 mode=None, 1272 num_streams=None, 1273 rate=None, 1274 short_gi=0): 1275 """Function that sets rate. 1276 1277 Args: 1278 network: string containing network identifier (2G, 5G_1, 5G_2) 1279 mode: string indicating the WiFi standard to use 1280 num_streams: number of MIMO streams. used only for VHT 1281 rate: data rate of MCS index to use 1282 short_gi: boolean controlling the use of short guard interval 1283 """ 1284 if "2G" in network: 1285 interface = self.access_point.wlan_2g 1286 interface_short = "2.4" 1287 elif "5G_1" in network: 1288 interface = self.access_point.wlan_5g 1289 interface_short = "5" 1290 1291 if "legacy" in mode.lower(): 1292 cmd_string = "iw dev {0} set bitrates legacy-{1} {2} ht-mcs-{1} vht-mcs-{1}".format( 1293 interface, interface_short, rate) 1294 elif "vht" in mode.lower(): 1295 cmd_string = "iw dev {0} set bitrates legacy-{1} ht-mcs-{1} vht-mcs-{1} {2}:{3}".format( 1296 interface, interface_short, num_streams, rate) 1297 if short_gi: 1298 cmd_string = cmd_string + " sgi-interface_short" 1299 elif "ht" in mode.lower(): 1300 cmd_string = "iw dev {0} set bitrates legacy-{1} ht-mcs-{1} {2} vht-mcs-{1}".format( 1301 interface, interface_short, rate) 1302 self.access_point.ssh.run(cmd_string) 1303