Snap for 4793185 from e0f05231f9780315700a3ed1f2080f9189622ae0 to pi-release
Change-Id: I80a1e2785ca97625747c043ab7976ff897614ae7
diff --git a/acts/framework/acts/controllers/ap_lib/ap_get_interface.py b/acts/framework/acts/controllers/ap_lib/ap_get_interface.py
index 65c8938..b80add5 100644
--- a/acts/framework/acts/controllers/ap_lib/ap_get_interface.py
+++ b/acts/framework/acts/controllers/ap_lib/ap_get_interface.py
@@ -140,6 +140,10 @@
if wan:
return wan
+ output = self.ssh.run('ifconfig')
+ interfaces_all = output.stdout.split('\n')
+ logging.info("IFCONFIG output = %s" % interfaces_all)
+
raise ApInterfacesError('No WAN interface available')
def get_lan_interface(self):
diff --git a/acts/framework/acts/test_utils/wifi/wifi_retail_ap.py b/acts/framework/acts/test_utils/wifi/wifi_retail_ap.py
index 6e1ca7b..a05eca4 100644
--- a/acts/framework/acts/test_utils/wifi/wifi_retail_ap.py
+++ b/acts/framework/acts/test_utils/wifi/wifi_retail_ap.py
@@ -14,8 +14,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import splinter
+import time
+from acts.libs.proc import job
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
-from time import sleep
BROWSER_WAIT_SHORT = 1
BROWSER_WAIT_MED = 3
@@ -24,7 +25,7 @@
def create(configs):
- """ Factory method for retail AP class.
+ """Factory method for retail AP class.
Args:
configs: list of dicts containing ap settings. ap settings must contain
@@ -51,8 +52,55 @@
return
+def start_chrome_browser(headless, max_allowed_sessions, timeout):
+ """Method to start chrome browser for retail AP configuration
+
+ This function starts a chrome browser session to interface with the APs
+ web interface. The function attempts to maintain only one chromedriver
+ session by waiting until no chromedriver sessions are running on a machine.
+
+ Args:
+ headless: boolean controlling headless operation
+ max_allowed_sessions: maximum number of concurrent chrome sessions
+ timeout: maximum waiting time to launch chrome session
+ Returns:
+ browser: chrome browser session
+ Raises:
+ TimeoutError: raised when a browser session could not be started
+ withing the specified timeout
+ """
+ chrome_options = splinter.driver.webdriver.chrome.Options()
+ chrome_options.add_argument("--no-proxy-server")
+ chrome_options.add_argument("--no-sandbox")
+ chrome_options.add_argument("--allow-running-insecure-content")
+ chrome_options.add_argument("--ignore-certificate-errors")
+ chrome_capabilities = DesiredCapabilities.CHROME.copy()
+ chrome_capabilities["acceptSslCerts"] = True
+ chrome_capabilities["acceptInsecureCerts"] = True
+ if headless:
+ chrome_options.add_argument("--headless")
+ chrome_options.add_argument("--disable-gpu")
+
+ start_time = time.time()
+ end_time = start_time + timeout
+ while time.time() < end_time:
+ browsers_running = int(job.run('pgrep chromedriver | wc -l').stdout)
+ if browsers_running >= max_allowed_sessions:
+ time.sleep(BROWSER_WAIT_SHORT)
+ else:
+ try:
+ browser = splinter.Browser(
+ "chrome",
+ options=chrome_options,
+ desired_capabilities=chrome_capabilities)
+ return browser
+ except:
+ time.sleep(BROWSER_WAIT_SHORT)
+ raise TimeoutError("Could not start chrome browser in time.")
+
+
def visit_config_page(browser, url, page_load_timeout, num_tries):
- """ Method to visit Netgear AP webpages.
+ """Method to visit Netgear AP webpages.
This function visits a web page and checks the the resulting URL matches
the intended URL, i.e. no redirects have happened
@@ -67,7 +115,7 @@
try:
browser.visit(url)
except:
- browser.visit("https://www.google.com/")
+ browser.visit("about:blank")
if browser.url.split("/")[-1] == url.split("/")[-1]:
break
if idx == num_tries - 1:
@@ -75,7 +123,7 @@
class WifiRetailAP(object):
- """ Base class implementation for retail ap.
+ """Base class implementation for retail ap.
Base class provides functions whose implementation is shared by all aps.
If some functions such as set_power not supported by ap, checks will raise
@@ -86,7 +134,7 @@
raise NotImplementedError
def read_ap_settings(self):
- """ Function that reads current ap settings.
+ """Function that reads current ap settings.
Function implementation is AP dependent and thus base class raises exception
if function not implemented in child class.
@@ -94,7 +142,7 @@
raise NotImplementedError
def validate_ap_settings(self):
- """ Function to validate ap settings.
+ """Function to validate ap settings.
This function compares the actual ap settings read from the web GUI
with the assumed settings saved in the AP object. When called after AP
@@ -111,7 +159,7 @@
"Discrepancy in AP settings. Potential configuration error.")
def configure_ap(self):
- """ Function that configures ap based on values of ap_settings.
+ """Function that configures ap based on values of ap_settings.
Function implementation is AP dependent and thus base class raises exception
if function not implemented in child class.
@@ -119,7 +167,7 @@
raise NotImplementedError
def set_radio_on_off(self, network, status):
- """ Function that turns the radio on or off.
+ """Function that turns the radio on or off.
Args:
network: string containing network identifier (2G, 5G_1, 5G_2)
@@ -129,7 +177,7 @@
self.update_ap_settings(setting_to_update)
def set_ssid(self, network, ssid):
- """ Function that sets network SSID.
+ """Function that sets network SSID.
Args:
network: string containing network identifier (2G, 5G_1, 5G_2)
@@ -139,7 +187,7 @@
self.update_ap_settings(setting_to_update)
def set_channel(self, network, channel):
- """ Function that sets network channel.
+ """Function that sets network channel.
Args:
network: string containing network identifier (2G, 5G_1, 5G_2)
@@ -149,7 +197,7 @@
self.update_ap_settings(setting_to_update)
def set_bandwidth(self, network, bandwidth):
- """ Function that sets network bandwidth/mode.
+ """Function that sets network bandwidth/mode.
Args:
network: string containing network identifier (2G, 5G_1, 5G_2)
@@ -159,7 +207,7 @@
self.update_ap_settings(setting_to_update)
def set_power(self, network, power):
- """ Function that sets network transmit power.
+ """Function that sets network transmit power.
Args:
network: string containing network identifier (2G, 5G_1, 5G_2)
@@ -169,7 +217,7 @@
self.update_ap_settings(setting_to_update)
def set_security(self, network, security_type, *password):
- """ Function that sets network security setting and password.
+ """Function that sets network security setting and password.
Args:
network: string containing network identifier (2G, 5G_1, 5G_2)
@@ -188,7 +236,7 @@
self.update_ap_settings(setting_to_update)
def update_ap_settings(self, *dict_settings, **named_settings):
- """ Function to update settings of existing AP.
+ """Function to update settings of existing AP.
Function copies arguments into ap_settings and calls configure_retail_ap
to apply them.
@@ -225,54 +273,53 @@
self.configure_ap()
def band_lookup_by_channel(self, channel):
- """ Function that gives band name by channel number.
+ """Function that gives band name by channel number.
Args:
channel: channel number to lookup
Returns:
band: name of band which this channel belongs to on this ap
"""
- for key, value in self.CHANNEL_BAND_MAP.items():
+ for key, value in self.channel_band_map.items():
if channel in value:
return key
raise ValueError("Invalid channel passed in argument.")
class NetgearR7000AP(WifiRetailAP):
- """ Class that implements Netgear R7500 AP."""
+ """Class that implements Netgear R7500 AP."""
def __init__(self, ap_settings):
self.ap_settings = ap_settings.copy()
- self.CONFIG_PAGE = "{}://{}:{}@{}:{}/WLG_wireless_dual_band_r10.htm".format(
+ self.init_gui_data()
+ # Read and update AP settings
+ self.read_ap_settings()
+ if ap_settings.items() <= self.ap_settings.items():
+ return
+ else:
+ self.update_ap_settings(ap_settings)
+
+ def init_gui_data(self):
+ """Function to initialize data used while interacting with web GUI"""
+ self.config_page = "{}://{}:{}@{}:{}/WLG_wireless_dual_band_r10.htm".format(
self.ap_settings["protocol"], self.ap_settings["admin_username"],
self.ap_settings["admin_password"], self.ap_settings["ip_address"],
self.ap_settings["port"])
- self.CONFIG_PAGE_NOLOGIN = "{}://{}:{}/WLG_wireless_dual_band_r10.htm".format(
+ self.config_page_nologin = "{}://{}:{}/WLG_wireless_dual_band_r10.htm".format(
self.ap_settings["protocol"], self.ap_settings["ip_address"],
self.ap_settings["port"])
- self.CONFIG_PAGE_ADVANCED = "{}://{}:{}/WLG_adv_dual_band2.htm".format(
+ self.config_page_advanced = "{}://{}:{}/WLG_adv_dual_band2.htm".format(
self.ap_settings["protocol"], self.ap_settings["ip_address"],
self.ap_settings["port"])
- self.CHROME_OPTIONS = splinter.driver.webdriver.chrome.Options()
- self.CHROME_OPTIONS.add_argument("--no-proxy-server")
- self.CHROME_OPTIONS.add_argument("--no-sandbox")
- self.CHROME_OPTIONS.add_argument("--allow-running-insecure-content")
- self.CHROME_OPTIONS.add_argument("--ignore-certificate-errors")
- self.CHROME_CAPABILITIES = DesiredCapabilities.CHROME.copy()
- self.CHROME_CAPABILITIES["acceptSslCerts"] = True
- self.CHROME_CAPABILITIES["acceptInsecureCerts"] = True
- if self.ap_settings["headless_browser"]:
- self.CHROME_OPTIONS.add_argument("--headless")
- self.CHROME_OPTIONS.add_argument("--disable-gpu")
- self.NETWORKS = ["2G", "5G_1"]
- self.CHANNEL_BAND_MAP = {
+ self.networks = ["2G", "5G_1"]
+ self.channel_band_map = {
"2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
"5G_1": [
36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120,
124, 128, 132, 136, 140, 149, 153, 157, 161, 165
]
}
- self.REGION_MAP = {
+ self.region_map = {
"1": "Africa",
"2": "Asia",
"3": "Australia",
@@ -296,7 +343,7 @@
"23": "Singapore",
"24": "Taiwan"
}
- self.CONFIG_PAGE_FIELDS = {
+ self.config_page_fields = {
"region": "WRegion",
("2G", "status"): "enable_ap",
("5G_1", "status"): "enable_ap_an",
@@ -313,61 +360,54 @@
("2G", "password"): "passphrase",
("5G_1", "password"): "passphrase_an"
}
- self.BW_MODE_VALUES = {
+ self.bw_mode_values = {
"g and b": "11g",
"145Mbps": "VHT20",
"300Mbps": "VHT40",
"HT80": "VHT80"
}
- self.POWER_MODE_VALUES = {
+ self.power_mode_values = {
"1": "100%",
"2": "75%",
"3": "50%",
"4": "25%"
}
- self.BW_MODE_TEXT = {
+ self.bw_mode_text = {
"11g": "Up to 54 Mbps",
"VHT20": "Up to 289 Mbps",
"VHT40": "Up to 600 Mbps",
"VHT80": "Up to 1300 Mbps"
}
- self.read_ap_settings()
- if ap_settings.items() <= self.ap_settings.items():
- return
- else:
- self.update_ap_settings(ap_settings)
def read_ap_settings(self):
- """ Function to read ap settings."""
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
+ """Function to read ap settings."""
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
# Visit URL
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE_NOLOGIN,
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page_nologin,
BROWSER_WAIT_MED, 10)
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "status" in key:
- visit_config_page(browser, self.CONFIG_PAGE_ADVANCED,
+ visit_config_page(browser, self.config_page_advanced,
BROWSER_WAIT_MED, 10)
config_item = browser.find_by_name(value)
self.ap_settings["{}_{}".format(key[1], key[0])] = int(
config_item.first.checked)
- visit_config_page(browser, self.CONFIG_PAGE_NOLOGIN,
+ visit_config_page(browser, self.config_page_nologin,
BROWSER_WAIT_MED, 10)
else:
config_item = browser.find_by_name(value)
if "bandwidth" in key:
self.ap_settings["{}_{}".format(key[1], key[
- 0])] = self.BW_MODE_VALUES[config_item.first.value]
+ 0])] = self.bw_mode_values[config_item.first.value]
elif "power" in key:
self.ap_settings["{}_{}".format(
- key[1], key[0])] = self.POWER_MODE_VALUES[
+ key[1], key[0])] = self.power_mode_values[
config_item.first.value]
elif "region" in key:
- self.ap_settings["region"] = self.REGION_MAP[
+ self.ap_settings["region"] = self.region_map[
config_item.first.value]
elif "security_type" in key:
for item in config_item:
@@ -381,21 +421,19 @@
return self.ap_settings.copy()
def configure_ap(self):
- """ Function to configure ap wireless settings."""
+ """Function to configure ap wireless settings."""
# Turn radios on or off
self.configure_radio_on_off()
# Configure radios
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
# Visit URL
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE_NOLOGIN,
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page_nologin,
BROWSER_WAIT_MED, 10)
# Update region, and power/bandwidth for each network
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "power" in key:
config_item = browser.find_by_name(value).first
config_item.select_by_text(self.ap_settings["{}_{}".format(
@@ -406,18 +444,18 @@
elif "bandwidth" in key:
config_item = browser.find_by_name(value).first
config_item.select_by_text(
- self.BW_MODE_TEXT[self.ap_settings["{}_{}".format(
+ self.bw_mode_text[self.ap_settings["{}_{}".format(
key[1], key[0])]])
# Update security settings (passwords updated only if applicable)
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "security_type" in key:
browser.choose(value, self.ap_settings["{}_{}".format(
key[1], key[0])])
if self.ap_settings["{}_{}".format(key[1],
key[0])] == "WPA2-PSK":
config_item = browser.find_by_name(
- self.CONFIG_PAGE_FIELDS[(key[0],
+ self.config_page_fields[(key[0],
"password")]).first
config_item.fill(self.ap_settings["{}_{}".format(
"password", key[0])])
@@ -427,7 +465,7 @@
# wherein channel and SSID get overwritten when some other
# variables are changed. However, region does have to be set before
# channel in all cases.
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "ssid" in key:
config_item = browser.find_by_name(value).first
config_item.fill(self.ap_settings["{}_{}".format(
@@ -436,40 +474,38 @@
config_item = browser.find_by_name(value).first
config_item.select(self.ap_settings["{}_{}".format(
key[1], key[0])])
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
try:
alert = browser.get_alert()
alert.accept()
except:
pass
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
browser.find_by_name("Apply").first.click()
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
try:
alert = browser.get_alert()
alert.accept()
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
except:
- sleep(BROWSER_WAIT_SHORT)
- visit_config_page(browser, self.CONFIG_PAGE,
+ time.sleep(BROWSER_WAIT_SHORT)
+ visit_config_page(browser, self.config_page,
BROWSER_WAIT_EXTRA_LONG, 10)
- self.validate_ap_settings()
+ self.validate_ap_settings()
def configure_radio_on_off(self):
- """ Helper configuration function to turn radios on/off."""
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
+ """Helper configuration function to turn radios on/off."""
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
# Visit URL
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE_ADVANCED,
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page_advanced,
BROWSER_WAIT_MED, 10)
# Turn radios on or off
status_toggled = False
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "status" in key:
config_item = browser.find_by_name(value).first
current_status = int(config_item.checked)
@@ -482,48 +518,47 @@
config_item.uncheck()
if status_toggled:
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
browser.find_by_name("Apply").first.click()
- sleep(BROWSER_WAIT_EXTRA_LONG)
- visit_config_page(browser, self.CONFIG_PAGE,
+ time.sleep(BROWSER_WAIT_EXTRA_LONG)
+ visit_config_page(browser, self.config_page,
BROWSER_WAIT_EXTRA_LONG, 10)
class NetgearR7500AP(WifiRetailAP):
- """ Class that implements Netgear R7500 AP."""
+ """Class that implements Netgear R7500 AP."""
def __init__(self, ap_settings):
self.ap_settings = ap_settings.copy()
- self.CONFIG_PAGE = "{}://{}:{}@{}:{}/index.htm".format(
+ self.init_gui_data()
+ # Read and update AP settings
+ self.read_ap_settings()
+ if ap_settings.items() <= self.ap_settings.items():
+ return
+ else:
+ self.update_ap_settings(ap_settings)
+
+ def init_gui_data(self):
+ """Function to initialize data used while interacting with web GUI"""
+ self.config_page = "{}://{}:{}@{}:{}/index.htm".format(
self.ap_settings["protocol"], self.ap_settings["admin_username"],
self.ap_settings["admin_password"], self.ap_settings["ip_address"],
self.ap_settings["port"])
- self.CONFIG_PAGE_NOLOGIN = "{}://{}:{}/index.htm".format(
+ self.config_page_nologin = "{}://{}:{}/index.htm".format(
self.ap_settings["protocol"], self.ap_settings["ip_address"],
self.ap_settings["port"])
- self.CONFIG_PAGE_ADVANCED = "{}://{}:{}/adv_index.htm".format(
+ self.config_page_advanced = "{}://{}:{}/adv_index.htm".format(
self.ap_settings["protocol"], self.ap_settings["ip_address"],
self.ap_settings["port"])
- self.CHROME_OPTIONS = splinter.driver.webdriver.chrome.Options()
- self.CHROME_OPTIONS.add_argument("--no-proxy-server")
- self.CHROME_OPTIONS.add_argument("--no-sandbox")
- self.CHROME_OPTIONS.add_argument("--allow-running-insecure-content")
- self.CHROME_OPTIONS.add_argument("--ignore-certificate-errors")
- self.CHROME_CAPABILITIES = DesiredCapabilities.CHROME.copy()
- self.CHROME_CAPABILITIES["acceptSslCerts"] = True
- self.CHROME_CAPABILITIES["acceptInsecureCerts"] = True
- if self.ap_settings["headless_browser"]:
- self.CHROME_OPTIONS.add_argument("--headless")
- self.CHROME_OPTIONS.add_argument("--disable-gpu")
- self.NETWORKS = ["2G", "5G_1"]
- self.CHANNEL_BAND_MAP = {
+ self.networks = ["2G", "5G_1"]
+ self.channel_band_map = {
"2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
"5G_1": [
36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120,
124, 128, 132, 136, 140, 149, 153, 157, 161, 165
]
}
- self.CONFIG_PAGE_FIELDS = {
+ self.config_page_fields = {
"region": "WRegion",
("2G", "status"): "enable_ap",
("5G_1", "status"): "enable_ap_an",
@@ -538,7 +573,7 @@
("2G", "password"): "passphrase",
("5G_1", "password"): "passphrase_an"
}
- self.REGION_MAP = {
+ self.region_map = {
"0": "Africa",
"1": "Asia",
"2": "Australia",
@@ -562,17 +597,17 @@
"20": "Singapore",
"21": "Taiwan"
}
- self.BW_MODE_TEXT_2G = {
+ self.bw_mode_text_2g = {
"11g": "Up to 54 Mbps",
"VHT20": "Up to 289 Mbps",
"VHT40": "Up to 600 Mbps"
}
- self.BW_MODE_TEXT_5G = {
+ self.bw_mode_text_5g = {
"VHT20": "Up to 347 Mbps",
"VHT40": "Up to 800 Mbps",
"VHT80": "Up to 1733 Mbps"
}
- self.BW_MODE_VALUES = {
+ self.bw_mode_values = {
"1": "11g",
"2": "VHT20",
"3": "VHT40",
@@ -580,39 +615,32 @@
"8": "VHT40",
"9": "VHT80"
}
- self.read_ap_settings()
- if ap_settings.items() <= self.ap_settings.items():
- return
- else:
- self.update_ap_settings(ap_settings)
def read_ap_settings(self):
- """ Function to read ap wireless settings."""
+ """Function to read ap wireless settings."""
# Get radio status (on/off)
self.read_radio_on_off()
# Get radio configuration. Note that if both radios are off, the below
# code will result in an error
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- sleep(BROWSER_WAIT_SHORT)
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ time.sleep(BROWSER_WAIT_SHORT)
wireless_button = browser.find_by_id("wireless").first
wireless_button.click()
- sleep(BROWSER_WAIT_MED)
+ time.sleep(BROWSER_WAIT_MED)
with browser.get_iframe("formframe") as iframe:
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "bandwidth" in key:
config_item = iframe.find_by_name(value).first
self.ap_settings["{}_{}".format(
key[1],
- key[0])] = self.BW_MODE_VALUES[config_item.value]
+ key[0])] = self.bw_mode_values[config_item.value]
elif "region" in key:
config_item = iframe.find_by_name(value).first
- self.ap_settings["region"] = self.REGION_MAP[
+ self.ap_settings["region"] = self.region_map[
config_item.value]
elif "password" in key:
try:
@@ -635,29 +663,27 @@
return self.ap_settings.copy()
def configure_ap(self):
- """ Function to configure ap wireless settings."""
+ """Function to configure ap wireless settings."""
# Turn radios on or off
self.configure_radio_on_off()
# Configure radios
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- sleep(BROWSER_WAIT_SHORT)
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ time.sleep(BROWSER_WAIT_SHORT)
wireless_button = browser.find_by_id("wireless").first
wireless_button.click()
- sleep(BROWSER_WAIT_MED)
+ time.sleep(BROWSER_WAIT_MED)
with browser.get_iframe("formframe") as iframe:
# Update AP region. Must be done before channel setting
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "region" in key:
config_item = iframe.find_by_name(value).first
config_item.select_by_text(self.ap_settings["region"])
# Update wireless settings for each network
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "ssid" in key:
config_item = iframe.find_by_name(value).first
config_item.fill(self.ap_settings["{}_{}".format(
@@ -674,69 +700,67 @@
elif key == ("2G", "bandwidth"):
config_item = iframe.find_by_name(value).first
config_item.select_by_text(
- str(self.BW_MODE_TEXT_2G[self.ap_settings[
+ str(self.bw_mode_text_2g[self.ap_settings[
"{}_{}".format(key[1], key[0])]]))
elif key == ("5G_1", "bandwidth"):
config_item = iframe.find_by_name(value).first
config_item.select_by_text(
- str(self.BW_MODE_TEXT_5G[self.ap_settings[
+ str(self.bw_mode_text_5g[self.ap_settings[
"{}_{}".format(key[1], key[0])]]))
# Update passwords for WPA2-PSK protected networks
# (Must be done after security type is selected)
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "security_type" in key:
iframe.choose(value, self.ap_settings["{}_{}".format(
key[1], key[0])])
if self.ap_settings["{}_{}".format(
key[1], key[0])] == "WPA2-PSK":
config_item = iframe.find_by_name(
- self.CONFIG_PAGE_FIELDS[(key[0],
+ self.config_page_fields[(key[0],
"password")]).first
config_item.fill(self.ap_settings["{}_{}".format(
"password", key[0])])
apply_button = iframe.find_by_name("Apply")
apply_button[0].click()
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
try:
alert = browser.get_alert()
alert.accept()
except:
pass
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
try:
alert = browser.get_alert()
alert.accept()
except:
pass
- sleep(BROWSER_WAIT_SHORT)
- sleep(BROWSER_WAIT_EXTRA_LONG)
- visit_config_page(browser, self.CONFIG_PAGE,
+ time.sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_EXTRA_LONG)
+ visit_config_page(browser, self.config_page,
BROWSER_WAIT_EXTRA_LONG, 10)
- self.validate_ap_settings()
+ self.validate_ap_settings()
def configure_radio_on_off(self):
- """ Helper configuration function to turn radios on/off."""
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE_ADVANCED,
+ """Helper configuration function to turn radios on/off."""
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page_advanced,
BROWSER_WAIT_MED, 10)
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
wireless_button = browser.find_by_id("advanced_bt").first
wireless_button.click()
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
wireless_button = browser.find_by_id("wladv").first
wireless_button.click()
- sleep(BROWSER_WAIT_MED)
+ time.sleep(BROWSER_WAIT_MED)
with browser.get_iframe("formframe") as iframe:
# Turn radios on or off
status_toggled = False
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "status" in key:
config_item = iframe.find_by_name(value).first
current_status = int(config_item.checked)
@@ -750,31 +774,29 @@
config_item.uncheck()
if status_toggled:
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
browser.find_by_name("Apply").first.click()
- sleep(BROWSER_WAIT_EXTRA_LONG)
- visit_config_page(browser, self.CONFIG_PAGE,
+ time.sleep(BROWSER_WAIT_EXTRA_LONG)
+ visit_config_page(browser, self.config_page,
BROWSER_WAIT_EXTRA_LONG, 10)
def read_radio_on_off(self):
- """ Helper configuration function to read radio status."""
- with splinter.Browser(
- "chrome",
- options=self.CHROME_OPTIONS,
- desired_capabilities=self.CHROME_CAPABILITIES) as browser:
- visit_config_page(browser, self.CONFIG_PAGE, BROWSER_WAIT_MED, 10)
- visit_config_page(browser, self.CONFIG_PAGE_ADVANCED,
+ """Helper configuration function to read radio status."""
+ with start_chrome_browser(self.ap_settings["headless_browser"], 1,
+ 600) as browser:
+ visit_config_page(browser, self.config_page, BROWSER_WAIT_MED, 10)
+ visit_config_page(browser, self.config_page_advanced,
BROWSER_WAIT_MED, 10)
wireless_button = browser.find_by_id("advanced_bt").first
wireless_button.click()
- sleep(BROWSER_WAIT_SHORT)
+ time.sleep(BROWSER_WAIT_SHORT)
wireless_button = browser.find_by_id("wladv").first
wireless_button.click()
- sleep(BROWSER_WAIT_MED)
+ time.sleep(BROWSER_WAIT_MED)
with browser.get_iframe("formframe") as iframe:
# Turn radios on or off
- for key, value in self.CONFIG_PAGE_FIELDS.items():
+ for key, value in self.config_page_fields.items():
if "status" in key:
config_item = iframe.find_by_name(value).first
self.ap_settings["{}_{}".format(key[1], key[0])] = int(
@@ -782,96 +804,18 @@
class NetgearR7800AP(NetgearR7500AP):
- """ Class that implements Netgear R7800 AP."""
+ """Class that implements Netgear R7800 AP.
+
+ Since most of the class' implementation is shared with the R7500, this
+ class inherits from NetgearR7500AP and simply redifines config parameters
+ """
def __init__(self, ap_settings):
self.ap_settings = ap_settings.copy()
- self.CONFIG_PAGE = "{}://{}:{}@{}:{}/index.htm".format(
- self.ap_settings["protocol"], self.ap_settings["admin_username"],
- self.ap_settings["admin_password"], self.ap_settings["ip_address"],
- self.ap_settings["port"])
- self.CONFIG_PAGE_NOLOGIN = "{}://{}:{}/index.htm".format(
- self.ap_settings["protocol"], self.ap_settings["ip_address"],
- self.ap_settings["port"])
- self.CONFIG_PAGE_ADVANCED = "{}://{}:{}/adv_index.htm".format(
- self.ap_settings["protocol"], self.ap_settings["ip_address"],
- self.ap_settings["port"])
- self.CHROME_OPTIONS = splinter.driver.webdriver.chrome.Options()
- self.CHROME_OPTIONS.add_argument("--no-proxy-server")
- self.CHROME_OPTIONS.add_argument("--no-sandbox")
- self.CHROME_OPTIONS.add_argument("--allow-running-insecure-content")
- self.CHROME_OPTIONS.add_argument("--ignore-certificate-errors")
- self.CHROME_CAPABILITIES = DesiredCapabilities.CHROME.copy()
- self.CHROME_CAPABILITIES["acceptSslCerts"] = True
- self.CHROME_CAPABILITIES["acceptInsecureCerts"] = True
- if self.ap_settings["headless_browser"]:
- self.CHROME_OPTIONS.add_argument("--headless")
- self.CHROME_OPTIONS.add_argument("--disable-gpu")
- self.NETWORKS = ["2G", "5G_1"]
- self.CHANNEL_BAND_MAP = {
- "2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
- "5G_1": [
- 36, 40, 44, 48, 52, 56, 60, 64, 100, 104, 108, 112, 116, 120,
- 124, 128, 132, 136, 140, 149, 153, 157, 161, 165
- ]
- }
- self.CONFIG_PAGE_FIELDS = {
- "region": "WRegion",
- ("2G", "status"): "enable_ap",
- ("5G_1", "status"): "enable_ap_an",
- ("2G", "ssid"): "ssid",
- ("5G_1", "ssid"): "ssid_an",
- ("2G", "channel"): "w_channel",
- ("5G_1", "channel"): "w_channel_an",
- ("2G", "bandwidth"): "opmode",
- ("5G_1", "bandwidth"): "opmode_an",
- ("2G", "security_type"): "security_type",
- ("5G_1", "security_type"): "security_type_an",
- ("2G", "password"): "passphrase",
- ("5G_1", "password"): "passphrase_an"
- }
- self.REGION_MAP = {
- "0": "Africa",
- "1": "Asia",
- "2": "Australia",
- "3": "Canada",
- "4": "Europe",
- "5": "Israel",
- "6": "Japan",
- "7": "Korea",
- "8": "Mexico",
- "9": "South America",
- "10": "United States",
- "11": "China",
- "12": "India",
- "13": "Malaysia",
- "14": "Middle East(Algeria/Syria/Yemen)",
- "15": "Middle East(Iran/Labanon/Qatar)",
- "16": "Middle East(Turkey/Egypt/Tunisia/Kuwait)",
- "17": "Middle East(Saudi Arabia)",
- "18": "Middle East(United Arab Emirates)",
- "19": "Russia",
- "20": "Singapore",
- "21": "Taiwan"
- }
- self.BW_MODE_TEXT_2G = {
- "11g": "Up to 54 Mbps",
- "VHT20": "Up to 347 Mbps",
- "VHT40": "Up to 600 Mbps"
- }
- self.BW_MODE_TEXT_5G = {
- "VHT20": "Up to 347 Mbps",
- "VHT40": "Up to 800 Mbps",
- "VHT80": "Up to 1733 Mbps"
- }
- self.BW_MODE_VALUES = {
- "1": "11g",
- "2": "VHT20",
- "3": "VHT40",
- "7": "VHT20",
- "8": "VHT40",
- "9": "VHT80"
- }
+ self.init_gui_data()
+ # Overwrite minor differences from R7500 AP
+ self.bw_mode_text_2g["VHT20"] = "Up to 347 Mbps"
+ # Read and update AP settings
self.read_ap_settings()
if ap_settings.items() <= self.ap_settings.items():
return
@@ -880,7 +824,7 @@
class NetgearR8000AP(NetgearR7000AP):
- """ Class that implements Netgear R8000 AP.
+ """Class that implements Netgear R8000 AP.
Since most of the class' implementation is shared with the R7000, this
class inherits from NetgearR7000AP and simply redifines config parameters
@@ -888,50 +832,25 @@
def __init__(self, ap_settings):
self.ap_settings = ap_settings.copy()
- self.CONFIG_PAGE = "http://{}:{}@{}/WLG_wireless_dual_band_r8000.htm".format(
- self.ap_settings["admin_username"],
- self.ap_settings["admin_password"], self.ap_settings["ip_address"])
- self.CONFIG_PAGE_NOLOGIN = "http://{}/WLG_wireless_dual_band_r8000.htm".format(
- self.ap_settings["ip_address"])
- self.CONFIG_PAGE_ADVANCED = "http://{}/WLG_adv_dual_band2_r8000.htm".format(
- self.ap_settings["ip_address"])
- self.CHROME_OPTIONS = splinter.driver.webdriver.chrome.Options()
- self.CHROME_OPTIONS.add_argument("--no-proxy-server")
- self.CHROME_OPTIONS.add_argument("--no-sandbox")
- if self.ap_settings["headless_browser"]:
- self.CHROME_OPTIONS.add_argument("--headless")
- self.CHROME_OPTIONS.add_argument("--disable-gpu")
- self.NETWORKS = ["2G", "5G_1", "5G_2"]
- self.CHANNEL_BAND_MAP = {
+ self.init_gui_data()
+ # Overwrite minor differences from R7000 AP
+ self.config_page = "{}://{}:{}@{}:{}/WLG_wireless_dual_band_r8000.htm".format(
+ self.ap_settings["protocol"], self.ap_settings["admin_username"],
+ self.ap_settings["admin_password"], self.ap_settings["ip_address"],
+ self.ap_settings["port"])
+ self.config_page_nologin = "{}://{}:{}/WLG_wireless_dual_band_r8000.htm".format(
+ self.ap_settings["protocol"], self.ap_settings["ip_address"],
+ self.ap_settings["port"])
+ self.config_page_advanced = "{}://{}:{}/WLG_adv_dual_band2_r8000.htm".format(
+ self.ap_settings["protocol"], self.ap_settings["ip_address"],
+ self.ap_settings["port"])
+ self.networks = ["2G", "5G_1", "5G_2"]
+ self.channel_band_map = {
"2G": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
"5G_1": [36, 40, 44, 48],
"5G_2": [149, 153, 157, 161, 165]
}
- self.REGION_MAP = {
- "1": "Africa",
- "2": "Asia",
- "3": "Australia",
- "4": "Canada",
- "5": "Europe",
- "6": "Israel",
- "7": "Japan",
- "8": "Korea",
- "9": "Mexico",
- "10": "South America",
- "11": "United States",
- "12": "Middle East(Algeria/Syria/Yemen)",
- "14": "Russia",
- "16": "China",
- "17": "India",
- "18": "Malaysia",
- "19": "Middle East(Iran/Labanon/Qatar)",
- "20": "Middle East(Turkey/Egypt/Tunisia/Kuwait)",
- "21": "Middle East(Saudi Arabia)",
- "22": "Middle East(United Arab Emirates)",
- "23": "Singapore",
- "24": "Taiwan"
- }
- self.CONFIG_PAGE_FIELDS = {
+ self.config_page_fields = {
"region": "WRegion",
("2G", "status"): "enable_ap",
("5G_1", "status"): "enable_ap_an",
@@ -952,6 +871,7 @@
("5G_1", "password"): "passphrase_an",
("5G_2", "password"): "passphrase_an_2"
}
+ # Read and update AP settings
self.read_ap_settings()
if ap_settings.items() <= self.ap_settings.items():
return
diff --git a/acts/tests/google/wifi/rtt/functional/AwareDiscoveryWithRangingTest.py b/acts/tests/google/wifi/rtt/functional/AwareDiscoveryWithRangingTest.py
index 8c073b9..b2120f9 100644
--- a/acts/tests/google/wifi/rtt/functional/AwareDiscoveryWithRangingTest.py
+++ b/acts/tests/google/wifi/rtt/functional/AwareDiscoveryWithRangingTest.py
@@ -23,6 +23,8 @@
from acts.test_utils.wifi.aware import aware_const as aconsts
from acts.test_utils.wifi.aware import aware_test_utils as autils
from acts.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest
+from acts.test_utils.wifi.rtt import rtt_const as rconsts
+from acts.test_utils.wifi.rtt import rtt_test_utils as rutils
from acts.test_utils.wifi.rtt.RttBaseTest import RttBaseTest
@@ -1459,4 +1461,106 @@
event = autils.wait_for_event(dut1, autils.decorate_event(
aconsts.SESSION_CB_ON_SERVICE_DISCOVERED, ff_s_id))
asserts.assert_true(aconsts.SESSION_CB_KEY_DISTANCE_MM in event["data"],
- "Discovery with ranging for FF expected!")
\ No newline at end of file
+ "Discovery with ranging for FF expected!")
+
+
+ def test_discovery_direct_concurrency(self):
+ """Verify the behavior of Wi-Fi Aware Ranging used as part of discovery and
+ as direct ranging to a peer device.
+
+ Process:
+ - Start YYY service with ranging in-range
+ - Start XXX service with ranging out-of-range
+ - Start performing direct Ranging
+ - While above going on update XXX to be in-range
+ - Keep performing direct Ranging in context of YYY
+ - Stop direct Ranging and look for XXX to discover
+ """
+ dut1 = self.android_devices[0]
+ dut1.pretty_name = "DUT1"
+ dut2 = self.android_devices[1]
+ dut2.pretty_name = "DUT2"
+
+ # DUTs: attach and wait for confirmation
+ dut1_id = dut1.droid.wifiAwareAttach(False)
+ autils.wait_for_event(dut1, aconsts.EVENT_CB_ON_ATTACHED)
+ time.sleep(self.device_startup_offset)
+ dut2_id = dut2.droid.wifiAwareAttach(True)
+ event = autils.wait_for_event(dut2, aconsts.EVENT_CB_ON_IDENTITY_CHANGED)
+ dut2_mac = event['data']['mac']
+
+ # DUT1: publishers bring-up
+ xxx_p_id = dut1.droid.wifiAwarePublish(dut1_id, autils.add_ranging_to_pub(
+ autils.create_discovery_config("XXX", aconsts.PUBLISH_TYPE_UNSOLICITED),
+ enable_ranging=True), True)
+ autils.wait_for_event(dut1, autils.decorate_event(
+ aconsts.SESSION_CB_ON_PUBLISH_STARTED, xxx_p_id))
+ yyy_p_id = dut1.droid.wifiAwarePublish(dut1_id, autils.add_ranging_to_pub(
+ autils.create_discovery_config("YYY", aconsts.PUBLISH_TYPE_UNSOLICITED),
+ enable_ranging=True), True)
+ autils.wait_for_event(dut1, autils.decorate_event(
+ aconsts.SESSION_CB_ON_PUBLISH_STARTED, yyy_p_id))
+
+ # DUT2: subscribers bring-up
+ xxx_s_id = dut2.droid.wifiAwareSubscribe(dut2_id, autils.add_ranging_to_sub(
+ autils.create_discovery_config("XXX", aconsts.SUBSCRIBE_TYPE_PASSIVE),
+ min_distance_mm=1000000, max_distance_mm=1000001), True)
+ autils.wait_for_event(dut2, autils.decorate_event(
+ aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED, xxx_s_id))
+ yyy_s_id = dut2.droid.wifiAwareSubscribe(dut2_id, autils.add_ranging_to_sub(
+ autils.create_discovery_config("YYY", aconsts.SUBSCRIBE_TYPE_PASSIVE),
+ min_distance_mm=None, max_distance_mm=1000000), True)
+ autils.wait_for_event(dut2, autils.decorate_event(
+ aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED, yyy_s_id))
+
+ # Service discovery: YYY (with range info), but no XXX
+ event = autils.wait_for_event(dut2, autils.decorate_event(
+ aconsts.SESSION_CB_ON_SERVICE_DISCOVERED, yyy_s_id))
+ asserts.assert_true(aconsts.SESSION_CB_KEY_DISTANCE_MM in event["data"],
+ "Discovery with ranging for YYY expected!")
+ yyy_peer_id_on_sub = event['data'][aconsts.SESSION_CB_KEY_PEER_ID]
+
+ autils.fail_on_event(dut2, autils.decorate_event(
+ aconsts.SESSION_CB_ON_SERVICE_DISCOVERED, xxx_s_id))
+
+ # Direct ranging
+ results21 = []
+ for iter in range(10):
+ id = dut2.droid.wifiRttStartRangingToAwarePeerId(yyy_peer_id_on_sub)
+ event = autils.wait_for_event(dut2, rutils.decorate_event(
+ rconsts.EVENT_CB_RANGING_ON_RESULT, id))
+ results21.append(event["data"][rconsts.EVENT_CB_RANGING_KEY_RESULTS][0])
+
+ time.sleep(5) # while switching roles
+
+ results12 = []
+ for iter in range(10):
+ id = dut1.droid.wifiRttStartRangingToAwarePeerMac(dut2_mac)
+ event = autils.wait_for_event(dut1, rutils.decorate_event(
+ rconsts.EVENT_CB_RANGING_ON_RESULT, id))
+ results12.append(event["data"][rconsts.EVENT_CB_RANGING_KEY_RESULTS][0])
+
+ stats = [rutils.extract_stats(results12, 0, 0, 0),
+ rutils.extract_stats(results21, 0, 0, 0)]
+
+ # Update XXX to be within range
+ dut2.droid.wifiAwareUpdateSubscribe(xxx_s_id, autils.add_ranging_to_sub(
+ autils.create_discovery_config("XXX", aconsts.SUBSCRIBE_TYPE_PASSIVE),
+ min_distance_mm=None, max_distance_mm=1000000))
+ autils.wait_for_event(dut2, autils.decorate_event(
+ aconsts.SESSION_CB_ON_SESSION_CONFIG_UPDATED, xxx_s_id))
+
+ # Expect discovery on XXX - wait until discovery with ranging:
+ # - 0 or more: without ranging info (due to concurrency limitations)
+ # - 1 or more: with ranging (once concurrency limitation relieved)
+ num_events = 0
+ while True:
+ event = autils.wait_for_event(dut2, autils.decorate_event(
+ aconsts.SESSION_CB_ON_SERVICE_DISCOVERED, xxx_s_id))
+ if aconsts.SESSION_CB_KEY_DISTANCE_MM in event["data"]:
+ break
+ num_events = num_events + 1
+ asserts.assert_true(num_events < 10, # arbitrary safety valve
+ "Way too many discovery events without ranging!")
+
+ asserts.explicit_pass("Discovery/Direct RTT Concurrency Pass", extras=stats)
\ No newline at end of file