# 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 six.moves import range test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator PROFILE_CONNECT_WAIT = 15 SUSPEND_SEC = 15 EXPECT_NO_WAKE_SUSPEND_SEC = 30 EXPECT_PEER_WAKE_SUSPEND_SEC = 60 STRESS_ITERATIONS = 25 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_left_click(self, device): self.test_hid_device_created(device.address) return self.test_mouse_left_click(device) # --------------------------------------------------------------- # 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_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) for device_type, device, device_test in devtuples: # Only reconnect if we don't expect automatic reconnect if not auto_reconnect: 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) 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_left_click)]) @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_left_click)]) # 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_left_click), ('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_left_click), ('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_left_click), ('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_left_click)], 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_left_click)], 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) # --------------------------------------------------------------- # Wake from suspend tests # --------------------------------------------------------------- def run_peer_wakeup_device(self, device_type, device, device_test=None, iterations=1, should_wake=True): """ Uses paired peer device to wake the device from suspend. @param device_type: the device type (used to determine if it's LE) @param device: the meta device with the paired device @param device_test: What to test to run after waking and connecting the adapter/host @param iterations: Number of suspend + peer wake loops to run @param should_wake: Whether wakeup should occur on this test. With HID peers, this should be True. With non-HID peers, this should be false. """ boot_id = self.host.get_boot_id() if should_wake: sleep_time = EXPECT_PEER_WAKE_SUSPEND_SEC resume_time = SUSPEND_SEC resume_slack = 5 # Allow 5s slack for resume timeout else: sleep_time = EXPECT_NO_WAKE_SUSPEND_SEC resume_time = EXPECT_NO_WAKE_SUSPEND_SEC # Negative resume slack lets us wake a bit earlier than expected # If suspend takes a while to enter, this may be necessary to get # the timings right. resume_slack = -5 # Clear wake before testing self.test_adapter_set_wake_disabled() try: self.assert_discover_and_pair(device) self.assert_on_fail( self.test_device_set_discoverable(device, False)) # Confirm connection completed self.assert_on_fail(self.test_device_is_connected(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 peer wake'.format( it + 1, iterations)) # Start a new suspend instance suspend = self.suspend_async(suspend_time=sleep_time, expect_bt_wake=should_wake) start_time = self.bluetooth_facade.get_device_time() if should_wake: self.test_device_wake_allowed(device.address) # Also wait until powerd marks adapter as wake enabled self.test_adapter_wake_enabled() else: self.test_device_wake_not_allowed(device.address) # Trigger suspend, asynchronously wake and wait for resume self.test_suspend_and_wait_for_sleep(suspend, sleep_timeout=5) # Trigger peer wakeup adapter_address = self.bluetooth_facade.address peer_wake = self.device_connect_async(device_type, device, adapter_address, delay_wake=5, should_wake=should_wake) peer_wake.start() # Expect a quick resume. If a timeout occurs, test fails. Since # we delay sending the wake signal, we should accommodate that # in our expected timeout. self.test_wait_for_resume(boot_id, suspend, resume_timeout=resume_time, test_start_time=start_time, resume_slack=resume_slack, fail_on_timeout=should_wake, fail_early_wake=not should_wake) # Finish peer wake process peer_wake.join() # Only check peer device connection state if we expected to wake # from it. Otherwise, we may or may not be connected based on # the specific profile's reconnection policy. if should_wake: # Make sure we're actually connected self.test_device_is_connected(device.address) # Verify the profile is working if device_test is not None: device_test(device) finally: self.test_remove_pairing(device.address) # 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_left_click) # 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_left_click) # 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_left_click, 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_left_click, 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_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_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_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_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()