• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Lint as: python2, python3
2# Copyright 2020 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6"""A Bluetooth adapter MTBF test of some Bluetooth use cases.
7
8   To add a new use case we just need to inherit from the existing test class
9   and then call the desired test methods in the batch method below. This allows
10   the test case to be used as both part of a MTBF batch and a normal batch.
11"""
12
13from __future__ import absolute_import
14from __future__ import division
15from __future__ import print_function
16
17import threading
18import time
19
20import common
21from autotest_lib.client.common_lib import error
22from autotest_lib.server.cros.bluetooth.bluetooth_adapter_audio_tests import (
23        BluetoothAdapterAudioTests)
24from autotest_lib.server.cros.bluetooth.bluetooth_adapter_better_together \
25        import BluetoothAdapterBetterTogether
26from autotest_lib.server.cros.bluetooth.bluetooth_adapter_hidreports_tests \
27        import BluetoothAdapterHIDReportTests
28from autotest_lib.server.cros.bluetooth.bluetooth_adapter_quick_tests import (
29        BluetoothAdapterQuickTests)
30from autotest_lib.server.cros.bluetooth.bluetooth_adapter_tests import (
31        TABLET_MODELS)
32from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import A2DP_LONG
33from six.moves import range
34
35# Iterations to run the mouse report test, this equals about 10 mins
36MOUSE_TEST_ITERATION = 15
37# Iterations to run the keyboard report test, this equals about 10 mins
38KEYBOARD_TEST_ITERATION = 60
39A2DP_TEST_DURATION_SEC = 600
40# Wait for some time before stating a new concurrent thread
41SLEEP_BETWEEN_THREADS = 15
42
43class bluetooth_AdapterMTBF(BluetoothAdapterBetterTogether,
44                            BluetoothAdapterHIDReportTests,
45                            BluetoothAdapterAudioTests):
46    """A Batch of Bluetooth adapter tests for MTBF. This test is written
47       as a batch of tests in order to reduce test time, since auto-test ramp up
48       time is costly. The batch is using BluetoothAdapterQuickTests wrapper
49       methods to start and end a test and a batch of tests.
50
51       This class can be called to run the entire test batch or to run a
52       specific test only
53    """
54
55    MTBF_TIMEOUT_MINS = 300
56    batch_wrapper = BluetoothAdapterQuickTests.quick_test_batch_decorator
57    mtbf_wrapper = BluetoothAdapterQuickTests.quick_test_mtbf_decorator
58    test_wrapper = BluetoothAdapterQuickTests.quick_test_test_decorator
59
60    @test_wrapper('MTBF Typical Use Cases',
61                  devices={'BLE_MOUSE': 1,
62                           'BLE_PHONE': 1,
63                           'BLUETOOTH_AUDIO': 1,
64                           'KEYBOARD': 1})
65    def typical_use_cases_test(self):
66        """Do some initialization work then start the typical MTBF test loop"""
67
68        # TODO(b/165606673) - Remove blooglet once the bug is fixed
69        self.skip_wake_test = self.host.get_model_from_cros_config() in \
70                              TABLET_MODELS + ['blooglet']
71        mouse = self.devices['BLE_MOUSE'][0]
72        phone = self.devices['BLE_PHONE'][0]
73        audio = self.devices['BLUETOOTH_AUDIO'][0]
74        keyboard = self.devices['KEYBOARD'][0]
75
76        self.test_device_pairing(mouse)
77        self.test_device_pairing(keyboard)
78
79        self.run_typical_use_cases(mouse, phone, audio, keyboard)
80
81
82    @mtbf_wrapper(timeout_mins=MTBF_TIMEOUT_MINS, test_name='typical_use_cases')
83    def run_typical_use_cases(self, mouse, phone, audio, keyboard):
84        """Run typical MTBF test scenarios:
85           1. Run the better together test
86           2. Run the concurrent mouse and A2DP tests for 10 minutes
87           3. Suspend/Resume
88           4. Run the concurrent mouse amd keyboard tests for 10 minutes
89           5. Suspend and wake up the DUT by mouse
90           6. Run the concurrent mouse, keyboard and A2DP tests for 10 minutes
91        """
92        # Run the better together test on the phone
93        self.test_better_together(phone)
94
95        # Restore the discovery filter since better together test changed it
96        self.test_set_discovery_filter({'Transport':'auto'})
97
98        self.test_mouse_and_audio(mouse, audio)
99        self.test_suspend_resume(mouse, keyboard)
100        self.test_mouse_and_keyboard(mouse, keyboard)
101        # Mouse wakeup test
102        self.test_suspend_and_mouse_wakeup(mouse, keyboard)
103        self.test_hid_and_audio(mouse, keyboard, audio)
104
105
106    def test_mouse_and_audio(self, mouse, audio):
107        """Run the mouse and audio tests concurrently for 10 mins"""
108        audio_thread = threading.Thread(
109            target=self.test_audio, args=(audio, A2DP_TEST_DURATION_SEC))
110        mouse_thread = threading.Thread(
111            target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION))
112
113        audio_thread.start()
114        time.sleep(SLEEP_BETWEEN_THREADS)
115        mouse_thread.start()
116        time.sleep(SLEEP_BETWEEN_THREADS)
117        audio_thread.join()
118        mouse_thread.join()
119        if self.fails:
120            raise error.TestFail(self.fails)
121
122
123    def test_mouse_and_keyboard(self, mouse, keyboard):
124        """Run the mouse and keyboard tests concurrently for 10 mins"""
125        mouse_thread = threading.Thread(
126            target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION))
127        keyboard_thread = \
128            threading.Thread(target=self.test_keyboard,
129                             args=(keyboard, KEYBOARD_TEST_ITERATION))
130        time.sleep(SLEEP_BETWEEN_THREADS)
131        mouse_thread.start()
132        time.sleep(SLEEP_BETWEEN_THREADS)
133        keyboard_thread.start()
134        mouse_thread.join()
135        keyboard_thread.join()
136        if self.fails:
137            raise error.TestFail(self.fails)
138
139
140    def test_hid_and_audio(self, mouse, keyboard, audio):
141        """Run the audio, mouse and keyboard tests concurrently for 10 mins"""
142        audio_thread = threading.Thread(
143            target=self.test_audio, args=(audio, A2DP_TEST_DURATION_SEC))
144        mouse_thread = threading.Thread(
145            target=self.test_mouse, args=(mouse, MOUSE_TEST_ITERATION))
146        keyboard_thread = \
147            threading.Thread(target=self.test_keyboard,
148                             args=(keyboard, KEYBOARD_TEST_ITERATION))
149        audio_thread.start()
150        time.sleep(SLEEP_BETWEEN_THREADS)
151        mouse_thread.start()
152        time.sleep(SLEEP_BETWEEN_THREADS)
153        keyboard_thread.start()
154        audio_thread.join()
155        mouse_thread.join()
156        keyboard_thread.join()
157        if self.fails:
158            raise error.TestFail(self.fails)
159
160
161    def test_mouse(self, mouse, iteration):
162        """Run mouse report test for certain iterations"""
163        for i in range(iteration):
164            self.run_mouse_tests(device=mouse)
165
166
167    def test_keyboard(self, keyboard, iteration):
168        """Run keyboard report test for certain iterations"""
169        for i in range(iteration):
170            self.run_keyboard_tests(device=keyboard)
171
172
173    def test_audio(self, device, duration):
174        """Test A2DP
175
176           This test plays A2DP audio on the DUT and record on the peer device,
177           then verify the legitimacy of the frames recorded.
178
179        """
180        self.bluetooth_facade.remove_device_object(device.address)
181        device.RemoveDevice(self.bluetooth_facade.address)
182
183        self.initialize_bluetooth_audio(device, A2DP_LONG)
184        self.test_device_set_discoverable(device, True)
185        self.test_discover_device(device.address)
186        self.test_pairing(device.address, device.pin, trusted=True)
187        device.SetTrustedByRemoteAddress(self.bluetooth_facade.address)
188        self.test_connection_by_adapter(device.address)
189        self.test_a2dp_sinewaves(device, A2DP_LONG, duration)
190        self.test_disconnection_by_adapter(device.address)
191        self.cleanup_bluetooth_audio(device, A2DP_LONG)
192        self.test_remove_device_object(device.address)
193
194
195    def test_device_pairing(self, device):
196        """Test device pairing"""
197
198        # Remove the pairing first
199        self.bluetooth_facade.remove_device_object(device.address)
200        device.RemoveDevice(self.bluetooth_facade.address)
201
202        self.test_device_set_discoverable(device, True)
203        self.test_discover_device(device.address)
204        time.sleep(self.TEST_SLEEP_SECS)
205        self.test_pairing(device.address, device.pin, trusted=True)
206        time.sleep(self.TEST_SLEEP_SECS)
207        self.test_connection_by_adapter(device.address)
208
209
210    def test_suspend_resume(self, mouse, keyboard):
211        """Test the device can connect after suspending and resuming"""
212        boot_id = self.host.get_boot_id()
213        suspend = self.suspend_async(suspend_time=15)
214        start_time = self.bluetooth_facade.get_device_utc_time()
215
216        self.test_device_set_discoverable(mouse, False)
217
218        self.test_suspend_and_wait_for_sleep(
219            suspend, sleep_timeout=15)
220        self.test_wait_for_resume(boot_id,
221                                  suspend,
222                                  resume_timeout=15,
223                                  test_start_time=start_time)
224
225        # LE can't reconnect without advertising/discoverable
226        self.test_device_set_discoverable(mouse, True)
227        self.test_device_is_connected(mouse.address)
228        self.test_hid_device_created(mouse.address)
229
230        self.test_connection_by_device(keyboard)
231        self.test_hid_device_created(keyboard.address)
232
233
234    def test_suspend_and_mouse_wakeup(self, mouse, keyboard):
235        """Test the device can be waken up by the mouse"""
236        if self.skip_wake_test:
237            return
238
239        self.run_peer_wakeup_device('MOUSE', mouse, should_pair=False)
240        # Make sure we're actually connected
241        self.test_device_is_connected(mouse.address)
242        self.test_hid_device_created(mouse.address)
243
244        self.test_connection_by_device(keyboard)
245        self.test_hid_device_created(keyboard.address)
246
247
248    @test_wrapper('MTBF Better Together Stress', devices={'BLE_PHONE': 1})
249    def better_together_stress_test(self):
250        """Run better together stress test"""
251
252        phone = self.devices['BLE_PHONE'][0]
253        phone.RemoveDevice(self.bluetooth_facade.address)
254        self.run_better_together_stress(address=phone.address)
255
256
257    def test_better_together(self, phone):
258        """Test better together"""
259        # Clean up the environment
260        self.bluetooth_facade.disconnect_device(phone.address)
261        self.bluetooth_facade.remove_device_object(phone.address)
262        phone.RemoveDevice(self.bluetooth_facade.address)
263        self.test_smart_unlock(address=phone.address)
264
265
266    @mtbf_wrapper(
267        timeout_mins=MTBF_TIMEOUT_MINS, test_name='better_together_stress')
268    def run_better_together_stress(self, address):
269        """Run better together stress test"""
270
271        self.test_smart_unlock(address)
272
273
274    @batch_wrapper('Adapter MTBF')
275    def mtbf_batch_run(self, num_iterations=1, test_name=None):
276        """Run the Bluetooth MTBF test batch or a specific
277           given test. The wrapper of this method is implemented in
278           batch_decorator. Using the decorator a test batch method can
279           implement the only its core tests invocations and let the
280           decorator handle the wrapper, which is taking care for whether to
281           run a specific test or the batch as a whole, and running the batch
282           in iterations
283
284           @param num_iterations: how many iterations to run
285           @param test_name: specific test to run otherwise None to run the
286                             whole batch
287        """
288        # TODO: finalize the test cases that need to be run as MTBF
289        self.typical_use_cases_test()
290
291
292    def run_once(self,
293                 host,
294                 num_iterations=1,
295                 test_name=None,
296                 args_dict=None):
297        """Run the batch of Bluetooth MTBF tests
298
299        @param host: the DUT, usually a chromebook
300        @param num_iterations: the number of rounds to execute the test
301        @test_name: the test to run, or None for all tests
302        """
303
304        # Initialize and run the test batch or the requested specific test
305        self.set_fail_fast(args_dict, True)
306        self.quick_test_init(host, use_btpeer=True, args_dict=args_dict)
307
308        self.mtbf_batch_run(num_iterations, test_name)
309
310        self.quick_test_cleanup()
311