| # Copyright 2020 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Server side bluetooth tests on Advertisement Monitor API""" |
| |
| import time |
| import logging |
| import array |
| |
| from autotest_lib.client.bin import utils |
| from autotest_lib.server.cros.bluetooth import bluetooth_adapter_tests |
| from autotest_lib.client.common_lib import error |
| |
| |
| class TestMonitor(): |
| """Local object hosting the test values for Advertisement Monitor object. |
| |
| This class holds the values of parameters for creating an Advertisement |
| Monitor object. |
| |
| """ |
| |
| # Index of the pattern data in the patterns filter. |
| PATTERN_DATA_IDX = 2 |
| |
| def __init__(self, app_id): |
| """Construction of a local monitor object. |
| |
| @param app_id: the app id associated with the monitor. |
| |
| """ |
| self.type = None |
| self.rssi = [] |
| self.patterns = [] |
| self.monitor_id = None |
| self.app_id = app_id |
| |
| |
| def _bytes(self, str_data): |
| """Convert string data to byte array. |
| |
| @param str_data: the string data. |
| |
| @returns: the byte array. |
| |
| """ |
| return [b for b in array.array('B', str_data)] |
| |
| |
| def update_type(self, monitor_type): |
| """Update the monitor type. |
| |
| @param monitor_type: type of the monitor. |
| |
| """ |
| self.type = monitor_type |
| |
| |
| def update_rssi(self, monitor_rssi): |
| """Update the RSSI filter values. |
| |
| @param rssi: the list of rssi threshold and timeout values. |
| |
| """ |
| self.rssi = monitor_rssi |
| |
| |
| def update_patterns(self, monitor_patterns): |
| """Update the content filter patterns. |
| |
| @param patterns: the list of start position, ad type and patterns. |
| |
| """ |
| # Convert string patterns to byte array, if any. |
| for pattern in monitor_patterns: |
| if isinstance(pattern[self.PATTERN_DATA_IDX], str): |
| pattern[self.PATTERN_DATA_IDX] = self._bytes( |
| pattern[self.PATTERN_DATA_IDX]) |
| |
| self.patterns = monitor_patterns |
| |
| |
| def update_monitor_id(self, monitor_id): |
| """Store the monitor id returned by add_monitor(). |
| |
| @param monitor_id: the monitor id. |
| |
| """ |
| self.monitor_id = monitor_id |
| |
| |
| def get_monitor_data(self): |
| """Return the monitor parameters. |
| |
| @returns: List containing the monitor data. |
| |
| """ |
| return [self.type, self.rssi, self.patterns] |
| |
| |
| def get_monitor_id(self): |
| """Return the monitor id. |
| |
| @returns: monitor id if monitor is already added, None otherwise. |
| |
| """ |
| return self.monitor_id |
| |
| |
| def get_app_id(self): |
| """Return the application id. |
| |
| @returns: app id associated to the monitor object. |
| |
| """ |
| return self.app_id |
| |
| |
| class BluetoothAdapterAdvMonitorTests( |
| bluetooth_adapter_tests.BluetoothAdapterTests): |
| """Server side bluetooth adapter advertising Test. |
| |
| This class comprises a number of test cases to verify bluetooth |
| Advertisement Monitor API. |
| |
| Refer to the test plan doc for more details: go/bt-advmon-api-test-plan |
| |
| """ |
| |
| ADD_MONITOR_POLLING_TIMEOUT_SECS = 3 |
| ADD_MONITOR_POLLING_SLEEP_SECS = 1 |
| PAIR_TEST_SLEEP_SECS = 5 |
| |
| # Non-zero count value is used to indicate the case where multiple |
| # DeviceFound/DeviceLost events are expected to occur. |
| MULTIPLE_EVENTS = -1 |
| |
| # Number of cycle to observe during a test |
| INTERLEAVE_SCAN_TEST_CYCLE = 10 |
| # Acceptable extra delay of interleave scan duration, in sec |
| INTERLEAVE_SCAN_DURATION_TOLERANCE = 0.1 |
| # Acceptable delay of cancelling interleave scan, in sec |
| INTERLEAVE_SCAN_CANCEL_TOLERANCE = 2 |
| # Acceptable extra/missing cycles in interleave scan |
| INTERLEAVE_SCAN_CYCLE_NUM_TOLERANCE = 2 |
| |
| # Duration of kernel perform 'start discovery', in sec |
| DISCOVERY_DURATION = 10.24 |
| |
| test_case_log = bluetooth_adapter_tests.test_case_log |
| test_retry_and_log = bluetooth_adapter_tests.test_retry_and_log |
| |
| |
| def advmon_check_manager_interface_exist(self): |
| """Check if AdvertisementMonitorManager1 interface is available. |
| |
| @returns: True if Manager interface is available, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_check_manager_interface_exist() |
| |
| |
| def read_supported_types(self): |
| """Read the Advertisement Monitor supported monitor types. |
| |
| @returns: List of supported advertisement monitor types. |
| |
| """ |
| return self.bluetooth_facade.advmon_read_supported_types() |
| |
| |
| def read_supported_features(self): |
| """Read the Advertisement Monitor supported features. |
| |
| @returns: List of supported advertisement monitor features. |
| |
| """ |
| return self.bluetooth_facade.advmon_read_supported_features() |
| |
| |
| def create_app(self): |
| """Create an advertisement monitor app. |
| |
| @returns: app id, once the app is created. |
| |
| """ |
| return self.bluetooth_facade.advmon_create_app() |
| |
| |
| def exit_app(self, app_id): |
| """Exit an advertisement monitor app. |
| |
| @param app_id: the app id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_exit_app(app_id) |
| |
| |
| def kill_app(self, app_id): |
| """Kill an advertisement monitor app by sending SIGKILL. |
| |
| @param app_id: the app id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_kill_app(app_id) |
| |
| |
| def register_app(self, app_id): |
| """Register an advertisement monitor app. |
| |
| @param app_id: the app id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_register_app(app_id) |
| |
| |
| def unregister_app(self, app_id): |
| """Unregister an advertisement monitor app. |
| |
| @param app_id: the app id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_unregister_app(app_id) |
| |
| |
| def add_monitor(self, app_id, monitor_data): |
| """Create an Advertisement Monitor object. |
| |
| @param app_id: the app id. |
| @param monitor_data: the list containing monitor type, RSSI filter |
| values and patterns. |
| |
| @returns: monitor id, once the monitor is created, None otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_add_monitor(app_id, monitor_data) |
| |
| |
| def remove_monitor(self, app_id, monitor_id): |
| """Remove the Advertisement Monitor object. |
| |
| @param app_id: the app id. |
| @param monitor_id: the monitor id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_remove_monitor(app_id, monitor_id) |
| |
| |
| def get_event_count(self, app_id, monitor_id, event='All'): |
| """Read the count of a particular event on the given monitor. |
| |
| @param app_id: the app id. |
| @param monitor_id: the monitor id. |
| @param event: name of the specific event or 'All' for all events. |
| |
| @returns: count of the specific event or dict of counts of all events. |
| |
| """ |
| return self.bluetooth_facade.advmon_get_event_count(app_id, |
| monitor_id, |
| event) |
| |
| |
| def reset_event_count(self, app_id, monitor_id, event='All'): |
| """Reset the count of a particular event on the given monitor. |
| |
| @param app_id: the app id. |
| @param monitor_id: the monitor id. |
| @param event: name of the specific event or 'All' for all events. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_reset_event_count(app_id, |
| monitor_id, |
| event) |
| |
| |
| def interleave_logger_start(self): |
| """ Start interleave logger recording |
| """ |
| self.bluetooth_facade.advmon_interleave_scan_logger_start() |
| |
| def interleave_logger_stop(self): |
| """ Stop interleave logger recording |
| |
| @returns: True if logs were successfully collected, |
| False otherwise. |
| |
| """ |
| return self.bluetooth_facade.advmon_interleave_scan_logger_stop() |
| |
| def interleave_logger_get_records(self): |
| """ Get records in previous log collections |
| |
| @returns: a list of records, where each item is a record of |
| interleave |state| and the |time| the state starts. |
| |state| could be {'no filter', 'allowlist'} |
| |time| is system time in sec |
| |
| """ |
| return self.bluetooth_facade.\ |
| advmon_interleave_scan_logger_get_records() |
| |
| def interleave_logger_get_cancel_event(self): |
| """ Get cancel event in previous log collections |
| |
| @returns: the first cancel event in the collections, |
| None if no cancel event was found |
| |
| """ |
| events = self.bluetooth_facade.\ |
| advmon_interleave_scan_logger_get_cancel_events() |
| if len(events) == 0: |
| event = None |
| else: |
| event = events[0] |
| if len(events) > 1: |
| logging.warning('More than one cancel events found %s', events) |
| return event |
| |
| @test_retry_and_log(False) |
| def test_supported_types(self): |
| """Test supported monitor types. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| supported_types = self.read_supported_types() |
| for supported_type in supported_types: |
| logging.info('type: %s', supported_type) |
| |
| # TODO(b/169658213) - add check for supported types. |
| return True |
| |
| |
| @test_retry_and_log(False) |
| def test_supported_features(self): |
| """Test supported features. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| supported_features = self.read_supported_features() |
| for supported_feature in supported_features: |
| logging.info('feature: %s', supported_feature) |
| |
| # TODO(b/169658213) - add check for supported features. |
| return True |
| |
| |
| def test_is_controller_offloading_supported(self): |
| """Check if controller based RSSI filtering is supported. |
| |
| By default the LE_SCAN_FILTER_DUP flag is enabled on all platforms. |
| Due to this, the host does not receive as many advertisements during |
| passive scanning, which causes SW based RSSI filtering not to work |
| as intended. So, if the controller offloading is not supported, skip |
| the tests that involves RSSI filtering and raise TEST_NA. |
| |
| @raises: TestNA if controller based RSSI filtering is not supported. |
| |
| """ |
| supported_features = self.read_supported_features() |
| if not supported_features: |
| logging.info('Controller offloading not supported') |
| raise error.TestNAError('Controller offloading not supported') |
| |
| |
| def test_is_adv_monitoring_supported(self, require_rssi_filtering = False): |
| """Check if Adv Monitor API is supported. |
| |
| If AdvMonitor API is not supported by the platform, |
| AdvertisementMonitorManager1 interface won't be exposed by |
| bluetoothd. In such case, skip the test and raise TestNA. |
| |
| @param require_rssi_filtering: True if test requires RSSI filtering. |
| |
| @raises: TestNA if Adv Monitor API is not supported or if controller |
| based RSSI filtering is not supported. |
| |
| """ |
| if not self.advmon_check_manager_interface_exist(): |
| logging.info('Advertisement Monitor API not supported') |
| raise error.TestNAError('Advertisement Monitor API not supported') |
| |
| if require_rssi_filtering: |
| self.test_is_controller_offloading_supported() |
| |
| |
| @test_retry_and_log(False) |
| def test_exit_app(self, app_id): |
| """Test exit application. |
| |
| @param app_id: the app id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.exit_app(app_id) |
| |
| |
| @test_retry_and_log(False) |
| def test_kill_app(self, app_id): |
| """Test kill application. |
| |
| @param app_id: the app id. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.kill_app(app_id) |
| |
| |
| @test_retry_and_log(False) |
| def test_register_app(self, app_id, expected=True): |
| """Test register application. |
| |
| @param app_id: the app id. |
| @param expected: expected result of the RegisterMonitor method. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.register_app(app_id) == expected |
| |
| |
| @test_retry_and_log(False) |
| def test_unregister_app(self, app_id, expected=True): |
| """Test unregister application. |
| |
| @param app_id: the app id. |
| @param expected: expected result of the UnregisterMonitor method. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.unregister_app(app_id) == expected |
| |
| |
| @test_retry_and_log(False) |
| def test_monitor_activate(self, monitor, expected): |
| """Test if the Activate method on the monitor has been invoked or not. |
| |
| @param monitor: the local monitor object. |
| @param expected: expected state of the Activate event. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| app_id = monitor.get_app_id() |
| monitor_id = monitor.get_monitor_id() |
| if monitor_id is None: |
| return False |
| |
| def _check_activate(): |
| """Handler for the activate event.""" |
| return self.get_event_count(app_id, monitor_id, 'Activate') == 1 |
| |
| activated = False |
| try: |
| utils.poll_for_condition( |
| condition=_check_activate, |
| timeout=self.ADD_MONITOR_POLLING_TIMEOUT_SECS, |
| sleep_interval=self.ADD_MONITOR_POLLING_SLEEP_SECS, |
| desc='Waiting for activate') |
| activated = True |
| except utils.TimeoutError as e: |
| logging.error('activate: %s', e) |
| except: |
| logging.error('activate: unexpected error') |
| |
| return expected == activated |
| |
| |
| @test_retry_and_log(False) |
| def test_monitor_release(self, monitor, expected): |
| """Test if the Release method on the monitor has been invoked or not. |
| |
| @param monitor: the local monitor object. |
| @param expected: expected state of the Release event. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| app_id = monitor.get_app_id() |
| monitor_id = monitor.get_monitor_id() |
| if monitor_id is None: |
| return False |
| |
| def _check_release(): |
| """Handler for the release event.""" |
| return self.get_event_count(app_id, monitor_id, 'Release') == 1 |
| |
| released = False |
| try: |
| utils.poll_for_condition( |
| condition=_check_release, |
| timeout=self.ADD_MONITOR_POLLING_TIMEOUT_SECS, |
| sleep_interval=self.ADD_MONITOR_POLLING_SLEEP_SECS, |
| desc='Waiting for release') |
| released = True |
| except utils.TimeoutError as e: |
| logging.error('release: %s', e) |
| except Exception as e: |
| logging.error('release: %s', e) |
| except: |
| logging.error('release: unexpected error') |
| |
| return expected == released |
| |
| |
| @test_retry_and_log(False) |
| def test_device_found(self, monitor, count, delay=0): |
| """Test if the DeviceFound method on a monitor has been invoked or not. |
| |
| @param monitor: the local monitor object. |
| @param count: expected count of the DeviceFound events. |
| @param delay: wait until 'delay' seconds before reading the event count. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| app_id = monitor.get_app_id() |
| monitor_id = monitor.get_monitor_id() |
| if monitor_id is None: |
| return False |
| |
| if delay: |
| time.sleep(delay) |
| |
| checked_count = self.get_event_count(app_id, monitor_id, 'DeviceFound') |
| |
| if count == self.MULTIPLE_EVENTS: |
| return checked_count > 1 |
| |
| return checked_count == count |
| |
| |
| @test_retry_and_log(False) |
| def test_device_lost(self, monitor, count, delay=0): |
| """Test if the DeviceLost method on a monitor has been invoked or not. |
| |
| @param monitor: the local monitor object. |
| @param count: expected count of the DeviceLost events. |
| @param delay: wait until 'delay' seconds before reading the event count. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| app_id = monitor.get_app_id() |
| monitor_id = monitor.get_monitor_id() |
| if monitor_id is None: |
| return False |
| |
| if delay: |
| time.sleep(delay) |
| |
| checked_count = self.get_event_count(app_id, monitor_id, 'DeviceLost') |
| |
| if count == self.MULTIPLE_EVENTS: |
| return checked_count > 1 |
| |
| return checked_count == count |
| |
| |
| @test_retry_and_log(False) |
| def test_reset_event_count(self, monitor, event='All'): |
| """Test resetting count of a particular event on the given monitor. |
| |
| @param monitor: the local monitor object. |
| @param event: name of the specific event or 'All' for all events. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| return self.reset_event_count(monitor.get_app_id(), |
| monitor.get_monitor_id(), |
| event) |
| |
| |
| @test_retry_and_log(False) |
| def test_add_monitor(self, monitor, expected_activate=None, |
| expected_release=None): |
| """Test adding a monitor. |
| |
| @param monitor: the local monitor object. |
| @param expected_activate: expected state of the Activate event. |
| @param expected_release: expected state of the Release event. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| app_id = monitor.get_app_id() |
| monitor_id = self.add_monitor(app_id, monitor.get_monitor_data()) |
| if monitor_id is None: |
| return False |
| monitor.update_monitor_id(monitor_id) |
| |
| checked_activate = True |
| if expected_activate is not None: |
| checked_activate = self.test_monitor_activate( |
| monitor, expected_activate) |
| |
| checked_release = True |
| if expected_release is not None: |
| checked_release = self.test_monitor_release( |
| monitor, expected_release) |
| |
| if self.get_event_count(app_id, monitor_id, 'Release') != 0: |
| self.remove_monitor(app_id, monitor_id) |
| monitor.update_monitor_id(None) |
| |
| self.results = { |
| 'activated': checked_activate, |
| 'released': checked_release |
| } |
| return all(self.results.values()) |
| |
| |
| @test_retry_and_log(False) |
| def test_remove_monitor(self, monitor): |
| """Test removing a monitor. |
| |
| @param monitor: the local monitor object. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| app_id = monitor.get_app_id() |
| monitor_id = monitor.get_monitor_id() |
| if monitor_id is None: |
| return False |
| |
| ret = self.remove_monitor(app_id, monitor_id) |
| monitor.update_monitor_id(None) |
| |
| if ret is None: |
| return False |
| |
| return True |
| |
| |
| @test_retry_and_log(False) |
| def test_setup_peer_devices(self): |
| """Test availability of the peer devices. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| self.peer_keybd = None |
| self.peer_mouse = None |
| |
| for device_type, device_list in self.devices.items(): |
| for device in device_list: |
| if device_type is 'BLE_KEYBOARD': |
| self.peer_keybd = device |
| elif device_type is 'BLE_MOUSE': |
| self.peer_mouse = device |
| |
| if self.peer_keybd is not None and self.peer_mouse is not None: |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| |
| self.results = { |
| 'keybd': self.peer_keybd is not None, |
| 'mouse': self.peer_mouse is not None |
| } |
| return all(self.results.values()) |
| |
| |
| @test_retry_and_log(False) |
| def test_start_peer_device_adv(self, device, duration=0): |
| """Test enabling the peer device advertisements. |
| |
| @param device: the device object. |
| @param duration: the duration of the advertisement. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| ret = self.test_device_set_discoverable(device, True) |
| |
| if duration: |
| time.sleep(duration) |
| |
| return ret |
| |
| |
| @test_retry_and_log(False) |
| def test_stop_peer_device_adv(self, device, duration=0): |
| """Test disabling the peer device advertisements. |
| |
| @param device: the device object. |
| @param duration: the duration of the advertisement disable. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| ret = self.test_device_set_discoverable(device, False) |
| |
| if duration: |
| time.sleep(duration) |
| |
| return ret |
| |
| def check_records_interleaving(self, durations, records, expect_cycles): |
| """ Check the state of records is interleaving and also the duration is |
| as expected. |
| |
| @param durations: a dict of {'allowlist': allowlist_duration, |
| 'no filter': no_filter_duration} |
| @param records: a list of records |
| |
| @returns: a dict of {'Interleaved': record_state_is_interleaved, |
| 'Span within range': duration_is_expected} |
| |
| """ |
| |
| actual_cycle = len(records) / len(durations.keys()) |
| offset = self.INTERLEAVE_SCAN_CYCLE_NUM_TOLERANCE |
| expect_cycle_lowerbound = max(1, expect_cycles - offset) |
| expect_cycle_upperbound = expect_cycles + offset |
| enough_cycle_num = (actual_cycle >= expect_cycle_lowerbound |
| and actual_cycle <= expect_cycle_upperbound) |
| interleaved = True |
| span_within_range = True |
| expected_state = None |
| |
| def _next_state(state): |
| if state == 'allowlist': |
| return 'no filter' |
| elif state == 'no filter': |
| return 'allowlist' |
| else: |
| logging.warning('Unexpected state %s', state) |
| return None |
| |
| for i, record in enumerate(records): |
| state = record['state'] |
| nstate = _next_state(state) |
| |
| # We can't count span on single data point and expected_state |
| # hasn't set |
| if i != 0: |
| span = (record['time'] - records[i - 1]['time']) |
| |
| if state != expected_state: |
| interleaved = False |
| |
| if span < durations[nstate] -\ |
| self.INTERLEAVE_SCAN_DURATION_TOLERANCE: |
| span_within_range = False |
| |
| if span > durations[nstate] +\ |
| self.INTERLEAVE_SCAN_DURATION_TOLERANCE: |
| span_within_range = False |
| |
| expected_state = nstate |
| |
| return { |
| 'Enough cycle number': enough_cycle_num, |
| 'Interleaved': interleaved, |
| 'Span within range': span_within_range |
| } |
| |
| def check_records_paused(self, records, cancel_event, expect_paused_time, |
| expect_resume): |
| """ Check if the interleave scan is paused |
| |
| @param records: a list of records |
| @param cancel_event: the timestamp interleave was canceled |
| @param expect_paused_time: minimum duration of interleave scan paused |
| @param expect_resume: True if interleave scan should restart, |
| False if ***we don't care*** |
| |
| @returns: a dict of {'Cancel event': (bool), |
| 'Non-empty records before paused': (bool), |
| 'Non-empty records after paused': (bool), |
| 'Paused enough time': (bool) |
| } |
| Note: some entries might not exist if it doesn't make sense |
| in that case. |
| |
| """ |
| |
| result = {} |
| |
| result.update({'Cancel event': cancel_event is not None}) |
| if cancel_event is None: |
| return result |
| |
| canceled_time = cancel_event + self.INTERLEAVE_SCAN_CANCEL_TOLERANCE |
| |
| before_paused_rec = [r for r in records if r['time'] < canceled_time] |
| after_paused_rec = [r for r in records if r['time'] >= canceled_time] |
| |
| result.update({ |
| 'Non-empty records before paused': |
| len(before_paused_rec) != 0 |
| }) |
| |
| if expect_resume: |
| result.update({ |
| 'Non-empty records after paused': |
| len(after_paused_rec) != 0 |
| }) |
| |
| if len(before_paused_rec) > 0 and len(after_paused_rec) > 0: |
| # Records are stored chronologically. |
| last_time_before_paused = before_paused_rec[-1]['time'] |
| first_time_after_paused = after_paused_rec[0]['time'] |
| paused_time = first_time_after_paused - last_time_before_paused |
| result.update( |
| {'Paused enough time': paused_time >= expect_paused_time}) |
| |
| return result |
| |
| def get_interleave_scan_durations(self): |
| """ Get interleave scan duration. |
| |
| @returns: a dict of {'allowlist': allowlist_duration, |
| 'no filter': no_filter_duration} |
| |
| """ |
| |
| # TODO(b/171844106): get this parameters via |
| # MGMT_OP_READ_DEF_SYSTEM_CONFIG |
| durations = {'allowlist': 300, 'no filter': 500} |
| |
| # Change the unit from msec to second for future convenience. |
| durations = {key: value * 0.001 for key, value in durations.items()} |
| return durations |
| |
| @test_retry_and_log(False) |
| def test_interleaving_state(self, |
| expect_true, |
| cycles=INTERLEAVE_SCAN_TEST_CYCLE): |
| """ Test for checking if kernel is doing interleave scan or not. |
| |
| @params expect_true: True if kernel should be running interleave scan |
| False if kernel shouldn't. |
| @params cycles: number of cycles to collect logs |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| durations = self.get_interleave_scan_durations() |
| interleave_period = sum(durations.values()) |
| log_time = interleave_period * cycles |
| self.interleave_logger_start() |
| time.sleep(log_time) |
| self.interleave_logger_stop() |
| records = self.interleave_logger_get_records() |
| |
| logging.debug(records) |
| |
| if not expect_true: |
| self.results = {'No records': len(records) == 0} |
| else: |
| self.results = self.check_records_interleaving( |
| durations, records, cycles) |
| |
| return all(self.results.values()) |
| |
| @test_retry_and_log(False) |
| def test_interleaving_suspend_resume(self): |
| """ Test for checking if kernel paused interleave scan during system |
| suspended. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| durations = self.get_interleave_scan_durations() |
| interleave_period = sum(durations.values()) |
| |
| # make sure suspend time is long enough to verify there is no |
| # interleaving during suspended |
| expect_suspend_time = max(self.SUSPEND_TIME_SECS, |
| 2 * interleave_period) |
| |
| # make sure we'll get some records before/after system suspended |
| extra_sleep_time = 2 * interleave_period |
| |
| self.interleave_logger_start() |
| time.sleep(extra_sleep_time) |
| self.suspend_resume(suspend_time=expect_suspend_time) |
| time.sleep(extra_sleep_time) |
| self.interleave_logger_stop() |
| records = self.interleave_logger_get_records() |
| cancel_event = self.interleave_logger_get_cancel_event() |
| |
| logging.debug(records) |
| logging.debug(cancel_event) |
| |
| # Currently resume time is not very reliable. It is likely the actual |
| # time in sleeping is less than expect_suspend_time. |
| # Check the interleave scan paused for at least one cycle long instead. |
| self.results = self.check_records_paused(records, cancel_event, |
| interleave_period, True) |
| return all(self.results.values()) |
| |
| @test_retry_and_log(False) |
| def test_interleaving_active_scan_cycle(self): |
| """ Test for checking if kernel paused interleave scan during active |
| scan. |
| |
| @returns: True on success, False otherwise. |
| |
| """ |
| durations = self.get_interleave_scan_durations() |
| interleave_period = sum(durations.values()) |
| |
| # make sure we'll get some records before/after active scan |
| extra_sleep_time = 2 * interleave_period |
| |
| self.interleave_logger_start() |
| time.sleep(extra_sleep_time) |
| self.test_start_discovery() |
| time.sleep(extra_sleep_time + self.INTERLEAVE_SCAN_CANCEL_TOLERANCE) |
| self.interleave_logger_stop() |
| records = self.interleave_logger_get_records() |
| cancel_event = self.interleave_logger_get_cancel_event() |
| |
| logging.debug(records) |
| logging.debug(cancel_event) |
| |
| # BlueZ pauses discovery for every DISCOVERY_DURATION then restarts it |
| # 5 seconds later. Interleave scan also get restarted during the paused |
| # time. |
| self.results = self.check_records_paused(records, cancel_event, |
| self.DISCOVERY_DURATION, |
| False) |
| self.test_stop_discovery() |
| return all(self.results.values()) |
| |
| def advmon_test_monitor_creation(self): |
| """Test case: MONITOR_CREATION |
| |
| Validate register/unregister app and create/remove monitor. |
| |
| """ |
| self.test_is_adv_monitoring_supported() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_rssi([-40, 5, -60, 5]) |
| monitor1.update_patterns([ |
| [0, 0x19, [0xc2, 0x03]], |
| ]) |
| |
| monitor2 = TestMonitor(app1) |
| monitor2.update_type('or_patterns') |
| monitor2.update_rssi([-40, 10, -60, 10]) |
| monitor2.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| |
| # Read supported types and features, should not fail. |
| self.test_supported_types() |
| self.test_supported_features() |
| |
| # Activate/Release should not get called. |
| self.test_add_monitor(monitor1, |
| expected_activate=False, |
| expected_release=False) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| # Already registered app path, should fail with AlreadyExists. |
| self.test_register_app(app1, expected=False) |
| |
| # Activate should get called for the monitor added before register app. |
| self.test_monitor_activate(monitor1, expected=True) |
| |
| # Correct monitor parameters, activate should get called. |
| self.test_add_monitor(monitor2, expected_activate=True) |
| |
| # Remove a monitor, should not fail. |
| self.test_remove_monitor(monitor1) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Already unregistered app path, should fail with DoesNotExists. |
| self.test_unregister_app(app1, expected=False) |
| |
| # Release should get called for a monitor not removed before unregister. |
| self.test_monitor_release(monitor2, expected=True) |
| |
| # Remove another monitor, should not fail. |
| self.test_remove_monitor(monitor2) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_monitor_validity(self): |
| """Test case: MONITOR_VALIDITY |
| |
| Validate monitor parameters - monitor type, patterns, RSSI filter |
| values. |
| |
| """ |
| self.test_is_adv_monitoring_supported() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('incorrect_pattern') |
| monitor1.update_rssi([-40, 5, -60, 5]) |
| monitor1.update_patterns([ |
| [0, 0x19, [0xc2, 0x03]], |
| ]) |
| |
| monitor2 = TestMonitor(app1) |
| monitor2.update_type('or_patterns') |
| monitor2.update_rssi([-40, 10, -60, 10]) |
| monitor2.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| # Incorrect monitor type, release should get called. |
| self.test_add_monitor(monitor1, expected_release=True) |
| |
| # Incorrect rssi parameters, release should get called. |
| monitor2.update_rssi([-40, 0, -60, 10]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_rssi([-40, 10, -60, 0]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_rssi([40, 10, -60, 10]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_rssi([-140, 10, -60, 10]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_rssi([-40, 10, 60, 10]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_rssi([-40, 10, -160, 10]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_rssi([-60, 10, -40, 10]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| # Unset the rssi filter parameters. |
| monitor2.update_rssi([127, 0, 127, 0]) |
| |
| # Incorrect pattern parameters, release should get called. |
| monitor2.update_patterns([ |
| [32, 0x09, 'MOUSE'], |
| ]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_patterns([ |
| [0, 0x00, 'MOUSE'], |
| ]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_patterns([ |
| [0, 0x40, 'MOUSE'], |
| ]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_patterns([ |
| [0, 0x09, '0123456789ABCDEF0123456789ABCDEF0'], |
| ]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_patterns([ |
| [32, 0x09, [0xc2, 0x03]], |
| [0, 3, [0x12, 0x18]], |
| ]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| monitor2.update_patterns([ |
| [0, 0x19, [0xc2, 0x03]], |
| [0, 0x00, [0x12, 0x18]], |
| ]) |
| self.test_add_monitor(monitor2, expected_release=True) |
| |
| # Correct pattern parameters, activate should get called. |
| monitor2.update_patterns([ |
| [0, 0x09, 'MOUSE'], |
| ]) |
| self.test_add_monitor(monitor2, expected_activate=True) |
| self.test_remove_monitor(monitor2) |
| |
| monitor2.update_rssi([-40, 10, -60, 10]) |
| monitor2.update_patterns([ |
| [0, 0x19, [0xc2, 0x03]], |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| self.test_add_monitor(monitor2, expected_activate=True) |
| self.test_remove_monitor(monitor2) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_pattern_filter_1(self): |
| """Test case: PATTERN_FILTER_1 |
| |
| Verify matching of advertisements w.r.t. various pattern values and |
| different AD Data Types - Local Name Service UUID and Device Type. |
| |
| """ |
| self.test_is_adv_monitoring_supported(require_rssi_filtering = True) |
| self.test_setup_peer_devices() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_rssi([-60, 3, -80, 3]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| monitor1.update_patterns([ |
| [5, 0x09, '_REF'], |
| ]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Local name 'KEYBD_REF' should match. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=1) |
| |
| # Local name 'MOUSE_REF' should match. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| self.test_device_found(monitor1, count=2) |
| |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| self.test_remove_monitor(monitor1) |
| |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Service UUID 0x1812 should match. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=1) |
| |
| # Service UUID 0x1812 should match. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| self.test_device_found(monitor1, count=2) |
| |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| self.test_remove_monitor(monitor1) |
| |
| monitor1.update_patterns([ |
| [0, 0x19, [0xc1, 0x03]], |
| [0, 0x09, 'MOUSE'], |
| ]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Device type 0xc103 should match. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=1) |
| |
| # Local name 'MOUSE_REF' should match. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| self.test_device_found(monitor1, count=2) |
| |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| self.test_remove_monitor(monitor1) |
| |
| monitor1.update_patterns([ |
| [0, 0x19, [0xc1, 0x03]], |
| [0, 0x19, [0xc3, 0x03]], |
| ]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Device type 0xc103 should match. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=1) |
| |
| # Device type 0xc203 should not match. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| self.test_device_found(monitor1, count=1) |
| |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| self.test_remove_monitor(monitor1) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_rssi_filter_1(self): |
| """Test case: RSSI_FILTER_1 |
| |
| Verify unset RSSI filter and filter with no matching RSSI values. |
| |
| """ |
| self.test_is_adv_monitoring_supported(require_rssi_filtering = True) |
| self.test_setup_peer_devices() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| monitor1.update_rssi([127, 0, 127, 0]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Unset RSSI filter, adv should match multiple times. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=self.MULTIPLE_EVENTS) |
| |
| # Unset RSSI filter, DeviceLost should not get triggered. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_lost(monitor1, count=0) |
| |
| self.test_remove_monitor(monitor1) |
| |
| monitor1.update_rssi([-10, 5, -20, 5]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Adv RSSI lower than RSSI filter, DeviceFound should not get triggered. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=10) |
| self.test_device_found(monitor1, count=0) |
| |
| # No device was found earlier, so DeviceLost should not get triggered. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=10) |
| self.test_device_lost(monitor1, count=0) |
| |
| self.test_remove_monitor(monitor1) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_rssi_filter_2(self): |
| """Test case: RSSI_FILTER_2 |
| |
| Verify RSSI filter matching with multiple peer devices. |
| |
| """ |
| self.test_is_adv_monitoring_supported(require_rssi_filtering = True) |
| self.test_setup_peer_devices() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| monitor1.update_rssi([-60, 3, -80, 3]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # DeviceFound should get triggered only once per device. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=10) |
| self.test_device_found(monitor1, count=1) |
| |
| # DeviceFound should get triggered for another device. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=10) |
| self.test_device_found(monitor1, count=2) |
| |
| # DeviceLost should get triggered only once per device. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=10) |
| self.test_device_lost(monitor1, count=1) |
| |
| # DeviceLost should get triggered for another device. |
| self.test_stop_peer_device_adv(self.peer_mouse, duration=10) |
| self.test_device_lost(monitor1, count=2) |
| |
| self.test_remove_monitor(monitor1) |
| |
| monitor1.update_rssi([-60, 10, -80, 10]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Device was online for short period of time, so DeviceFound should |
| # not get triggered. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=0) |
| |
| # Device did not come back online, DeviceFound should not get triggered. |
| # No device was found earlier, so DeviceLost should not get triggered. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=15) |
| self.test_device_found(monitor1, count=0) |
| self.test_device_lost(monitor1, count=0) |
| |
| self.test_remove_monitor(monitor1) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_rssi_filter_3(self): |
| """Test case: RSSI_FILTER_3 |
| |
| Verify reset of RSSI timers based on advertisements. |
| |
| """ |
| self.test_is_adv_monitoring_supported(require_rssi_filtering = True) |
| self.test_setup_peer_devices() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| monitor1.update_rssi([-60, 10, -80, 10]) |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # DeviceFound should not get triggered before timeout. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=0) |
| |
| # DeviceFound should not get triggered as device went offline. |
| # No device was found earlier, so DeviceLost should not get triggered. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=10) |
| self.test_device_found(monitor1, count=0) |
| self.test_device_lost(monitor1, count=0) |
| |
| # Timer should get reset, so DeviceFound should not get triggered. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=0) |
| |
| # DeviceFound should get triggered once timer completes. |
| self.test_device_found(monitor1, count=1, delay=10) |
| |
| # DeviceLost should not get triggered before timeout. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_lost(monitor1, count=0) |
| |
| # Timer should get reset, so DeviceLost should not get triggered. |
| # DeviceFound should not get triggered as device is not lost yet. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_lost(monitor1, count=0) |
| self.test_device_found(monitor1, count=1) |
| |
| # Timer should get reset, so DeviceLost should not get triggered. |
| self.test_stop_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_lost(monitor1, count=0) |
| |
| # DeviceLost should get triggered once timer completes. |
| self.test_device_lost(monitor1, count=1, delay=10) |
| |
| self.test_remove_monitor(monitor1) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_multi_client(self): |
| """Test case: MULTI_CLIENT |
| |
| Verify working of patterns filter and RSSI filters with multiple |
| clients and multiple monitors. |
| |
| """ |
| self.test_is_adv_monitoring_supported(require_rssi_filtering = True) |
| self.test_setup_peer_devices() |
| |
| # Create two test app instances. |
| app1 = self.create_app() |
| app2 = self.create_app() |
| |
| # Register both apps, should not fail. |
| self.test_register_app(app1) |
| self.test_register_app(app2) |
| |
| # Monitors with same pattern and RSSI filter values in both apps. |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| [0, 0x19, [0xc1, 0x03]], |
| ]) |
| monitor1.update_rssi([-60, 3, -80, 3]) |
| |
| monitor2 = TestMonitor(app2) |
| monitor2.update_type('or_patterns') |
| monitor2.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| [0, 0x19, [0xc1, 0x03]], |
| ]) |
| monitor2.update_rssi([-60, 3, -80, 3]) |
| |
| # Activate should get invoked. |
| self.test_add_monitor(monitor1, expected_activate=True) |
| self.test_add_monitor(monitor2, expected_activate=True) |
| |
| # DeviceFound should get triggered for keyboard. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=1) |
| self.test_device_found(monitor2, count=1) |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| |
| # Remove a monitor from one app. |
| self.test_remove_monitor(monitor1) |
| |
| # Monitors with same pattern but different RSSI filter values. |
| monitor3 = TestMonitor(app1) |
| monitor3.update_type('or_patterns') |
| monitor3.update_patterns([ |
| [0, 0x19, [0xc2, 0x03]], |
| ]) |
| monitor3.update_rssi([-60, 3, -80, 3]) |
| |
| monitor4 = TestMonitor(app2) |
| monitor4.update_type('or_patterns') |
| monitor4.update_patterns([ |
| [0, 0x19, [0xc2, 0x03]], |
| ]) |
| monitor4.update_rssi([-60, 10, -80, 10]) |
| |
| # Activate should get invoked. |
| self.test_add_monitor(monitor3, expected_activate=True) |
| self.test_add_monitor(monitor4, expected_activate=True) |
| |
| # DeviceFound should get triggered for mouse. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| self.test_device_found(monitor2, count=2) |
| self.test_device_found(monitor3, count=1) |
| |
| # Since the RSSI timeouts are different for monitor4, DeviceFound |
| # event should get triggered after total of 10 seconds. |
| self.test_device_found(monitor4, count=0) |
| self.test_device_found(monitor4, count=1, delay=5) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| |
| # Unregister both apps, should not fail. |
| self.test_unregister_app(app1) |
| self.test_unregister_app(app2) |
| |
| # Terminate the both test app instances. |
| self.test_exit_app(app1) |
| self.test_exit_app(app2) |
| |
| |
| def advmon_test_fg_bg_combination(self): |
| """Test case: FG_BG_COMBINATION |
| |
| Verify background scanning and foreground scanning do not interfere |
| working of each other. |
| |
| """ |
| self.test_is_adv_monitoring_supported() |
| self.test_setup_peer_devices() |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| monitor1.update_rssi([127, 0, 127, 0]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| # Activate should get invoked. |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Pair/connect LE Mouse. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| time.sleep(self.PAIR_TEST_SLEEP_SECS) |
| self.test_discover_device(self.peer_mouse.address) |
| time.sleep(self.PAIR_TEST_SLEEP_SECS) |
| self.test_pairing(self.peer_mouse.address, self.peer_mouse.pin) |
| time.sleep(self.PAIR_TEST_SLEEP_SECS) |
| self.test_connection_by_adapter(self.peer_mouse.address) |
| self.test_connection_by_device(self.peer_mouse) |
| |
| # DeviceFound should get triggered for keyboard. |
| self.test_reset_event_count(monitor1) |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=self.MULTIPLE_EVENTS) |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| |
| # Start foreground scanning. |
| self.test_start_discovery() |
| |
| # Disconnect LE mouse. |
| self.test_disconnection_by_device(self.peer_mouse) |
| |
| # Remove the monitor. |
| self.test_remove_monitor(monitor1) |
| |
| # Activate should get invoked. |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # Connect LE mouse. |
| self.test_connection_by_device(self.peer_mouse) |
| |
| # DeviceFound should get triggered for keyboard. |
| self.test_reset_event_count(monitor1) |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=self.MULTIPLE_EVENTS) |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| |
| # Stop foreground scanning. |
| self.test_stop_discovery() |
| |
| # Disconnect LE mouse. |
| self.test_disconnection_by_device(self.peer_mouse) |
| |
| # DeviceFound should get triggered for keyboard. |
| self.test_reset_event_count(monitor1) |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor1, count=self.MULTIPLE_EVENTS) |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| |
| # Remove the monitor. |
| self.test_remove_monitor(monitor1) |
| |
| # Connect LE mouse. |
| self.test_connection_by_device(self.peer_mouse) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
| |
| |
| def advmon_test_suspend_resume(self): |
| """Test case: SUSPEND_RESUME |
| |
| Verify working of background scanning with suspend/resume. |
| |
| """ |
| self.test_is_adv_monitoring_supported(require_rssi_filtering = True) |
| self.test_setup_peer_devices() |
| |
| # Create two test app instances. |
| app1 = self.create_app() |
| app2 = self.create_app() |
| |
| # Register both apps, should not fail. |
| self.test_register_app(app1) |
| self.test_register_app(app2) |
| |
| # Add monitors in both apps. |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ [0, 0x03, [0x12, 0x18]], ]) |
| monitor1.update_rssi([-60, 3, -80, 3]) |
| |
| monitor2 = TestMonitor(app1) |
| monitor2.update_type('or_patterns') |
| monitor2.update_patterns([ [0, 0x19, [0xc2, 0x03]], ]) |
| monitor2.update_rssi([-60, 10, -80, 10]) |
| |
| monitor3 = TestMonitor(app2) |
| monitor3.update_type('or_patterns') |
| monitor3.update_patterns([ [0, 0x03, [0x12, 0x18]], ]) |
| monitor3.update_rssi([-60, 3, -80, 3]) |
| |
| monitor4 = TestMonitor(app2) |
| monitor4.update_type('or_patterns') |
| monitor4.update_patterns([ [0, 0x19, [0xc2, 0x03]], ]) |
| monitor4.update_rssi([-60, 15, -80, 15]) |
| |
| # Activate should get invoked. |
| self.test_add_monitor(monitor1, expected_activate=True) |
| self.test_add_monitor(monitor2, expected_activate=True) |
| self.test_add_monitor(monitor3, expected_activate=True) |
| self.test_add_monitor(monitor4, expected_activate=True) |
| |
| # DeviceFound for mouse should get triggered only for monitors |
| # satisfying the RSSI timers. |
| self.test_start_peer_device_adv(self.peer_mouse, duration=5) |
| self.test_device_found(monitor1, count=1) |
| self.test_device_found(monitor2, count=0) |
| self.test_device_found(monitor3, count=1) |
| self.test_device_found(monitor4, count=0) |
| |
| # Initiate suspend/resume. |
| self.suspend_resume() |
| |
| # Remove a monitor from one app, shouldn't affect working of other |
| # monitors or apps. |
| self.test_remove_monitor(monitor1) |
| |
| # DeviceFound should get triggered for monitors with higher RSSI timers. |
| self.test_device_found(monitor2, count=1, delay=10) |
| self.test_device_found(monitor4, count=1, delay=5) |
| self.test_stop_peer_device_adv(self.peer_mouse) |
| |
| # Terminate an app, shouldn't affect working of monitors in other apps. |
| self.test_exit_app(app1) |
| |
| # DeviceFound should get triggered for keyboard. |
| self.test_start_peer_device_adv(self.peer_keybd, duration=5) |
| self.test_device_found(monitor3, count=2) |
| self.test_stop_peer_device_adv(self.peer_keybd) |
| |
| # Unregister the running app, should not fail. |
| self.test_unregister_app(app2) |
| |
| # Terminate the running test app instance. |
| self.test_exit_app(app2) |
| |
| |
| def advmon_test_interleaved_scan(self): |
| """ Test cases for verifying interleave scan """ |
| |
| self.test_is_adv_monitoring_supported() |
| |
| # cycles to collect logs for tests expect no interleave scan |
| EXPECT_FALSE_TEST_CYCLE = 3 |
| |
| # Create a test app instance. |
| app1 = self.create_app() |
| |
| monitor1 = TestMonitor(app1) |
| monitor1.update_type('or_patterns') |
| monitor1.update_patterns([ |
| [0, 0x03, [0x12, 0x18]], |
| ]) |
| monitor1.update_rssi([127, 0, 127, 0]) |
| |
| # Register the app, should not fail. |
| self.test_register_app(app1) |
| |
| # Activate should get invoked. |
| self.test_add_monitor(monitor1, expected_activate=True) |
| |
| # No device in allowlist, interleave with idle |
| self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE) |
| |
| # No device in allowlist, interleave with idle, interrupted by active |
| # scan |
| self.test_start_discovery() |
| self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE) |
| self.test_stop_discovery() |
| self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE) |
| |
| # No device in allowlist, interleave with idle, interrupted by suspend |
| # resume |
| self.suspend_resume() |
| self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE) |
| |
| # Pair/connect LE Mouse. |
| device = self.devices['BLE_MOUSE'][0] |
| self.test_discover_device(device.address) |
| time.sleep(self.PAIR_TEST_SLEEP_SECS) |
| self.test_pairing(device.address, device.pin, trusted=True) |
| time.sleep(self.PAIR_TEST_SLEEP_SECS) |
| |
| # BLE_MOUSE in allowlist, interleave with allowlist passive scan |
| self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE) |
| device.AdapterPowerOff() |
| # Make sure the peer is disconnected |
| self.test_device_is_not_connected(device.address) |
| self.test_interleaving_state(True) |
| |
| # Interleaving with allowlist should get paused during active scan |
| self.test_interleaving_active_scan_cycle() |
| |
| # Interleaving with allowlist should get resumed after stopping scan |
| self.test_interleaving_state(True) |
| |
| # Interleaving with allowlist should get paused during system suspend, |
| # get resumed after system awake |
| self.test_interleaving_suspend_resume() |
| self.test_interleaving_state(True) |
| |
| self.test_remove_monitor(monitor1) |
| self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE) |
| |
| # Unregister the app, should not fail. |
| self.test_unregister_app(app1) |
| |
| # Terminate the test app instance. |
| self.test_exit_app(app1) |
