| # Lint as: python2, python3 |
| # Copyright 2020 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """ Server-side bluetooth adapter tests that involve suspend/resume with peers |
| |
| paired and/or connected. |
| |
| Single btpeer tests: |
| - Reconnect on resume test |
| - Classic HID |
| - LE HID |
| - A2DP |
| - Wake from suspend test |
| - Classic HID |
| - LE HID |
| - A2DP shouldn't wake from suspend |
| - Suspend while discovering (discovering should pause and unpause) |
| - Suspend while advertising (advertising should pause and unpause) |
| |
| Multiple btpeer tests: |
| - Reconnect on resume test |
| - One classic HID, One LE HID |
| - Two classic HID |
| - Two LE HID |
| - Wake from suspend test |
| - Two classic HID |
| - Two classic LE |
| """ |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import logging |
| import time |
| |
| from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import A2DP |
| from autotest_lib.server.cros.bluetooth.bluetooth_adapter_tests import ( |
| TABLET_MODELS, SUSPEND_POWER_DOWN_CHIPSETS, SUSPEND_POWER_DOWN_MODELS) |
| from autotest_lib.server.cros.bluetooth.bluetooth_adapter_qr_tests import ( |
| QR_UNSUPPORTED_CHIPSETS, QR_EVENT_PERIOD, BluetoothAdapterQRTests) |
| from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import ( |
| BluetoothAdapterQuickTests) |
| from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import ( |
| PROFILE_CONNECT_WAIT, SUSPEND_SEC, EXPECT_NO_WAKE_SUSPEND_SEC) |
| from six.moves import range |
| |
| test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator |
| batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator |
| |
| STRESS_ITERATIONS = 50 |
| |
| # Intel controllers may have up to 2 seconds deviation. |
| # Use a longer suspend delay to make sure QR events are received during |
| # the pre-suspend phase. |
| SUSPEND_DELAY_FOR_QR_SECS = QR_EVENT_PERIOD + 3 |
| SUSPEND_DELAY_TIMEOUT_FOR_QR_SECS = QR_EVENT_PERIOD * 2 |
| WAKEUP_TIMEOUT_FOR_QR_SECS = 30 |
| RESUME_TIMEOUT_FOR_QR_SECS = (WAKEUP_TIMEOUT_FOR_QR_SECS - |
| SUSPEND_DELAY_TIMEOUT_FOR_QR_SECS) |
| |
| |
| class bluetooth_AdapterSRHealth(BluetoothAdapterQuickTests, |
| BluetoothAdapterQRTests): |
| """Server side bluetooth adapter suspend resume test with peer.""" |
| |
| def _test_keyboard_with_string(self, device): |
| # b/328691072: WCN3991 suspend fails with device tests |
| # TODO: Remove after the root cause is fixed. |
| if self.bluetooth_facade.get_chipset_name() == 'QCA-WCN3991': |
| return self.test_hid_device_created(device.address) |
| return (self.test_hid_device_created(device.address) |
| and self.test_keyboard_input_from_trace(device, "simple_text")) |
| |
| def _test_mouse(self, device): |
| # b/328691072: WCN3991 suspend fails with device tests |
| # TODO: Remove after the root cause is fixed. |
| if self.bluetooth_facade.get_chipset_name() == 'QCA-WCN3991': |
| return self.test_hid_device_created(device.address) |
| return (self.test_hid_device_created(device.address) |
| and self.test_mouse_left_click(device) |
| and self.test_mouse_move_in_xy(device, -60, 100) |
| and self.test_mouse_scroll_down(device, 70) |
| and self.test_mouse_click_and_drag(device, 90, 30)) |
| |
| # --------------------------------------------------------------- |
| # Reconnect after suspend tests |
| # --------------------------------------------------------------- |
| |
| def run_reconnect_device(self, |
| devtuples, |
| iterations=1, |
| auto_reconnect=False): |
| """ Reconnects a device after suspend/resume. |
| |
| @param devtuples: array of tuples consisting of the following |
| * device_type: MOUSE, BLE_MOUSE, etc. |
| * device: meta object for peer device |
| * device_test: Optional; test function to run w/ |
| device (for example, mouse click) |
| @params iterations: number of suspend/resume + reconnect iterations |
| @params auto_reconnect: Expect host to automatically reconnect to peer |
| """ |
| boot_id = self.host.get_boot_id() |
| |
| try: |
| # Set up the device; any failures should assert |
| for _, device, device_test in devtuples: |
| self.assert_discover_and_pair(device) |
| self.assert_on_fail( |
| self.test_device_set_discoverable(device, False)) |
| self.assert_on_fail( |
| self.test_connection_by_adapter(device.address)) |
| |
| # Profile connection may not have completed yet and this will |
| # race with a subsequent disconnection (due to suspend). Use the |
| # device test to force profile connect or wait if no test was |
| # given. |
| if device_test is not None: |
| self.assert_on_fail(device_test(device)) |
| else: |
| time.sleep(PROFILE_CONNECT_WAIT) |
| |
| for it in range(iterations): |
| logging.info('Running iteration {}/{} of suspend reconnection'. |
| format(it + 1, iterations)) |
| |
| # Start the suspend process |
| suspend = self.suspend_async(suspend_time=SUSPEND_SEC) |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| # Trigger suspend, wait for regular resume, verify we can reconnect |
| # and run device specific test |
| self.test_suspend_and_wait_for_sleep(suspend, |
| sleep_timeout=SUSPEND_SEC) |
| self.test_wait_for_resume(boot_id, |
| suspend, |
| resume_timeout=SUSPEND_SEC, |
| test_start_time=start_time) |
| |
| # Only reconnect if we don't expect automatic reconnect. |
| # Let the devices initiate connections before the DUT initiates |
| # auto reconnections. |
| # Complete reconnecting all peers before running device tests. |
| # Otherwise, we may have a race between auto reconnection |
| # from the dut and peer initiated connection. See b/177870286 |
| if not auto_reconnect: |
| for device_type, device, _ in devtuples: |
| if 'BLE' in device_type: |
| # LE can't reconnect without |
| # advertising/discoverable |
| self.test_device_set_discoverable(device, True) |
| # Make sure we're actually connected |
| self.test_device_is_connected(device.address) |
| # Set not discoverable for the next iteration |
| self.test_device_set_discoverable(device, False) |
| else: |
| # Classic requires peer to initiate a connection to |
| # wake up the dut |
| self.test_connection_by_device(device) |
| |
| for _, device, device_test in devtuples: |
| if device_test is not None: |
| device_test(device) |
| |
| finally: |
| for _, device, __ in devtuples: |
| self.test_remove_pairing(device.address) |
| |
| @test_wrapper('Reconnect Classic HID', |
| devices={'MOUSE': 1}, |
| supports_floss=True) |
| def sr_reconnect_classic_hid(self): |
| """ Reconnects a classic HID device after suspend/resume. """ |
| device_type = 'MOUSE' |
| device = self.devices[device_type][0] |
| self.run_reconnect_device([(device_type, device, |
| self._test_mouse)]) |
| |
| @test_wrapper('Reboot and Reconnect Classic HID', |
| devices={'MOUSE': 1}, |
| supports_floss=True) |
| def sr_reboot_and_reconnect_classic_hid(self): |
| """ Reboot and reconnects a classic HID device after suspend/resume. """ |
| device_type = 'MOUSE' |
| device = self.devices[device_type][0] |
| # Reboot the DUT so that it starts with a clean state. |
| logging.info('Rebooting the DUT') |
| self.reboot() |
| self.run_reconnect_device([(device_type, device, |
| self._test_mouse)]) |
| |
| @test_wrapper('Reconnect LE HID', |
| devices={'BLE_MOUSE': 1}, |
| supports_floss=True) |
| def sr_reconnect_le_hid(self): |
| """ Reconnects a LE HID device after suspend/resume. """ |
| device_type = 'BLE_MOUSE' |
| device = self.devices[device_type][0] |
| self.run_reconnect_device([(device_type, device, |
| self._test_mouse)]) |
| |
| # TODO(b/163143005) - Hana can't handle two concurrent HID connections |
| @test_wrapper('Reconnect Multiple Classic HID', |
| devices={ |
| 'MOUSE': 1, |
| 'KEYBOARD': 1 |
| }, |
| skip_models=['hana'], |
| supports_floss=True) |
| def sr_reconnect_multiple_classic_hid(self): |
| """ Reconnects multiple classic HID devices after suspend/resume. """ |
| devices = [('MOUSE', self.devices['MOUSE'][0], |
| self._test_mouse), |
| ('KEYBOARD', self.devices['KEYBOARD'][0], |
| self._test_keyboard_with_string)] |
| self.run_reconnect_device(devices) |
| |
| @test_wrapper('Reconnect Multiple LE HID', |
| devices={ |
| 'BLE_MOUSE': 1, |
| 'BLE_KEYBOARD': 1 |
| }, |
| supports_floss=True) |
| def sr_reconnect_multiple_le_hid(self): |
| """ Reconnects multiple LE HID devices after suspend/resume. """ |
| devices = [('BLE_MOUSE', self.devices['BLE_MOUSE'][0], |
| self._test_mouse), |
| ('BLE_KEYBOARD', self.devices['BLE_KEYBOARD'][0], |
| self._test_keyboard_with_string)] |
| self.run_reconnect_device(devices) |
| |
| @test_wrapper('Reconnect one of each classic+LE HID', |
| devices={ |
| 'BLE_MOUSE': 1, |
| 'KEYBOARD': 1 |
| }, |
| supports_floss=True) |
| def sr_reconnect_multiple_classic_le_hid(self): |
| """ Reconnects one of each classic and LE HID devices after |
| suspend/resume. |
| """ |
| devices = [('BLE_MOUSE', self.devices['BLE_MOUSE'][0], |
| self._test_mouse), |
| ('KEYBOARD', self.devices['KEYBOARD'][0], |
| self._test_keyboard_with_string)] |
| self.run_reconnect_device(devices) |
| |
| @test_wrapper('Reconnect Classic HID Stress Test', devices={'MOUSE': 1}) |
| def sr_reconnect_classic_hid_stress(self): |
| """ Reconnects a classic HID device after suspend/resume. """ |
| device_type = 'MOUSE' |
| device = self.devices[device_type][0] |
| self.run_reconnect_device( |
| [(device_type, device, self._test_mouse)], |
| iterations=STRESS_ITERATIONS) |
| |
| @test_wrapper('Reconnect LE HID Stress Test', devices={'BLE_MOUSE': 1}) |
| def sr_reconnect_le_hid_stress(self): |
| """ Reconnects a LE HID device after suspend/resume. """ |
| device_type = 'BLE_MOUSE' |
| device = self.devices[device_type][0] |
| self.run_reconnect_device( |
| [(device_type, device, self._test_mouse)], |
| iterations=STRESS_ITERATIONS) |
| |
| @test_wrapper('Reconnect A2DP', |
| devices={'BLUETOOTH_AUDIO': 1}, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_reconnect_a2dp(self): |
| """ Reconnects an A2DP device after suspend/resume. """ |
| device_type = 'BLUETOOTH_AUDIO' |
| device = self.devices[device_type][0] |
| self.initialize_bluetooth_audio(device, A2DP) |
| self.run_reconnect_device( |
| [(device_type, device, self.test_device_a2dp_connected)], |
| auto_reconnect=True) |
| |
| |
| @test_wrapper('Peer wakeup Classic HID', |
| devices={'MOUSE': 1}, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_peer_wake_classic_hid(self): |
| """ Use classic HID device to wake from suspend. """ |
| device = self.devices['MOUSE'][0] |
| self.run_peer_wakeup_device('MOUSE', |
| device, |
| device_test=self._test_mouse) |
| |
| @test_wrapper('Peer wakeup dark resume Classic HID', |
| devices={'MOUSE': 1}, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_peer_wake_dark_resume_classic_hid(self): |
| """ Use classic HID device to wake from suspend with dark resume enabled. """ |
| device = self.devices['MOUSE'][0] |
| self.run_peer_wakeup_device('MOUSE', |
| device, |
| device_test=self._test_mouse, |
| dark_resume=True) |
| |
| @test_wrapper('Peer wakeup LE HID', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_peer_wake_le_hid(self): |
| """ Use LE HID device to wake from suspend. """ |
| device = self.devices['BLE_MOUSE'][0] |
| self.run_peer_wakeup_device('BLE_MOUSE', |
| device, |
| device_test=self._test_mouse) |
| |
| @test_wrapper('Peer wakeup dark resume LE HID', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_peer_wake_dark_resume_le_hid(self): |
| """ Use LE HID device to wake from suspend with dark resume enabled. """ |
| device = self.devices['BLE_MOUSE'][0] |
| self.run_peer_wakeup_device('BLE_MOUSE', |
| device, |
| device_test=self._test_mouse, |
| dark_resume=True) |
| |
| @test_wrapper('Peer wakeup LE HID with reconnect LE HID', |
| devices={ |
| 'BLE_MOUSE': 1, |
| 'BLE_KEYBOARD': 1 |
| }, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_peer_wake_le_hid_reconnect_le_hid(self): |
| """ Use LE HID device to wake from suspend. And reconnects a secondary |
| LE HID device afterwards |
| """ |
| device = self.devices['BLE_MOUSE'][0] |
| device_reconnect = self.devices['BLE_KEYBOARD'][0] |
| |
| self.assert_discover_and_pair(device_reconnect) |
| self.test_device_set_discoverable(device_reconnect, False) |
| self.test_connection_by_adapter(device_reconnect.address) |
| self._test_keyboard_with_string(device_reconnect) |
| |
| self.run_peer_wakeup_device('BLE_MOUSE', |
| device, |
| device_test=self._test_mouse, |
| keep_paired=True) |
| |
| self.test_device_set_discoverable(device_reconnect, True) |
| self.test_device_is_connected(device_reconnect.address) |
| self._test_keyboard_with_string(device_reconnect) |
| |
| |
| @test_wrapper('Peer wakeup Classic HID', |
| devices={'MOUSE': 1}, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS) |
| def sr_peer_wake_classic_hid_stress(self): |
| """ Use classic HID device to wake from suspend. """ |
| device = self.devices['MOUSE'][0] |
| self.run_peer_wakeup_device('MOUSE', |
| device, |
| device_test=self._test_mouse, |
| iterations=STRESS_ITERATIONS) |
| |
| @test_wrapper('Peer wakeup LE HID', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=TABLET_MODELS + SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS) |
| def sr_peer_wake_le_hid_stress(self): |
| """ Use LE HID device to wake from suspend. """ |
| device = self.devices['BLE_MOUSE'][0] |
| self.run_peer_wakeup_device('BLE_MOUSE', |
| device, |
| device_test=self._test_mouse, |
| iterations=STRESS_ITERATIONS) |
| |
| @test_wrapper('Peer wakeup with A2DP should fail', |
| devices={'BLUETOOTH_AUDIO': 1}, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_peer_wake_a2dp_should_fail(self): |
| """ Use A2DP device to wake from suspend and fail. """ |
| device_type = 'BLUETOOTH_AUDIO' |
| device = self.devices[device_type][0] |
| self.initialize_bluetooth_audio(device, A2DP) |
| self.run_peer_wakeup_device( |
| device_type, |
| device, |
| device_test=self.test_device_a2dp_connected, |
| should_wake=False) |
| |
| # --------------------------------------------------------------- |
| # Suspend while discovering and advertising |
| # --------------------------------------------------------------- |
| |
| @test_wrapper('Suspend while discovering', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_while_discovering(self): |
| """ Suspend while discovering. """ |
| device = self.devices['BLE_MOUSE'][0] |
| boot_id = self.host.get_boot_id() |
| |
| self.test_device_set_discoverable(device, True) |
| |
| # Test discovery without setting discovery filter |
| # ---------------------------------------------------------------------- |
| suspend = self.suspend_async(suspend_time=EXPECT_NO_WAKE_SUSPEND_SEC) |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| # We don't pair to the peer device because we don't want it in the |
| # allowlist. However, we want an advertising peer in this test |
| # responding to the discovery requests. |
| self.test_start_discovery() |
| self.test_suspend_and_wait_for_sleep(suspend, |
| sleep_timeout=SUSPEND_SEC) |
| |
| # If discovery events wake us early, we will raise and suspend.exitcode |
| # will be non-zero |
| self.test_wait_for_resume(boot_id, |
| suspend, |
| resume_timeout=EXPECT_NO_WAKE_SUSPEND_SEC, |
| test_start_time=start_time) |
| |
| # Discovering should restore after suspend |
| self.test_is_discovering() |
| self.test_stop_discovery() |
| |
| # Test discovery with discovery filter set |
| # ---------------------------------------------------------------------- |
| suspend = self.suspend_async(suspend_time=EXPECT_NO_WAKE_SUSPEND_SEC) |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| # SetDiscoveryFilter was used for catching a regression around suspend |
| # on BlueZ. It won't be implemented on Floss. |
| if not self.floss: |
| self.test_set_discovery_filter({'Transport': 'auto'}) |
| self.test_start_discovery() |
| self.test_suspend_and_wait_for_sleep(suspend, |
| sleep_timeout=SUSPEND_SEC) |
| |
| # If discovery events wake us early, we will raise and suspend.exitcode |
| # will be non-zero |
| self.test_wait_for_resume(boot_id, |
| suspend, |
| resume_timeout=EXPECT_NO_WAKE_SUSPEND_SEC, |
| test_start_time=start_time) |
| |
| # Discovering should restore after suspend |
| self.test_is_discovering() |
| self.test_stop_discovery() |
| |
| @test_wrapper('Suspend while advertising', |
| devices={'MOUSE': 1}, |
| skip_models=SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS, |
| supports_floss=True) |
| def sr_while_advertising(self): |
| """ Suspend while advertising. """ |
| device = self.devices['MOUSE'][0] |
| boot_id = self.host.get_boot_id() |
| suspend = self.suspend_async(suspend_time=EXPECT_NO_WAKE_SUSPEND_SEC) |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| self.test_discoverable() |
| self.test_suspend_and_wait_for_sleep(suspend, |
| sleep_timeout=SUSPEND_SEC) |
| |
| # Peer device should not be able to discover us in suspend |
| self.test_discover_by_device_fails(device) |
| |
| self.test_wait_for_resume(boot_id, |
| suspend, |
| resume_timeout=EXPECT_NO_WAKE_SUSPEND_SEC, |
| test_start_time=start_time) |
| |
| # Test that we are properly discoverable again |
| self.test_is_discoverable() |
| self.test_discover_by_device(device) |
| |
| self.test_nondiscoverable() |
| |
| # --------------------------------------------------------------- |
| # Health checks |
| # --------------------------------------------------------------- |
| |
| @test_wrapper('Suspend while powered off', |
| devices={'MOUSE': 1}, |
| supports_floss=True) |
| def sr_while_powered_off(self): |
| """ Suspend while adapter is powered off. """ |
| device = self.devices['MOUSE'][0] |
| boot_id = self.host.get_boot_id() |
| suspend = self.suspend_async(suspend_time=SUSPEND_SEC) |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| # Pair device so we have something to do in suspend |
| self.assert_discover_and_pair(device) |
| |
| # Trigger power down and quickly suspend |
| self.test_power_off_adapter() |
| self.test_suspend_and_wait_for_sleep(suspend, |
| sleep_timeout=SUSPEND_SEC) |
| # Suspend and resume should succeed |
| self.test_wait_for_resume(boot_id, |
| suspend, |
| resume_timeout=SUSPEND_SEC, |
| test_start_time=start_time) |
| |
| # We should be able to power it back on |
| self.test_power_on_adapter() |
| |
| # Test that we can reconnect to the device after powering back on |
| self.test_connection_by_device(device) |
| |
| @test_wrapper('Suspend while receiving BQR test', |
| devices={ |
| 'BLUETOOTH_AUDIO': 1, |
| 'KEYBOARD': 1 |
| }, |
| flags=['Quick Health'], |
| skip_models=SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=QR_UNSUPPORTED_CHIPSETS + |
| SUSPEND_POWER_DOWN_CHIPSETS) |
| def sr_while_receiving_bqr(self): |
| """Suspend while receiving BQR to check the health.""" |
| audio_device = self.devices['BLUETOOTH_AUDIO'][0] |
| keyboard_device = self.devices['KEYBOARD'][0] |
| devices = (audio_device, keyboard_device) |
| test_profile = A2DP |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| # Enable BQR |
| self.qr_enable_quality_report() |
| self.dut_btmon_log_path = self.start_new_btmon() |
| |
| try: |
| # Connecting to all devices |
| for device in devices: |
| if device.device_type == 'BLUETOOTH_AUDIO': |
| self.initialize_bluetooth_audio(device, test_profile) |
| |
| self.test_discover_device(device.address) |
| self.test_pairing(device.address, device.pin, trusted=True) |
| self.test_connection_by_device(device) |
| time.sleep(2) |
| |
| # Connect to the audio device with A2DP profile. |
| self.test_device_a2dp_connected(audio_device) |
| |
| # Checking the bqr logs before suspend/resume. |
| time.sleep(QR_EVENT_PERIOD * 2) |
| self.test_send_log() |
| self.check_qr_event_log(num_devices=len(devices)) |
| |
| # Starting to suspend/resume. |
| boot_id = self.host.get_boot_id() |
| suspend = self.suspend_async(suspend_time=SUSPEND_SEC) |
| self.test_suspend_and_wait_for_sleep(suspend, |
| sleep_timeout=SUSPEND_SEC) |
| |
| self.test_wait_for_resume(boot_id, |
| suspend, |
| resume_timeout=SUSPEND_SEC, |
| test_start_time=start_time) |
| |
| # Open a 2nd btmon log for BQR log checking. |
| self.dut_btmon_log_path = self.start_new_btmon() |
| |
| # After suspend/resume DUT should be able to connect to devices. |
| # Expect the A2DP devices are connected after suspend/resume. |
| self.test_connection_by_device(keyboard_device) |
| |
| self.test_device_a2dp_connected(audio_device) |
| |
| # Checking the bqr logs after suspend/resume. |
| time.sleep(QR_EVENT_PERIOD * 2) |
| self.test_send_log() |
| self.check_qr_event_log(num_devices=len(devices)) |
| finally: |
| for device in devices: |
| if device.device_type == 'BLUETOOTH_AUDIO': |
| self.cleanup_bluetooth_audio(device, test_profile) |
| self.test_remove_pairing(device.address) |
| self.qr_disable_quality_report() |
| |
| def _sr_suspend_delay(self, devices, enable_BQR): |
| """Suspend with a delay to check the health. |
| |
| @param enable_BQR: to enable BQR for testing. |
| """ |
| boot_id = self.host.get_boot_id() |
| audio_test_profile = A2DP |
| |
| # Enable BQR |
| if enable_BQR: |
| self.qr_enable_quality_report() |
| |
| try: |
| # Connecting to all devices |
| for device in devices: |
| if device.device_type == 'BLUETOOTH_AUDIO': |
| self.initialize_bluetooth_audio(device, audio_test_profile) |
| |
| self.test_discover_device(device.address) |
| self.test_pairing(device.address, device.pin, trusted=True) |
| self.test_connection_by_device(device) |
| time.sleep(2) |
| |
| # Connect to the audio device with A2DP profile. |
| if device.device_type == 'BLUETOOTH_AUDIO': |
| self.test_device_a2dp_connected(device) |
| |
| # Start the suspend delay thread |
| suspend_delay_thread = self.suspend_delay_async( |
| SUSPEND_DELAY_FOR_QR_SECS, |
| SUSPEND_DELAY_TIMEOUT_FOR_QR_SECS, |
| WAKEUP_TIMEOUT_FOR_QR_SECS) |
| start_time = self.bluetooth_facade.get_device_utc_time() |
| |
| # Trigger suspend, wait for regular resume. Verify the system |
| # can suspend and resume correctly. |
| self.test_suspend_and_wait_for_sleep(suspend_delay_thread, |
| sleep_timeout=SUSPEND_SEC) |
| self.test_wait_for_resume(boot_id, |
| suspend_delay_thread, |
| resume_timeout=RESUME_TIMEOUT_FOR_QR_SECS, |
| test_start_time=start_time) |
| |
| finally: |
| # Disable BQR if it is enabled. |
| if enable_BQR: |
| self.qr_disable_quality_report() |
| |
| for device in devices: |
| if device.device_type == 'BLUETOOTH_AUDIO': |
| self.cleanup_bluetooth_audio(device, audio_test_profile) |
| self.test_remove_pairing(device.address) |
| |
| @test_wrapper('Suspend with a delay', |
| devices={ |
| 'BLUETOOTH_AUDIO': 1, |
| 'BLE_MOUSE': 1 |
| }, |
| flags=['Quick Health']) |
| def sr_suspend_delay(self): |
| """Suspend with a delay to check the health. |
| |
| In this test, the BQR feature is not enabled. Hence, this test is |
| expected to execute on all DUTs. |
| """ |
| audio_device = self.devices['BLUETOOTH_AUDIO'][0] |
| ble_mouse_device = self.devices['BLE_MOUSE'][0] |
| devices = (audio_device, ble_mouse_device) |
| self._sr_suspend_delay(devices, enable_BQR=False) |
| |
| @test_wrapper('Suspend with a delay while receiving BQR test', |
| devices={ |
| 'BLUETOOTH_AUDIO': 1, |
| 'KEYBOARD': 1 |
| }, |
| skip_models=SUSPEND_POWER_DOWN_MODELS, |
| skip_chipsets=QR_UNSUPPORTED_CHIPSETS + |
| SUSPEND_POWER_DOWN_CHIPSETS) |
| def sr_suspend_delay_while_receiving_bqr(self): |
| """Suspend with a delay while receiving BQR to check the health.""" |
| audio_device = self.devices['BLUETOOTH_AUDIO'][0] |
| keyboard_device = self.devices['KEYBOARD'][0] |
| devices = (audio_device, keyboard_device) |
| self._sr_suspend_delay(devices, enable_BQR=True) |
| |
| @batch_wrapper('SR with Peer Health') |
| def sr_health_batch_run(self, num_iterations=1, test_name=None): |
| """ Batch of suspend/resume peer health tests. """ |
| self.sr_reconnect_classic_hid() |
| self.sr_reconnect_le_hid() |
| self.sr_peer_wake_classic_hid() |
| self.sr_peer_wake_le_hid() |
| self.sr_while_discovering() |
| self.sr_while_advertising() |
| self.sr_while_receiving_bqr() |
| # Promote the sr_suspend_delay test later when proving to be stable. |
| # self.sr_suspend_delay() |
| self.sr_suspend_delay_while_receiving_bqr() |
| self.sr_reconnect_multiple_classic_hid() |
| self.sr_reconnect_multiple_le_hid() |
| self.sr_reconnect_multiple_classic_le_hid() |
| |
| def run_once(self, |
| host, |
| num_iterations=1, |
| args_dict=None, |
| test_name=None, |
| flag='Quick Health', |
| floss=False, |
| floss_lm_quirk=False): |
| """Running Bluetooth adapter suspend resume with peer autotest. |
| |
| @param host: the DUT, usually a chromebook |
| @param num_iterations: the number of times to execute the test |
| @param test_name: the test to run or None for all tests |
| @param flag: run tests with this flag (default: Quick Health) |
| |
| """ |
| |
| # Initialize and run the test batch or the requested specific test |
| self.quick_test_init(host, |
| use_btpeer=True, |
| flag=flag, |
| args_dict=args_dict, |
| floss=floss, |
| floss_lm_quirk=floss_lm_quirk) |
| self.sr_health_batch_run(num_iterations, test_name) |
| self.quick_test_cleanup() |
