| # Lint as: python2, python3 |
| # 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 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) |
| from autotest_lib.server.cros.bluetooth.bluetooth_adapter_audio_tests import ( |
| BluetoothAdapterAudioTests) |
| 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 |
| |
| |
| class bluetooth_AdapterSRHealth(BluetoothAdapterQuickTests, |
| BluetoothAdapterAudioTests): |
| """Server side bluetooth adapter suspend resume test with peer.""" |
| |
| def _test_keyboard_with_string(self, device): |
| self.test_hid_device_created(device.address) |
| return self.test_keyboard_input_from_trace(device, "simple_text") |
| |
| def _test_mouse(self, device): |
| self.test_hid_device_created(device.address) |
| return 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) |
| 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}) |
| 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('Reconnect LE HID', devices={'BLE_MOUSE': 1}) |
| 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']) |
| 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 |
| }) |
| 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 |
| }) |
| 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) |
| 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) |
| |
| |
| # TODO(b/151332866) - Bob can't wake from suspend due to wrong power/wakeup |
| # TODO(b/150897528) - Dru is powered down during suspend, won't wake up |
| @test_wrapper('Peer wakeup Classic HID', |
| devices={'MOUSE': 1}, |
| skip_models=TABLET_MODELS + ['bob', 'dru'], |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS) |
| 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) |
| |
| # TODO(b/151332866) - Bob can't wake from suspend due to wrong power/wakeup |
| # TODO(b/150897528) - Dru is powered down during suspend, won't wake up |
| @test_wrapper('Peer wakeup LE HID', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=TABLET_MODELS + ['bob', 'dru'], |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS) |
| 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) |
| |
| # TODO(b/151332866) - Bob can't wake from suspend due to wrong power/wakeup |
| # TODO(b/150897528) - Dru is powered down during suspend, won't wake up |
| @test_wrapper('Peer wakeup Classic HID', |
| devices={'MOUSE': 1}, |
| skip_models=TABLET_MODELS + ['bob', 'dru'], |
| 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) |
| |
| # TODO(b/151332866) - Bob can't wake from suspend due to wrong power/wakeup |
| # TODO(b/150897528) - Dru is powered down during suspend, won't wake up |
| @test_wrapper('Peer wakeup LE HID', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=TABLET_MODELS + ['bob', 'dru'], |
| 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) |
| 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 |
| # --------------------------------------------------------------- |
| |
| # TODO(b/150897528) - Scarlet Dru loses firmware around suspend |
| @test_wrapper('Suspend while discovering', |
| devices={'BLE_MOUSE': 1}, |
| skip_models=['dru'], |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS) |
| 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() |
| |
| 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() |
| |
| # TODO(b/150897528) - Scarlet Dru loses firmware around suspend |
| @test_wrapper('Suspend while advertising', |
| devices={'MOUSE': 1}, |
| skip_models=['dru'], |
| skip_chipsets=SUSPEND_POWER_DOWN_CHIPSETS) |
| 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}) |
| 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) |
| |
| @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_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'): |
| """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) |
| self.sr_health_batch_run(num_iterations, test_name) |
| self.quick_test_cleanup() |
