acts/tests/google/ble/system_tests/BleStressTest.py

# python3.4
# Copyright (C) 2014 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.

"""
Basic LE Stress tests.
"""

import concurrent
import pprint
import time

from queue import Empty
from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
from acts.test_utils.bt.bt_test_utils import generate_ble_advertise_objects
from acts.test_utils.bt.bt_test_utils import generate_ble_scan_objects
from acts.test_utils.bt.bt_test_utils import get_advanced_droid_list
from acts.test_utils.bt.bt_test_utils import reset_bluetooth
from acts.test_utils.bt.bt_test_utils import scan_result

class BleStressTest(BluetoothBaseTest):
    tests = None
    default_timeout = 10

    def __init__(self, controllers):
        BluetoothBaseTest.__init__(self, controllers)
        self.droid_list = get_advanced_droid_list(self.droids, self.eds)
        self.scn_droid, self.scn_ed = self.droids[0], self.eds[0]
        self.adv_droid, self.adv_ed = self.droids[1], self.eds[1]
        self.tests = (
            "test_loop_scanning_1000",
            "test_restart_scan_callback_after_bt_toggle",
            "test_loop_scanning_100_verify_no_hci_timeout",
            "test_start_le_scan_while_toggling_bt",
        )
        if self.droid_list[0]['max_advertisements'] > 0:
            self.tests = self.tests + ("test_loop_advertising_100",
                                       "test_restart_advertise_callback_after_bt_toggle",)

    def bleadvertise_verify_onsuccess_handler(self, event):
        test_result = True
        self.log.debug("Verifying onSuccess event")
        self.log.debug(pprint.pformat(event))
        return test_result

    @BluetoothBaseTest.bt_test_wrap
    def test_loop_scanning_1000(self):
        """Stress start/stop scan instances.

        This test will start and stop scan instances as fast as possible. This
        will guarantee that the scan instances are properly being cleaned up
        when the scan is stopped.

        Steps:
        1. Start a scan instance.
        2. Stop the scan instance.
        3. Repeat steps 1-2 1000 times.

        Expected Result:
        Neither starting or stopping scan instances causes any failures.

        Returns:
          Pass if True
          Fail if False

        TAGS: LE, Scanning, Stress
        Priority: 1
        """
        scan_droid, scan_event_dispatcher = self.droid, self.ed
        test_result = True
        for _ in range(1000):
            filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
                scan_droid)
            self.scn_droid.bleStartBleScan(
                filter_list, scan_settings, scan_callback)
            self.scn_droid.bleStopBleScan(scan_callback)
        return test_result

    @BluetoothBaseTest.bt_test_wrap
    def test_loop_scanning_100_verify_no_hci_timeout(self):
        """Stress start/stop scan instances variant.

        This test will start and stop scan instances with a one second timeout
        in between each iteration. This testcase was added because the specific
        timing combination caused hci timeouts.

        Steps:
        1. Start a scan instance.
        2. Stop the scan instance.
        3. Sleep for 1 second.
        4. Repeat steps 1-3 100 times.

        Expected Result:
        Neither starting or stopping scan instances causes any failures.

        Returns:
          Pass if True
          Fail if False

        TAGS: LE, Scanning, Stress
        Priority: 1
        """
        adv_callback, adv_data, adv_settings = generate_ble_advertise_objects(
            self.adv_droid)
        self.adv_droid.bleStartBleAdvertising(
            adv_callback, adv_data, adv_settings)
        test_result = True
        for _ in range(100):
            filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
                self.scn_droid)
            self.scn_droid.bleStartBleScan(
                filter_list, scan_settings, scan_callback)
            self.log.info(self.scn_ed.pop_event(
                scan_result.format(scan_callback)))
            self.scn_droid.bleStopBleScan(scan_callback)
            time.sleep(1)
        return test_result

    @BluetoothBaseTest.bt_test_wrap
    def test_loop_advertising_100(self):
        """Stress start/stop advertising instances.

        This test will start and stop advertising instances as fast as possible.

        Steps:
        1. Start a advertising instance.
        2. Find that an onSuccess callback is triggered.
        3. Stop the advertising instance.
        4. Repeat steps 1-3 100 times.

        Expected Result:
        Neither starting or stopping advertising instances causes any failures.

        Returns:
          Pass if True
          Fail if False

        TAGS: LE, Advertising, Stress
        Priority: 1
        """
        advertise_droid, advertise_event_dispatcher = self.droid, self.ed
        test_result = True
        for _ in range(100):
            advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects(
                advertise_droid)
            advertise_droid.bleStartBleAdvertising(
                advertise_callback, advertise_data, advertise_settings)
            expected_advertise_event_name = "".join(
                ["BleAdvertise", str(advertise_callback), "onSuccess"])
            worker = advertise_event_dispatcher.handle_event(
                self.bleadvertise_verify_onsuccess_handler, expected_advertise_event_name, (
                    []),
                self.default_timeout)
            try:
                self.log.debug(worker.result(self.default_timeout))
            except Empty as error:
                self.log.debug(
                    " ".join(["Test failed with Empty error:", str(error)]))
                test_result = False
            except concurrent.futures._base.TimeoutError as error:
                self.log.debug(
                    " ".join(["Test failed, filtering callback onSuccess never occurred:",
                              str(error)]))
                test_result = False
            advertise_droid.bleStopBleAdvertising(advertise_callback)
        return test_result

    @BluetoothBaseTest.bt_test_wrap
    def test_restart_advertise_callback_after_bt_toggle(self):
        """Test to reuse an advertise callback.

        This will verify if advertising objects can be reused after a bluetooth
        toggle.

        Steps:
        1. Start a advertising instance.
        2. Find that an onSuccess callback is triggered.
        3. Stop the advertising instance.
        4. Toggle bluetooth off and on.
        5. Start an advertising instance on the same objects used in step 1.
        6. Find that an onSuccess callback is triggered.

        Expected Result:
        Advertisement should start successfully.

        Returns:
          Pass if True
          Fail if False

        TAGS: LE, Advertising, Stress
        Priority: 1
        """
        test_result = True
        advertise_droid, advertise_event_dispatcher = self.droid, self.ed
        advertise_callback, advertise_data, advertise_settings = generate_ble_advertise_objects(
            advertise_droid)
        advertise_droid.bleStartBleAdvertising(
            advertise_callback, advertise_data, advertise_settings)
        expected_advertise_event_name = "".join(
            ["BleAdvertise", str(advertise_callback), "onSuccess"])
        worker = advertise_event_dispatcher.handle_event(
            self.bleadvertise_verify_onsuccess_handler, expected_advertise_event_name, (
                []),
            self.default_timeout)
        try:
            self.log.debug(worker.result(self.default_timeout))
        except Empty as error:
            self.log.debug(
                " ".join(["Test failed with Empty error:", str(error)]))
            test_result = False
        except concurrent.futures._base.TimeoutError as error:
            self.log.debug(
                " ".join(["Test failed, filtering callback onSuccess never occurred:",
                          str(error)]))
        test_result = reset_bluetooth([self.droids[0]], [self.eds[0]])
        if not test_result:
            return test_result
        time.sleep(5)
        advertise_droid.bleStartBleAdvertising(
            advertise_callback, advertise_data, advertise_settings)
        worker = advertise_event_dispatcher.handle_event(
            self.bleadvertise_verify_onsuccess_handler, expected_advertise_event_name, (
                []),
            self.default_timeout)
        try:
            self.log.debug(worker.result(self.default_timeout))
        except Empty as error:
            self.log.debug(
                " ".join(["Test failed with Empty error:", str(error)]))
            test_result = False
        except concurrent.futures._base.TimeoutError as error:
            self.log.debug(
                " ".join(["Test failed, filtering callback onSuccess never occurred:",
                          str(error)]))
        return test_result

    @BluetoothBaseTest.bt_test_wrap
    def test_restart_scan_callback_after_bt_toggle(self):
        """Test to reuse an scan callback.

        This will verify if scan objects can be reused after a bluetooth
        toggle.

        Steps:
        1. Start a scanning instance.
        3. Stop the scanning instance.
        4. Toggle bluetooth off and on.
        5. Start an scanning instance on the same objects used in step 1.

        Expected Result:
        Scanner should start successfully.

        Returns:
          Pass if True
          Fail if False

        TAGS: LE, Scanning, Stress
        Priority: 1
        """
        test_result = True
        scan_droid, scan_event_dispatcher = self.droid, self.ed
        filter_list, scan_settings, scan_callback = generate_ble_scan_objects(
            scan_droid)
        self.scn_droid.bleStartBleScan(
            filter_list, scan_settings, scan_callback)
        reset_bluetooth([scan_droid], [scan_event_dispatcher])
        self.scn_droid.bleStartBleScan(
            filter_list, scan_settings, scan_callback)

        return test_result