# 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

    # 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

    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)


    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 kernel 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


    @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 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)

        self.results = {}

        if cancel_event is None:
            self.results = {'Cancel event': cancel_event}
            return False

        canceled_time = cancel_event + self.INTERLEAVE_SCAN_CANCEL_TOLERANCE

        suspend_records = [r for r in records if r['time'] < canceled_time]
        resume_records = [r for r in records if r['time'] >= canceled_time]

        if len(suspend_records) == 0:
            self.results = {'non-empty records before suspended': False}
            return False

        if len(resume_records) == 0:
            self.results = {'non-empty records after resumed': False}
            return False

        # Records are stored chronologically.
        # The first record is the earliest, while the last one is the oldest.
        last_suspend = suspend_records[-1]['time']
        first_resume = resume_records[0]['time']
        suspend_time = first_resume - last_suspend

        # 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 = {
                'Paused enough time': suspend_time >= interleave_period
        }

        return all(self.results.values())

    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)

    def advmon_test_interleaved_scan(self):
        """ Test cases for verifying interleave scan """

        # 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.TurnOff()
        # 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_start_discovery()
        self.test_interleaving_state(False, cycles=EXPECT_FALSE_TEST_CYCLE)

        # Interleaving with allowlist should get resumed after stopping scan
        self.test_stop_discovery()
        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)