#!/usr/bin/env python3.4 # # Copyright (C) 2016 The Android Open Source Project # # 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 pprint import queue import acts.base_test import acts.test_utils.wifi.wifi_test_utils as wutils import acts.utils from acts import asserts from acts.controllers.sl4a_lib import rpc_client WifiEnums = wutils.WifiEnums # Macros for RttParam keywords RttParam = WifiEnums.RttParam # Macros for RttManager Rtt = WifiEnums.Rtt RttBW = WifiEnums.RttBW RttPreamble = WifiEnums.RttPreamble RttPeerType = WifiEnums.RttPeerType RttType = WifiEnums.RttType ScanResult = WifiEnums.ScanResult RTT_MARGIN_OF_ERROR = WifiEnums.RTT_MARGIN_OF_ERROR class WifiRTTRangingError(Exception): """Error in WifiScanner Rtt.""" class WifiRttManagerTest(acts.base_test.BaseTestClass): """Tests for wifi's RttManager APIs.""" tests = None MAX_RTT_AP = 10 def __init__(self, controllers): acts.base_test.BaseTestClass.__init__(self, controllers) self.tests = ("test_support_check", "test_invalid_params", "test_capability_check", "test_rtt_ranging_single_AP_stress", "test_regular_scan_then_rtt_ranging_stress", "test_gscan_then_rtt_ranging_stress") def setup_class(self): self.dut = self.android_devices[0] wutils.wifi_test_device_init(self.dut) required_params = ("support_models", "stress_num", "vht80_5g", "actual_distance") self.unpack_userparams(required_params) asserts.assert_true( self.actual_distance >= 5, "Actual distance should be no shorter than 5 meters.") self.visible_networks = (self.vht80_5g, ) self.default_rtt_params = { RttParam.request_type: RttType.TYPE_TWO_SIDED, RttParam.device_type: RttPeerType.PEER_TYPE_AP, RttParam.preamble: RttPreamble.PREAMBLE_HT, RttParam.bandwidth: RttBW.BW_80_SUPPORT } # Expected capability for devices that don't support RTT. rtt_cap_neg = { 'lcrSupported': False, 'bwSupported': 0, 'twoSided11McRttSupported': False, 'preambleSupported': 0, 'oneSidedRttSupported': False, 'lciSupported': False } rtt_cap_shamu = { 'lcrSupported': False, 'bwSupported': 0x1C, 'twoSided11McRttSupported': True, 'preambleSupported': 6, 'oneSidedRttSupported': False, 'lciSupported': False } rtt_cap_bullhead = { 'lcrSupported': True, 'bwSupported': 0x1C, 'twoSided11McRttSupported': True, 'preambleSupported': 7, 'oneSidedRttSupported': True, 'lciSupported': True } rtt_cap_angler = { 'lcrSupported': True, 'bwSupported': 0x1C, 'twoSided11McRttSupported': True, 'preambleSupported': 6, 'oneSidedRttSupported': False, 'lciSupported': True } self.rtt_cap_table = { "hammerhead": rtt_cap_neg, "shamu": rtt_cap_shamu, "volantis": rtt_cap_neg, "volantisg": rtt_cap_neg, "bullhead": rtt_cap_bullhead, "angler": rtt_cap_angler } """Helper Functions""" def invalid_params_logic(self, rtt_params): try: self.dut.droid.wifiRttStartRanging([rtt_params]) except rpc_client.Sl4aApiError as e: e_str = str(e) asserts.assert_true( "IllegalArgumentException" in e_str, "Missing IllegalArgumentException in %s." % e_str) msg = "Got expected exception with invalid param %s." % rtt_params self.log.info(msg) def get_rtt_results(self, rtt_params): """Starts RTT ranging and get results. Args: rtt_params: A list of dicts each representing an RttParam. Returns: Rtt ranging results. """ self.log.debug("Start ranging with:\n%s" % pprint.pformat(rtt_params)) idx = self.dut.droid.wifiRttStartRanging(rtt_params) event = None try: event = self.dut.ed.pop_events("WifiRttRanging%d" % idx, 30) if event[0]["name"].endswith("onSuccess"): results = event[0]["data"]["Results"] result_len = len(results) param_len = len(rtt_params) asserts.assert_true(result_len == param_len, "Expected %d results, got %d." % (param_len, result_len)) # Add acceptable margin of error to results, which will be used # during result processing. for i, r in enumerate(results): bw_mode = rtt_params[i][RttParam.bandwidth] r[RttParam.margin] = RTT_MARGIN_OF_ERROR[bw_mode] self.log.debug(pprint.pformat(event)) return event except queue.Empty: self.log.error("Waiting for RTT event timed out.") return None def network_selector(self, network_info): """Decides if a network should be used for rtt ranging. There are a few conditions: 1. This network supports 80211mc. 2. This network's info matches certain conditions. This is added to better control which networks to range against instead of blindly use all 80211mc networks in air. Args: network_info: A dict representing a WiFi network. Returns: True if the input network should be used for ranging, False otherwise. """ target_params = { "is80211McRTTResponder": True, WifiEnums.BSSID_KEY: self.vht80_5g[WifiEnums.BSSID_KEY], } for k, v in target_params.items(): if k not in network_info: return False if type(network_info[k]) is str: network_info[k] = network_info[k].lower() v = v.lower() if network_info[k] != v: return False return True def regular_scan_for_rtt_networks(self): """Scans for 11mc-capable WiFi networks using regular wifi scan. Networks are selected based on self.network_selector. Returns: A list of networks that have RTTResponders. """ wutils.start_wifi_connection_scan(self.dut) networks = self.dut.droid.wifiGetScanResults() rtt_networks = [] for nw in networks: if self.network_selector(nw): rtt_networks.append(nw) return rtt_networks def gscan_for_rtt_networks(self): """Scans for 11mc-capable WiFi networks using wifi gscan. Networks are selected based on self.network_selector. Returns: A list of networks that have RTTResponders. """ s = { "reportEvents": WifiEnums.REPORT_EVENT_FULL_SCAN_RESULT, "band": WifiEnums.WIFI_BAND_BOTH, "periodInMs": 10000, "numBssidsPerScan": 32 } idx = wutils.start_wifi_single_scan(self.android_devices[0], s)["Index"] self.log.info("Scan index is %d" % idx) event_name = "WifiScannerScan%donFullResult" % idx def condition(event): nw = event["data"]["Results"][0] return self.network_selector(nw) rtt_networks = [] try: for i in range(len(self.visible_networks)): event = self.dut.ed.wait_for_event(event_name, condition, 30) rtt_networks.append(event["data"]["Results"][0]) self.log.info("Waiting for gscan to finish.") event_name = "WifiScannerScan%donResults" % idx event = self.dut.ed.pop_event(event_name, 300) total_network_cnt = len(event["data"]["Results"][0]["ScanResults"]) self.log.info("Found %d networks in total." % total_network_cnt) self.log.debug(rtt_networks) return rtt_networks except queue.Empty: self.log.error("Timed out waiting for gscan result.") def process_rtt_events(self, events): """Processes rtt ranging events. Validates RTT event types. Validates RTT response status and measured RTT values. Enforces success rate. Args: events: A list of callback results from RTT ranging. """ total = aborted = failure = invalid = out_of_range = 0 for e in events: if e["name"].endswith("onAborted"): aborted += 1 if e["name"].endswith("onFailure"): failure += 1 if e["name"].endswith("onSuccess"): results = e["data"]["Results"] for r in results: total += 1 # Status needs to be "success". status = r["status"] if status != Rtt.STATUS_SUCCESS: self.log.warning("Got error status %d." % status) invalid += 1 continue # RTT value should be positive. value = r["rtt"] if value <= 0: self.log.warning("Got error RTT value %d." % value) invalid += 1 continue # Vadlidate values in successful responses. acd = self.actual_distance margin = r[RttParam.margin] # If the distance is >= 0, check distance only. d = r["distance"] / 100.0 if d > 0: # Distance should be in acceptable range. is_d_valid = (acd - margin) <= d <= acd + (margin) if not is_d_valid: self.log.warning( ("Reported distance %.2fm is out of the" " acceptable range %.2f±%.2fm.") % (d, acd, margin)) out_of_range += 1 continue # Check if the RTT value is in range. d = (value / 2) / 1E10 * wutils.SPEED_OF_LIGHT is_rtt_valid = (acd - margin) <= d <= (acd + margin) if not is_rtt_valid: self.log.warning(( "Distance calculated from RTT value %d - %.2fm is " "out of the acceptable range %.2f±%dm.") % (value, d, acd, margin)) out_of_range += 1 continue # Check if the RSSI value is in range. rssi = r["rssi"] # average rssi in 0.5dB steps, e.g. 143 implies -71.5dB, # so the valid range is 0 to 200 is_rssi_valid = 0 <= rssi <= 200 if not is_rssi_valid: self.log.warning(("Reported RSSI %d is out of the" " acceptable range 0-200") % rssi) out_of_range += 1 continue self.log.info(( "Processed %d RTT events. %d aborted, %s failed. Among" " the %d responses in successful callbacks, %s are invalid, %s has" " RTT values that are out of range.") % (len(events), aborted, failure, total, invalid, out_of_range)) asserts.assert_true(total > 0, "No RTT response received.") # Percentage of responses that are valid should be >= 90%. valid_total = float(total - invalid) valid_response_rate = valid_total / total self.log.info("%.2f%% of the responses are valid." % (valid_response_rate * 100)) asserts.assert_true(valid_response_rate >= 0.9, "Valid response rate is below 90%%.") # Among the valid responses, the percentage of having an in-range RTT # value should be >= 67%. valid_value_rate = (total - invalid - out_of_range) / valid_total self.log.info("%.2f%% of valid responses have in-range RTT value" % (valid_value_rate * 100)) msg = "In-range response rate is below 67%%." asserts.assert_true(valid_value_rate >= 0.67, msg) def scan_then_rtt_ranging_stress_logic(self, scan_func): """Test logic to scan then do rtt ranging based on the scan results. Steps: 1. Start scan and get scan results. 2. Filter out the networks that support rtt in scan results. 3. Start rtt ranging against those networks that support rtt. 4. Repeat 5. Process RTT events. Args: scan_func: A function that does a wifi scan and only returns the networks that support rtt in the scan results. Returns: True if rtt behaves as expected, False otherwise. """ total = self.stress_num failed = 0 all_results = [] for i in range(total): self.log.info("Iteration %d" % i) rtt_networks = scan_func() if not rtt_networks: self.log.warning("Found no rtt network, skip this iteration.") failed += 1 continue self.log.debug("Found rtt networks:%s" % rtt_networks) rtt_params = [] for rn in rtt_networks: rtt_params.append(self.rtt_config_from_scan_result(rn)) results = self.get_rtt_results(rtt_params) if results: self.log.debug(results) all_results += results self.process_rtt_events(all_results) def rtt_config_from_scan_result(self, scan_result): """Creates an Rtt configuration based on the scan result of a network. """ scan_result_channel_width_to_rtt = { ScanResult.CHANNEL_WIDTH_20MHZ: RttBW.BW_20_SUPPORT, ScanResult.CHANNEL_WIDTH_40MHZ: RttBW.BW_40_SUPPORT, ScanResult.CHANNEL_WIDTH_80MHZ: RttBW.BW_80_SUPPORT, ScanResult.CHANNEL_WIDTH_160MHZ: RttBW.BW_160_SUPPORT, ScanResult.CHANNEL_WIDTH_80MHZ_PLUS_MHZ: RttBW.BW_160_SUPPORT } p = {} freq = scan_result[RttParam.frequency] p[RttParam.frequency] = freq p[RttParam.BSSID] = scan_result[WifiEnums.BSSID_KEY] if freq > 5000: p[RttParam.preamble] = RttPreamble.PREAMBLE_VHT else: p[RttParam.preamble] = RttPreamble.PREAMBLE_HT cf0 = scan_result[RttParam.center_freq0] if cf0 > 0: p[RttParam.center_freq0] = cf0 cf1 = scan_result[RttParam.center_freq1] if cf1 > 0: p[RttParam.center_freq1] = cf1 cw = scan_result["channelWidth"] p[RttParam.channel_width] = cw p[RttParam.bandwidth] = scan_result_channel_width_to_rtt[cw] if scan_result["is80211McRTTResponder"]: p[RttParam.request_type] = RttType.TYPE_TWO_SIDED else: p[RttParam.request_type] = RttType.TYPE_ONE_SIDED return p """Tests""" def test_invalid_params(self): """Tests the check function in RttManager. """ param_list = [{ RttParam.device_type: 3 }, { RttParam.device_type: 1, RttParam.request_type: 3 }, { RttParam.device_type: 1, RttParam.request_type: 1, RttParam.BSSID: None }, { RttParam.BSSID: "xxxxxxxx", RttParam.number_burst: 1 }, { RttParam.number_burst: 0, RttParam.num_samples_per_burst: -1 }, { RttParam.num_samples_per_burst: 32 }, { RttParam.num_samples_per_burst: 5, RttParam.num_retries_per_measurement_frame: -1 }, { RttParam.num_retries_per_measurement_frame: 4 }, { RttParam.num_retries_per_measurement_frame: 2, RttParam.num_retries_per_FTMR: -1 }, { RttParam.num_retries_per_FTMR: 4 }] for param in param_list: self.invalid_params_logic(param) return True def test_support_check(self): """No device supports device-to-device RTT; only shamu and volantis devices support device-to-ap RTT. """ model = acts.utils.trim_model_name(self.dut.model) asserts.assert_true(self.dut.droid.wifiIsDeviceToDeviceRttSupported(), "Device to device is supposed to be supported.") if any([model in m for m in self.support_models]): asserts.assert_true(self.dut.droid.wifiIsDeviceToApRttSupported(), "%s should support device-to-ap RTT." % model) self.log.info("%s supports device-to-ap RTT as expected." % model) else: asserts.assert_false( self.dut.droid.wifiIsDeviceToApRttSupported(), "%s should not support device-to-ap RTT." % model) self.log.info( ("%s does not support device-to-ap RTT as expected.") % model) asserts.abort_class( "Device %s does not support RTT, abort." % model) return True def test_capability_check(self): """Checks the capabilities params are reported as expected. """ caps = self.dut.droid.wifiRttGetCapabilities() asserts.assert_true(caps, "Unable to get rtt capabilities.") self.log.debug("Got rtt capabilities %s" % caps) model = acts.utils.trim_model_name(self.dut.model) asserts.assert_true(model in self.rtt_cap_table, "Unknown model %s" % model) expected_caps = self.rtt_cap_table[model] for k, v in expected_caps.items(): asserts.assert_true(k in caps, "%s missing in capabilities." % k) asserts.assert_true(v == caps[k], "Expected %s for %s, got %s." % (v, k, caps[k])) return True def test_discovery(self): """Make sure all the expected 11mc BSSIDs are discovered properly, and they are all reported as 802.11mc Rtt Responder. Procedures: 1. Scan for wifi networks. Expect: All the RTT networks show up in scan results and their "is80211McRTTResponder" is True. All the non-RTT networks show up in scan results and their "is80211McRTTResponder" is False. """ wutils.start_wifi_connection_scan(self.dut) scan_results = self.dut.droid.wifiGetScanResults() self.log.debug(scan_results) for n in visible_networks: asserts.assert_true( wutils.match_networks(n, scan_results), "Network %s was not discovered properly." % n) return True def test_missing_bssid(self): """Start Rtt ranging with a config that does not have BSSID set. Should not get onSuccess. """ p = {} p[RttParam.request_type] = RttType.TYPE_TWO_SIDED p[RttParam.device_type] = RttPeerType.PEER_TYPE_AP p[RttParam.preamble] = RttPreamble.PREAMBLE_VHT p[RttParam.bandwidth] = RttBW.BW_80_SUPPORT p[RttParam.frequency] = self.vht80_5g[WifiEnums.frequency_key] p[RttParam.center_freq0] = self.vht80_5g[RttParam.center_freq0] results = self.get_rtt_results([p]) asserts.assert_true(results, "Did not get any result.") self.log.info(pprint.pformat(results)) def test_rtt_ranging_single_AP_stress(self): """Stress test for Rtt against one AP. Steps: 1. Do RTT ranging against the self.vht80_5g BSSID. 2. Repeat self.stress_num times. 3. Verify RTT results. """ p = {} p[RttParam.request_type] = RttType.TYPE_TWO_SIDED p[RttParam.device_type] = RttPeerType.PEER_TYPE_AP p[RttParam.preamble] = RttPreamble.PREAMBLE_VHT p[RttParam.bandwidth] = RttBW.BW_80_SUPPORT p[RttParam.BSSID] = self.vht80_5g[WifiEnums.BSSID_KEY] p[RttParam.frequency] = self.vht80_5g[WifiEnums.frequency_key] p[RttParam.center_freq0] = self.vht80_5g[RttParam.center_freq0] p[RttParam.channel_width] = ScanResult.CHANNEL_WIDTH_80MHZ all_results = [] for i in range(self.stress_num): self.log.info("RTT Ranging iteration %d" % (i + 1)) results = self.get_rtt_results([p]) if results: all_results += results else: self.log.warning("Did not get result for iteration %d." % i) frate = self.process_rtt_events(all_results) def test_regular_scan_then_rtt_ranging_stress(self): """Stress test for regular scan then start rtt ranging against the RTT compatible networks found by the scan. Steps: 1. Start a WiFi connection scan. 2. Get scan results. 3. Find all the 11mc capable BSSIDs and choose the ones to use (self.network_selector) 4. Do RTT ranging against the selected BSSIDs, with the info from the scan results. 5. Repeat self.stress_num times. 6. Verify RTT results. """ scan_func = self.regular_scan_for_rtt_networks self.scan_then_rtt_ranging_stress_logic(scan_func) def test_gscan_then_rtt_ranging_stress(self): """Stress test for gscan then start rtt ranging against the RTT compatible networks found by the scan. Steps: 1. Start a WifiScanner single shot scan on all channels. 2. Wait for full scan results of the expected 11mc capable BSSIDs. 3. Wait for single shot scan to finish on all channels. 4. Do RTT ranging against the selected BSSIDs, with the info from the scan results. 5. Repeat self.stress_num times. 6. Verify RTT results. """ scan_func = self.gscan_for_rtt_networks self.scan_then_rtt_ranging_stress_logic(scan_func)