| # 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 |
| |
| |
| 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 |
| |
| test_case_log = bluetooth_adapter_tests.test_case_log |
| test_retry_and_log = bluetooth_adapter_tests.test_retry_and_log |
| |
| |
| 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) |
| |
| |
| @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 |
| |
| |
| @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 advmon_test_monitor_creation(self): |
| """Test case: MONITOR_CREATION |
| |
| Validate register/unregister app and create/remove monitor. |
| |
| """ |
| # 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. |
| |
| """ |
| # 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_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_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_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_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_fg_bg_combination(self): |
| """Test case: FG_BG_COMBINATION |
| |
| Verify background scanning and foreground scanning do not interfere |
| working of each other. |
| |
| """ |
| 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) |
