| #!/usr/bin/env python3 |
| # |
| # Copyright 2017 - Google |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import base64 |
| import json |
| import queue |
| import re |
| import statistics |
| import time |
| from acts import asserts |
| |
| from acts_contrib.test_utils.net import connectivity_const as cconsts |
| from acts_contrib.test_utils.net import socket_test_utils as sutils |
| from acts_contrib.test_utils.wifi.aware import aware_const as aconsts |
| |
| # arbitrary timeout for events |
| EVENT_TIMEOUT = 10 |
| |
| # semi-arbitrary timeout for network formation events. Based on framework |
| # timeout for NDP (NAN data-path) negotiation to be completed. |
| EVENT_NDP_TIMEOUT = 20 |
| |
| # number of second to 'reasonably' wait to make sure that devices synchronize |
| # with each other - useful for OOB test cases, where the OOB discovery would |
| # take some time |
| WAIT_FOR_CLUSTER = 5 |
| |
| |
| def decorate_event(event_name, id): |
| return '%s_%d' % (event_name, id) |
| |
| |
| def wait_for_event(ad, event_name, timeout=EVENT_TIMEOUT): |
| """Wait for the specified event or timeout. |
| |
| Args: |
| ad: The android device |
| event_name: The event to wait on |
| timeout: Number of seconds to wait |
| Returns: |
| The event (if available) |
| """ |
| prefix = '' |
| if hasattr(ad, 'pretty_name'): |
| prefix = '[%s] ' % ad.pretty_name |
| try: |
| event = ad.ed.pop_event(event_name, timeout) |
| ad.log.info('%s%s: %s', prefix, event_name, event['data']) |
| return event |
| except queue.Empty: |
| ad.log.info('%sTimed out while waiting for %s', prefix, event_name) |
| asserts.fail(event_name) |
| |
| def _filter_callbacks(event, expected_kv): |
| """ |
| Helper method to use in |fail_on_event_with_keys| and |
| |wait_for_event_with_keys| |
| """ |
| for expected_k, expected_v in expected_kv: |
| actual_v = event['data'][expected_k] |
| if isinstance(expected_v, dict) and isinstance(actual_v, dict): |
| # |expected_v| not a subset of |actual_v| |
| if not(expected_v.items() <= actual_v.items()): |
| return False |
| else: |
| if actual_v != expected_v: |
| return False |
| return True |
| |
| |
| def wait_for_event_with_keys(ad, |
| event_name, |
| timeout=EVENT_TIMEOUT, |
| *keyvalues): |
| """Wait for the specified event contain the key/value pairs or timeout |
| |
| Args: |
| ad: The android device |
| event_name: The event to wait on |
| timeout: Number of seconds to wait |
| keyvalues: Expected (key, value) pairs. If the value for a key is a dict, |
| then this will perform subset matching for that key. |
| Returns: |
| The event (if available) |
| """ |
| prefix = '' |
| if hasattr(ad, 'pretty_name'): |
| prefix = '[%s] ' % ad.pretty_name |
| try: |
| event = ad.ed.wait_for_event(event_name, _filter_callbacks, timeout, |
| keyvalues) |
| ad.log.info('%s%s: %s', prefix, event_name, event['data']) |
| return event |
| except queue.Empty: |
| ad.log.info('%sTimed out while waiting for %s (%s)', prefix, |
| event_name, keyvalues) |
| asserts.fail(event_name) |
| |
| |
| def fail_on_event(ad, event_name, timeout=EVENT_TIMEOUT): |
| """Wait for a timeout period and looks for the specified event - fails if it |
| is observed. |
| |
| Args: |
| ad: The android device |
| event_name: The event to wait for (and fail on its appearance) |
| """ |
| prefix = '' |
| if hasattr(ad, 'pretty_name'): |
| prefix = '[%s] ' % ad.pretty_name |
| try: |
| event = ad.ed.pop_event(event_name, timeout) |
| ad.log.info('%sReceived unwanted %s: %s', prefix, event_name, |
| event['data']) |
| asserts.fail(event_name, extras=event) |
| except queue.Empty: |
| ad.log.info('%s%s not seen (as expected)', prefix, event_name) |
| return |
| |
| |
| def fail_on_event_with_keys(ad, event_name, timeout=EVENT_TIMEOUT, *keyvalues): |
| """Wait for a timeout period and looks for the specified event which contains |
| the key/value pairs - fails if it is observed. |
| |
| Args: |
| ad: The android device |
| event_name: The event to wait on |
| timeout: Number of seconds to wait |
| keyvalues: Expected (key, value) pairs. If the value for a key is a dict, |
| then this will perform subset matching for that key. |
| """ |
| prefix = '' |
| if hasattr(ad, 'pretty_name'): |
| prefix = '[%s] ' % ad.pretty_name |
| try: |
| event = ad.ed.wait_for_event(event_name, _filter_callbacks, timeout, |
| keyvalues) |
| ad.log.info('%sReceived unwanted %s: %s', prefix, event_name, |
| event['data']) |
| asserts.fail(event_name, extras=event) |
| except queue.Empty: |
| ad.log.info('%s%s (%s) not seen (as expected)', prefix, event_name, |
| keyvalues) |
| return |
| |
| |
| def verify_no_more_events(ad, timeout=EVENT_TIMEOUT): |
| """Verify that there are no more events in the queue. |
| """ |
| prefix = '' |
| if hasattr(ad, 'pretty_name'): |
| prefix = '[%s] ' % ad.pretty_name |
| should_fail = False |
| try: |
| while True: |
| event = ad.ed.pop_events('.*', timeout, freq=0) |
| ad.log.info('%sQueue contains %s', prefix, event) |
| should_fail = True |
| except queue.Empty: |
| if should_fail: |
| asserts.fail('%sEvent queue not empty' % prefix) |
| ad.log.info('%sNo events in the queue (as expected)', prefix) |
| return |
| |
| |
| def encode_list(list_of_objects): |
| """Converts the list of strings or bytearrays to a list of b64 encoded |
| bytearrays. |
| |
| A None object is treated as a zero-length bytearray. |
| |
| Args: |
| list_of_objects: A list of strings or bytearray objects |
| Returns: A list of the same objects, converted to bytes and b64 encoded. |
| """ |
| encoded_list = [] |
| for obj in list_of_objects: |
| if obj is None: |
| obj = bytes() |
| if isinstance(obj, str): |
| encoded_list.append( |
| base64.b64encode(bytes(obj, 'utf-8')).decode('utf-8')) |
| else: |
| encoded_list.append(base64.b64encode(obj).decode('utf-8')) |
| return encoded_list |
| |
| |
| def decode_list(list_of_b64_strings): |
| """Converts the list of b64 encoded strings to a list of bytearray. |
| |
| Args: |
| list_of_b64_strings: list of strings, each of which is b64 encoded array |
| Returns: a list of bytearrays. |
| """ |
| decoded_list = [] |
| for str in list_of_b64_strings: |
| decoded_list.append(base64.b64decode(str)) |
| return decoded_list |
| |
| |
| def construct_max_match_filter(max_size): |
| """Constructs a maximum size match filter that fits into the 'max_size' bytes. |
| |
| Match filters are a set of LVs (Length, Value pairs) where L is 1 byte. The |
| maximum size match filter will contain max_size/2 LVs with all Vs (except |
| possibly the last one) of 1 byte, the last V may be 2 bytes for odd max_size. |
| |
| Args: |
| max_size: Maximum size of the match filter. |
| Returns: an array of bytearrays. |
| """ |
| mf_list = [] |
| num_lvs = max_size // 2 |
| for i in range(num_lvs - 1): |
| mf_list.append(bytes([i])) |
| if (max_size % 2 == 0): |
| mf_list.append(bytes([255])) |
| else: |
| mf_list.append(bytes([254, 255])) |
| return mf_list |
| |
| |
| def assert_equal_strings(first, second, msg=None, extras=None): |
| """Assert equality of the string operands - where None is treated as equal to |
| an empty string (''), otherwise fail the test. |
| |
| Error message is "first != second" by default. Additional explanation can |
| be supplied in the message. |
| |
| Args: |
| first, seconds: The strings that are evaluated for equality. |
| msg: A string that adds additional info about the failure. |
| extras: An optional field for extra information to be included in |
| test result. |
| """ |
| if first == None: |
| first = '' |
| if second == None: |
| second = '' |
| asserts.assert_equal(first, second, msg, extras) |
| |
| |
| def get_aware_capabilities(ad): |
| """Get the Wi-Fi Aware capabilities from the specified device. The |
| capabilities are a dictionary keyed by aware_const.CAP_* keys. |
| |
| Args: |
| ad: the Android device |
| Returns: the capability dictionary. |
| """ |
| return json.loads(ad.adb.shell('cmd wifiaware state_mgr get_capabilities')) |
| |
| |
| def get_wifi_mac_address(ad): |
| """Get the Wi-Fi interface MAC address as a upper-case string of hex digits |
| without any separators (e.g. ':'). |
| |
| Args: |
| ad: Device on which to run. |
| """ |
| return ad.droid.wifiGetConnectionInfo()['mac_address'].upper().replace( |
| ':', '') |
| |
| |
| def validate_forbidden_callbacks(ad, limited_cb=None): |
| """Validate that the specified callbacks have not been called more then permitted. |
| |
| In addition to the input configuration also validates that forbidden callbacks |
| have never been called. |
| |
| Args: |
| ad: Device on which to run. |
| limited_cb: Dictionary of CB_EV_* ids and maximum permitted calls (0 |
| meaning never). |
| """ |
| cb_data = json.loads(ad.adb.shell('cmd wifiaware native_cb get_cb_count')) |
| |
| if limited_cb is None: |
| limited_cb = {} |
| |
| fail = False |
| for cb_event in limited_cb.keys(): |
| if cb_event in cb_data: |
| if cb_data[cb_event] > limited_cb[cb_event]: |
| fail = True |
| ad.log.info( |
| 'Callback %s observed %d times: more then permitted %d times', |
| cb_event, cb_data[cb_event], limited_cb[cb_event]) |
| |
| asserts.assert_false(fail, 'Forbidden callbacks observed', extras=cb_data) |
| |
| |
| def extract_stats(ad, data, results, key_prefix, log_prefix): |
| """Extract statistics from the data, store in the results dictionary, and |
| output to the info log. |
| |
| Args: |
| ad: Android device (for logging) |
| data: A list containing the data to be analyzed. |
| results: A dictionary into which to place the statistics. |
| key_prefix: A string prefix to use for the dict keys storing the |
| extracted stats. |
| log_prefix: A string prefix to use for the info log. |
| include_data: If True includes the raw data in the dictionary, |
| otherwise just the stats. |
| """ |
| num_samples = len(data) |
| results['%snum_samples' % key_prefix] = num_samples |
| |
| if not data: |
| return |
| |
| data_min = min(data) |
| data_max = max(data) |
| data_mean = statistics.mean(data) |
| data_cdf = extract_cdf(data) |
| data_cdf_decile = extract_cdf_decile(data_cdf) |
| |
| results['%smin' % key_prefix] = data_min |
| results['%smax' % key_prefix] = data_max |
| results['%smean' % key_prefix] = data_mean |
| results['%scdf' % key_prefix] = data_cdf |
| results['%scdf_decile' % key_prefix] = data_cdf_decile |
| results['%sraw_data' % key_prefix] = data |
| |
| if num_samples > 1: |
| data_stdev = statistics.stdev(data) |
| results['%sstdev' % key_prefix] = data_stdev |
| ad.log.info( |
| '%s: num_samples=%d, min=%.2f, max=%.2f, mean=%.2f, stdev=%.2f, cdf_decile=%s', |
| log_prefix, num_samples, data_min, data_max, data_mean, data_stdev, |
| data_cdf_decile) |
| else: |
| ad.log.info( |
| '%s: num_samples=%d, min=%.2f, max=%.2f, mean=%.2f, cdf_decile=%s', |
| log_prefix, num_samples, data_min, data_max, data_mean, |
| data_cdf_decile) |
| |
| |
| def extract_cdf_decile(cdf): |
| """Extracts the 10%, 20%, ..., 90% points from the CDF and returns their |
| value (a list of 9 values). |
| |
| Since CDF may not (will not) have exact x% value picks the value >= x%. |
| |
| Args: |
| cdf: a list of 2 lists, the X and Y of the CDF. |
| """ |
| decades = [] |
| next_decade = 10 |
| for x, y in zip(cdf[0], cdf[1]): |
| while 100 * y >= next_decade: |
| decades.append(x) |
| next_decade = next_decade + 10 |
| if next_decade == 100: |
| break |
| return decades |
| |
| |
| def extract_cdf(data): |
| """Calculates the Cumulative Distribution Function (CDF) of the data. |
| |
| Args: |
| data: A list containing data (does not have to be sorted). |
| |
| Returns: a list of 2 lists: the X and Y axis of the CDF. |
| """ |
| x = [] |
| cdf = [] |
| if not data: |
| return (x, cdf) |
| |
| all_values = sorted(data) |
| for val in all_values: |
| if not x: |
| x.append(val) |
| cdf.append(1) |
| else: |
| if x[-1] == val: |
| cdf[-1] += 1 |
| else: |
| x.append(val) |
| cdf.append(cdf[-1] + 1) |
| |
| scale = 1.0 / len(all_values) |
| for i in range(len(cdf)): |
| cdf[i] = cdf[i] * scale |
| |
| return (x, cdf) |
| |
| |
| def get_mac_addr(device, interface): |
| """Get the MAC address of the specified interface. Uses ifconfig and parses |
| its output. Normalizes string to remove ':' and upper case. |
| |
| Args: |
| device: Device on which to query the interface MAC address. |
| interface: Name of the interface for which to obtain the MAC address. |
| """ |
| out = device.adb.shell("ifconfig %s" % interface) |
| res = re.match(".* HWaddr (\S+).*", out, re.S) |
| asserts.assert_true(res, |
| 'Unable to obtain MAC address for interface %s' % |
| interface, |
| extras=out) |
| return res.group(1).upper().replace(':', '') |
| |
| |
| def get_ipv6_addr(device, interface): |
| """Get the IPv6 address of the specified interface. Uses ifconfig and parses |
| its output. Returns a None if the interface does not have an IPv6 address |
| (indicating it is not UP). |
| |
| Args: |
| device: Device on which to query the interface IPv6 address. |
| interface: Name of the interface for which to obtain the IPv6 address. |
| """ |
| out = device.adb.shell("ifconfig %s" % interface) |
| res = re.match(".*inet6 addr: (\S+)/.*", out, re.S) |
| if not res: |
| return None |
| return res.group(1) |
| |
| |
| def verify_socket_connect(dut_s, dut_c, ipv6_s, ipv6_c, port): |
| """Verify the socket connection between server (dut_s) and client (dut_c) |
| using the given IPv6 addresses. |
| |
| Opens a ServerSocket on the server and tries to connect to it |
| from the client. |
| |
| Args: |
| dut_s, dut_c: the server and client devices under test (DUTs) |
| ipv6_s, ipv6_c: the scoped link-local addresses of the server and client. |
| port: the port to use |
| Return: True on success, False otherwise |
| """ |
| server_sock = None |
| sock_c = None |
| sock_s = None |
| try: |
| server_sock = sutils.open_server_socket(dut_s, ipv6_s, port) |
| port_to_use = port |
| if port == 0: |
| port_to_use = dut_s.droid.getTcpServerSocketPort(server_sock) |
| sock_c, sock_s = sutils.open_connect_socket(dut_c, dut_s, ipv6_c, |
| ipv6_s, 0, port_to_use, |
| server_sock) |
| except: |
| return False |
| finally: |
| if sock_c is not None: |
| sutils.close_socket(dut_c, sock_c) |
| if sock_s is not None: |
| sutils.close_socket(dut_s, sock_s) |
| if server_sock is not None: |
| sutils.close_server_socket(dut_s, server_sock) |
| return True |
| |
| |
| def run_ping6(dut, target_ip, duration=60): |
| """Run ping test and return the latency result |
| |
| Args: |
| dut: the dut which run the ping cmd |
| target_ip: target IP Address for ping |
| duration: the duration time of the ping |
| |
| return: dict contains "min/avg/max/mdev" result |
| """ |
| cmd = "ping6 -w %d %s" % (duration, target_ip) |
| ping_result = dut.adb.shell(cmd, timeout=duration + 1) |
| res = re.match(".*mdev = (\S+) .*", ping_result, re.S) |
| asserts.assert_true(res, "Cannot reach the IP address %s", target_ip) |
| title = ["min", "avg", "max", "mdev"] |
| result = res.group(1).split("/") |
| latency_result = {} |
| for i in range(len(title)): |
| latency_result[title[i]] = result[i] |
| return latency_result |
| |
| |
| def reset_device_parameters(ad): |
| """Reset device configurations. |
| |
| Args: |
| ad: device to be reset |
| """ |
| ad.adb.shell("cmd wifiaware reset") |
| |
| |
| def reset_device_statistics(ad): |
| """Reset device statistics. |
| |
| Args: |
| ad: device to be reset |
| """ |
| ad.adb.shell("cmd wifiaware native_cb get_cb_count --reset") |
| |
| |
| def set_power_mode_parameters(ad, power_mode): |
| """Set device power mode. |
| |
| Set the power configuration DW parameters for the device based on any |
| configuration overrides (if provided) |
| |
| Args: |
| ad: android device |
| power_mode: Desired power mode (INTERACTIVE or NON_INTERACTIVE) |
| """ |
| if power_mode == "INTERACTIVE": |
| config_settings_high_power(ad) |
| elif power_mode == "NON_INTERACTIVE": |
| config_settings_low_power(ad) |
| else: |
| asserts.assert_false( |
| "The 'aware_default_power_mode' configuration must be INTERACTIVE or " |
| "NON_INTERACTIVE") |
| |
| |
| ######################################################### |
| # Aware primitives |
| ######################################################### |
| |
| |
| def request_network(dut, ns): |
| """Request a Wi-Fi Aware network. |
| |
| Args: |
| dut: Device |
| ns: Network specifier |
| Returns: the request key |
| """ |
| network_req = {"TransportType": 5, "NetworkSpecifier": ns} |
| return dut.droid.connectivityRequestWifiAwareNetwork(network_req) |
| |
| |
| def get_network_specifier(dut, id, dev_type, peer_mac, sec): |
| """Create a network specifier for the device based on the security |
| configuration. |
| |
| Args: |
| dut: device |
| id: session ID |
| dev_type: device type - Initiator or Responder |
| peer_mac: the discovery MAC address of the peer |
| sec: security configuration |
| """ |
| if sec is None: |
| return dut.droid.wifiAwareCreateNetworkSpecifierOob( |
| id, dev_type, peer_mac) |
| if isinstance(sec, str): |
| return dut.droid.wifiAwareCreateNetworkSpecifierOob( |
| id, dev_type, peer_mac, sec) |
| return dut.droid.wifiAwareCreateNetworkSpecifierOob( |
| id, dev_type, peer_mac, None, sec) |
| |
| |
| def configure_power_setting(device, mode, name, value): |
| """Use the command-line API to configure the power setting |
| |
| Args: |
| device: Device on which to perform configuration |
| mode: The power mode being set, should be "default", "inactive", or "idle" |
| name: One of the power settings from 'wifiaware set-power'. |
| value: An integer. |
| """ |
| device.adb.shell("cmd wifiaware native_api set-power %s %s %d" % |
| (mode, name, value)) |
| |
| |
| def configure_mac_random_interval(device, interval_sec): |
| """Use the command-line API to configure the MAC address randomization |
| interval. |
| |
| Args: |
| device: Device on which to perform configuration |
| interval_sec: The MAC randomization interval in seconds. A value of 0 |
| disables all randomization. |
| """ |
| device.adb.shell( |
| "cmd wifiaware native_api set mac_random_interval_sec %d" % |
| interval_sec) |
| |
| |
| def configure_ndp_allow_any_override(device, override_api_check): |
| """Use the command-line API to configure whether an NDP Responder may be |
| configured to accept an NDP request from ANY peer. |
| |
| By default the target API level of the requesting app determines whether such |
| configuration is permitted. This allows overriding the API check and allowing |
| it. |
| |
| Args: |
| device: Device on which to perform configuration. |
| override_api_check: True to allow a Responder to ANY configuration, False to |
| perform the API level check. |
| """ |
| device.adb.shell("cmd wifiaware state_mgr allow_ndp_any %s" % |
| ("true" if override_api_check else "false")) |
| |
| |
| def config_settings_high_power(device): |
| """Configure device's power settings values to high power mode - |
| whether device is in interactive or non-interactive modes""" |
| configure_power_setting(device, "default", "dw_24ghz", |
| aconsts.POWER_DW_24_INTERACTIVE) |
| configure_power_setting(device, "default", "dw_5ghz", |
| aconsts.POWER_DW_5_INTERACTIVE) |
| configure_power_setting(device, "default", "disc_beacon_interval_ms", |
| aconsts.POWER_DISC_BEACON_INTERVAL_INTERACTIVE) |
| configure_power_setting(device, "default", "num_ss_in_discovery", |
| aconsts.POWER_NUM_SS_IN_DISC_INTERACTIVE) |
| configure_power_setting(device, "default", "enable_dw_early_term", |
| aconsts.POWER_ENABLE_DW_EARLY_TERM_INTERACTIVE) |
| |
| configure_power_setting(device, "inactive", "dw_24ghz", |
| aconsts.POWER_DW_24_INTERACTIVE) |
| configure_power_setting(device, "inactive", "dw_5ghz", |
| aconsts.POWER_DW_5_INTERACTIVE) |
| configure_power_setting(device, "inactive", "disc_beacon_interval_ms", |
| aconsts.POWER_DISC_BEACON_INTERVAL_INTERACTIVE) |
| configure_power_setting(device, "inactive", "num_ss_in_discovery", |
| aconsts.POWER_NUM_SS_IN_DISC_INTERACTIVE) |
| configure_power_setting(device, "inactive", "enable_dw_early_term", |
| aconsts.POWER_ENABLE_DW_EARLY_TERM_INTERACTIVE) |
| |
| |
| def config_settings_low_power(device): |
| """Configure device's power settings values to low power mode - whether |
| device is in interactive or non-interactive modes""" |
| configure_power_setting(device, "default", "dw_24ghz", |
| aconsts.POWER_DW_24_NON_INTERACTIVE) |
| configure_power_setting(device, "default", "dw_5ghz", |
| aconsts.POWER_DW_5_NON_INTERACTIVE) |
| configure_power_setting(device, "default", "disc_beacon_interval_ms", |
| aconsts.POWER_DISC_BEACON_INTERVAL_NON_INTERACTIVE) |
| configure_power_setting(device, "default", "num_ss_in_discovery", |
| aconsts.POWER_NUM_SS_IN_DISC_NON_INTERACTIVE) |
| configure_power_setting(device, "default", "enable_dw_early_term", |
| aconsts.POWER_ENABLE_DW_EARLY_TERM_NON_INTERACTIVE) |
| |
| configure_power_setting(device, "inactive", "dw_24ghz", |
| aconsts.POWER_DW_24_NON_INTERACTIVE) |
| configure_power_setting(device, "inactive", "dw_5ghz", |
| aconsts.POWER_DW_5_NON_INTERACTIVE) |
| configure_power_setting(device, "inactive", "disc_beacon_interval_ms", |
| aconsts.POWER_DISC_BEACON_INTERVAL_NON_INTERACTIVE) |
| configure_power_setting(device, "inactive", "num_ss_in_discovery", |
| aconsts.POWER_NUM_SS_IN_DISC_NON_INTERACTIVE) |
| configure_power_setting(device, "inactive", "enable_dw_early_term", |
| aconsts.POWER_ENABLE_DW_EARLY_TERM_NON_INTERACTIVE) |
| |
| |
| def config_power_settings(device, |
| dw_24ghz, |
| dw_5ghz, |
| disc_beacon_interval=None, |
| num_ss_in_disc=None, |
| enable_dw_early_term=None): |
| """Configure device's discovery window (DW) values to the specified values - |
| whether the device is in interactive or non-interactive mode. |
| |
| Args: |
| dw_24ghz: DW interval in the 2.4GHz band. |
| dw_5ghz: DW interval in the 5GHz band. |
| disc_beacon_interval: The discovery beacon interval (in ms). If None then |
| not set. |
| num_ss_in_disc: Number of spatial streams to use for discovery. If None then |
| not set. |
| enable_dw_early_term: If True then enable early termination of the DW. If |
| None then not set. |
| """ |
| configure_power_setting(device, "default", "dw_24ghz", dw_24ghz) |
| configure_power_setting(device, "default", "dw_5ghz", dw_5ghz) |
| configure_power_setting(device, "inactive", "dw_24ghz", dw_24ghz) |
| configure_power_setting(device, "inactive", "dw_5ghz", dw_5ghz) |
| |
| if disc_beacon_interval is not None: |
| configure_power_setting(device, "default", "disc_beacon_interval_ms", |
| disc_beacon_interval) |
| configure_power_setting(device, "inactive", "disc_beacon_interval_ms", |
| disc_beacon_interval) |
| |
| if num_ss_in_disc is not None: |
| configure_power_setting(device, "default", "num_ss_in_discovery", |
| num_ss_in_disc) |
| configure_power_setting(device, "inactive", "num_ss_in_discovery", |
| num_ss_in_disc) |
| |
| if enable_dw_early_term is not None: |
| configure_power_setting(device, "default", "enable_dw_early_term", |
| enable_dw_early_term) |
| configure_power_setting(device, "inactive", "enable_dw_early_term", |
| enable_dw_early_term) |
| |
| |
| def create_discovery_config(service_name, |
| d_type, |
| ssi=None, |
| match_filter=None, |
| match_filter_list=None, |
| ttl=0, |
| term_cb_enable=True, |
| instant_mode=None): |
| """Create a publish discovery configuration based on input parameters. |
| |
| Args: |
| service_name: Service name - required |
| d_type: Discovery type (publish or subscribe constants) |
| ssi: Supplemental information - defaults to None |
| match_filter, match_filter_list: The match_filter, only one mechanism can |
| be used to specify. Defaults to None. |
| ttl: Time-to-live - defaults to 0 (i.e. non-self terminating) |
| term_cb_enable: True (default) to enable callback on termination, False |
| means that no callback is called when session terminates. |
| instant_mode: set the band to use instant communication mode, 2G or 5G |
| Returns: |
| publish discovery configuration object. |
| """ |
| config = {} |
| config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = service_name |
| config[aconsts.DISCOVERY_KEY_DISCOVERY_TYPE] = d_type |
| if ssi is not None: |
| config[aconsts.DISCOVERY_KEY_SSI] = ssi |
| if match_filter is not None: |
| config[aconsts.DISCOVERY_KEY_MATCH_FILTER] = match_filter |
| if match_filter_list is not None: |
| config[aconsts.DISCOVERY_KEY_MATCH_FILTER_LIST] = match_filter_list |
| if instant_mode is not None: |
| config[aconsts.DISCOVERY_KEY_INSTANT_COMMUNICATION_MODE] = instant_mode |
| config[aconsts.DISCOVERY_KEY_TTL] = ttl |
| config[aconsts.DISCOVERY_KEY_TERM_CB_ENABLED] = term_cb_enable |
| return config |
| |
| |
| def add_ranging_to_pub(p_config, enable_ranging): |
| """Add ranging enabled configuration to a publish configuration (only relevant |
| for publish configuration). |
| |
| Args: |
| p_config: The Publish discovery configuration. |
| enable_ranging: True to enable ranging, False to disable. |
| Returns: |
| The modified publish configuration. |
| """ |
| p_config[aconsts.DISCOVERY_KEY_RANGING_ENABLED] = enable_ranging |
| return p_config |
| |
| |
| def add_ranging_to_sub(s_config, min_distance_mm, max_distance_mm): |
| """Add ranging distance configuration to a subscribe configuration (only |
| relevant to a subscribe configuration). |
| |
| Args: |
| s_config: The Subscribe discovery configuration. |
| min_distance_mm, max_distance_mm: The min and max distance specification. |
| Used if not None. |
| Returns: |
| The modified subscribe configuration. |
| """ |
| if min_distance_mm is not None: |
| s_config[aconsts.DISCOVERY_KEY_MIN_DISTANCE_MM] = min_distance_mm |
| if max_distance_mm is not None: |
| s_config[aconsts.DISCOVERY_KEY_MAX_DISTANCE_MM] = max_distance_mm |
| return s_config |
| |
| |
| def attach_with_identity(dut): |
| """Start an Aware session (attach) and wait for confirmation and identity |
| information (mac address). |
| |
| Args: |
| dut: Device under test |
| Returns: |
| id: Aware session ID. |
| mac: Discovery MAC address of this device. |
| """ |
| id = dut.droid.wifiAwareAttach(True) |
| wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED) |
| event = wait_for_event(dut, aconsts.EVENT_CB_ON_IDENTITY_CHANGED) |
| mac = event["data"]["mac"] |
| |
| return id, mac |
| |
| |
| def create_discovery_pair(p_dut, |
| s_dut, |
| p_config, |
| s_config, |
| device_startup_offset, |
| msg_id=None): |
| """Creates a discovery session (publish and subscribe), and waits for |
| service discovery - at that point the sessions are connected and ready for |
| further messaging of data-path setup. |
| |
| Args: |
| p_dut: Device to use as publisher. |
| s_dut: Device to use as subscriber. |
| p_config: Publish configuration. |
| s_config: Subscribe configuration. |
| device_startup_offset: Number of seconds to offset the enabling of NAN on |
| the two devices. |
| msg_id: Controls whether a message is sent from Subscriber to Publisher |
| (so that publisher has the sub's peer ID). If None then not sent, |
| otherwise should be an int for the message id. |
| Returns: variable size list of: |
| p_id: Publisher attach session id |
| s_id: Subscriber attach session id |
| p_disc_id: Publisher discovery session id |
| s_disc_id: Subscriber discovery session id |
| peer_id_on_sub: Peer ID of the Publisher as seen on the Subscriber |
| peer_id_on_pub: Peer ID of the Subscriber as seen on the Publisher. Only |
| included if |msg_id| is not None. |
| """ |
| p_dut.pretty_name = 'Publisher' |
| s_dut.pretty_name = 'Subscriber' |
| |
| # Publisher+Subscriber: attach and wait for confirmation |
| p_id = p_dut.droid.wifiAwareAttach() |
| wait_for_event(p_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| time.sleep(device_startup_offset) |
| s_id = s_dut.droid.wifiAwareAttach() |
| wait_for_event(s_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| |
| # Publisher: start publish and wait for confirmation |
| p_disc_id = p_dut.droid.wifiAwarePublish(p_id, p_config) |
| wait_for_event(p_dut, aconsts.SESSION_CB_ON_PUBLISH_STARTED) |
| |
| # Subscriber: start subscribe and wait for confirmation |
| s_disc_id = s_dut.droid.wifiAwareSubscribe(s_id, s_config) |
| wait_for_event(s_dut, aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED) |
| |
| # Subscriber: wait for service discovery |
| discovery_event = wait_for_event(s_dut, |
| aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) |
| peer_id_on_sub = discovery_event['data'][aconsts.SESSION_CB_KEY_PEER_ID] |
| |
| # Optionally send a message from Subscriber to Publisher |
| if msg_id is not None: |
| ping_msg = 'PING' |
| |
| # Subscriber: send message to peer (Publisher) |
| s_dut.droid.wifiAwareSendMessage(s_disc_id, peer_id_on_sub, msg_id, |
| ping_msg, aconsts.MAX_TX_RETRIES) |
| sub_tx_msg_event = wait_for_event(s_dut, |
| aconsts.SESSION_CB_ON_MESSAGE_SENT) |
| asserts.assert_equal( |
| msg_id, |
| sub_tx_msg_event['data'][aconsts.SESSION_CB_KEY_MESSAGE_ID], |
| 'Subscriber -> Publisher message ID corrupted') |
| |
| # Publisher: wait for received message |
| pub_rx_msg_event = wait_for_event( |
| p_dut, aconsts.SESSION_CB_ON_MESSAGE_RECEIVED) |
| peer_id_on_pub = pub_rx_msg_event['data'][ |
| aconsts.SESSION_CB_KEY_PEER_ID] |
| asserts.assert_equal( |
| ping_msg, |
| pub_rx_msg_event['data'][aconsts.SESSION_CB_KEY_MESSAGE_AS_STRING], |
| 'Subscriber -> Publisher message corrupted') |
| return p_id, s_id, p_disc_id, s_disc_id, peer_id_on_sub, peer_id_on_pub |
| |
| return p_id, s_id, p_disc_id, s_disc_id, peer_id_on_sub |
| |
| |
| def create_ib_ndp(p_dut, s_dut, p_config, s_config, device_startup_offset): |
| """Create an NDP (using in-band discovery) |
| |
| Args: |
| p_dut: Device to use as publisher. |
| s_dut: Device to use as subscriber. |
| p_config: Publish configuration. |
| s_config: Subscribe configuration. |
| device_startup_offset: Number of seconds to offset the enabling of NAN on |
| the two devices. |
| """ |
| (p_id, s_id, p_disc_id, s_disc_id, peer_id_on_sub, |
| peer_id_on_pub) = create_discovery_pair(p_dut, |
| s_dut, |
| p_config, |
| s_config, |
| device_startup_offset, |
| msg_id=9999) |
| |
| # Publisher: request network |
| p_req_key = request_network( |
| p_dut, |
| p_dut.droid.wifiAwareCreateNetworkSpecifier(p_disc_id, peer_id_on_pub, |
| None)) |
| |
| # Subscriber: request network |
| s_req_key = request_network( |
| s_dut, |
| s_dut.droid.wifiAwareCreateNetworkSpecifier(s_disc_id, peer_id_on_sub, |
| None)) |
| |
| # Publisher & Subscriber: wait for network formation |
| p_net_event_nc = wait_for_event_with_keys( |
| p_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_CAPABILITIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, p_req_key)) |
| s_net_event_nc = wait_for_event_with_keys( |
| s_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_CAPABILITIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, s_req_key)) |
| |
| # validate no leak of information |
| asserts.assert_false( |
| cconsts.NETWORK_CB_KEY_NETWORK_SPECIFIER in p_net_event_nc["data"], |
| "Network specifier leak!") |
| asserts.assert_false( |
| cconsts.NETWORK_CB_KEY_NETWORK_SPECIFIER in s_net_event_nc["data"], |
| "Network specifier leak!") |
| |
| # note that Pub <-> Sub since IPv6 are of peer's! |
| p_ipv6 = s_net_event_nc["data"][aconsts.NET_CAP_IPV6] |
| s_ipv6 = p_net_event_nc["data"][aconsts.NET_CAP_IPV6] |
| |
| p_net_event_lp = wait_for_event_with_keys( |
| p_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, p_req_key)) |
| s_net_event_lp = wait_for_event_with_keys( |
| s_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, s_req_key)) |
| |
| p_aware_if = p_net_event_lp["data"][cconsts.NETWORK_CB_KEY_INTERFACE_NAME] |
| s_aware_if = s_net_event_lp["data"][cconsts.NETWORK_CB_KEY_INTERFACE_NAME] |
| |
| return p_req_key, s_req_key, p_aware_if, s_aware_if, p_ipv6, s_ipv6 |
| |
| |
| def create_oob_ndp_on_sessions(init_dut, resp_dut, init_id, init_mac, resp_id, |
| resp_mac): |
| """Create an NDP on top of existing Aware sessions (using OOB discovery) |
| |
| Args: |
| init_dut: Initiator device |
| resp_dut: Responder device |
| init_id: Initiator attach session id |
| init_mac: Initiator discovery MAC address |
| resp_id: Responder attach session id |
| resp_mac: Responder discovery MAC address |
| Returns: |
| init_req_key: Initiator network request |
| resp_req_key: Responder network request |
| init_aware_if: Initiator Aware data interface |
| resp_aware_if: Responder Aware data interface |
| init_ipv6: Initiator IPv6 address |
| resp_ipv6: Responder IPv6 address |
| """ |
| # Responder: request network |
| resp_req_key = request_network( |
| resp_dut, |
| resp_dut.droid.wifiAwareCreateNetworkSpecifierOob( |
| resp_id, aconsts.DATA_PATH_RESPONDER, init_mac, None)) |
| |
| # Initiator: request network |
| init_req_key = request_network( |
| init_dut, |
| init_dut.droid.wifiAwareCreateNetworkSpecifierOob( |
| init_id, aconsts.DATA_PATH_INITIATOR, resp_mac, None)) |
| |
| # Initiator & Responder: wait for network formation |
| init_net_event_nc = wait_for_event_with_keys( |
| init_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_CAPABILITIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, init_req_key)) |
| resp_net_event_nc = wait_for_event_with_keys( |
| resp_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_CAPABILITIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, resp_req_key)) |
| |
| # validate no leak of information |
| asserts.assert_false( |
| cconsts.NETWORK_CB_KEY_NETWORK_SPECIFIER in init_net_event_nc["data"], |
| "Network specifier leak!") |
| asserts.assert_false( |
| cconsts.NETWORK_CB_KEY_NETWORK_SPECIFIER in resp_net_event_nc["data"], |
| "Network specifier leak!") |
| |
| # note that Init <-> Resp since IPv6 are of peer's! |
| resp_ipv6 = init_net_event_nc["data"][aconsts.NET_CAP_IPV6] |
| init_ipv6 = resp_net_event_nc["data"][aconsts.NET_CAP_IPV6] |
| |
| init_net_event_lp = wait_for_event_with_keys( |
| init_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, init_req_key)) |
| resp_net_event_lp = wait_for_event_with_keys( |
| resp_dut, cconsts.EVENT_NETWORK_CALLBACK, EVENT_NDP_TIMEOUT, |
| (cconsts.NETWORK_CB_KEY_EVENT, |
| cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), |
| (cconsts.NETWORK_CB_KEY_ID, resp_req_key)) |
| |
| init_aware_if = init_net_event_lp['data'][ |
| cconsts.NETWORK_CB_KEY_INTERFACE_NAME] |
| resp_aware_if = resp_net_event_lp['data'][ |
| cconsts.NETWORK_CB_KEY_INTERFACE_NAME] |
| |
| return (init_req_key, resp_req_key, init_aware_if, resp_aware_if, |
| init_ipv6, resp_ipv6) |
| |
| |
| def create_oob_ndp(init_dut, resp_dut): |
| """Create an NDP (using OOB discovery) |
| |
| Args: |
| init_dut: Initiator device |
| resp_dut: Responder device |
| """ |
| init_dut.pretty_name = 'Initiator' |
| resp_dut.pretty_name = 'Responder' |
| |
| # Initiator+Responder: attach and wait for confirmation & identity |
| init_id = init_dut.droid.wifiAwareAttach(True) |
| wait_for_event(init_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| init_ident_event = wait_for_event(init_dut, |
| aconsts.EVENT_CB_ON_IDENTITY_CHANGED) |
| init_mac = init_ident_event['data']['mac'] |
| resp_id = resp_dut.droid.wifiAwareAttach(True) |
| wait_for_event(resp_dut, aconsts.EVENT_CB_ON_ATTACHED) |
| resp_ident_event = wait_for_event(resp_dut, |
| aconsts.EVENT_CB_ON_IDENTITY_CHANGED) |
| resp_mac = resp_ident_event['data']['mac'] |
| |
| # wait for for devices to synchronize with each other - there are no other |
| # mechanisms to make sure this happens for OOB discovery (except retrying |
| # to execute the data-path request) |
| time.sleep(WAIT_FOR_CLUSTER) |
| |
| (init_req_key, resp_req_key, init_aware_if, resp_aware_if, init_ipv6, |
| resp_ipv6) = create_oob_ndp_on_sessions(init_dut, resp_dut, init_id, |
| init_mac, resp_id, resp_mac) |
| |
| return (init_req_key, resp_req_key, init_aware_if, resp_aware_if, |
| init_ipv6, resp_ipv6) |
