| # Lint as: python3 |
| """Tests for blueberry.tests.bluetooth.bluetooth_throughput.""" |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import logging |
| import math |
| |
| from mobly import asserts |
| from mobly import test_runner |
| from mobly.controllers.android_device_lib.jsonrpc_client_base import ApiError |
| from mobly.signals import TestAbortClass |
| # Internal import |
| from blueberry.utils import blueberry_base_test |
| from blueberry.utils import metrics_utils |
| # Internal import |
| |
| |
| class BluetoothThroughputTest(blueberry_base_test.BlueberryBaseTest): |
| |
| @retry.logged_retry_on_exception( |
| retry_intervals=retry.FuzzedExponentialIntervals( |
| initial_delay_sec=2, factor=5, num_retries=5, max_delay_sec=300)) |
| def _measure_throughput(self, num_of_buffers, buffer_size): |
| """Measures the throughput of a data transfer. |
| |
| Sends data from the client device that is read by the server device. |
| Calculates the throughput for the transfer. |
| |
| Args: |
| num_of_buffers: An integer value designating the number of buffers |
| to be sent. |
| buffer_size: An integer value designating the size of each buffer, |
| in bytes. |
| |
| Returns: |
| The throughput of the transfer in bytes per second. |
| """ |
| |
| # TODO(user): Need to fix throughput send/receive methods |
| (self.phone.sl4a |
| .bluetoothConnectionThroughputSend(num_of_buffers, buffer_size)) |
| |
| throughput = (self.derived_bt_device.sl4a |
| .bluetoothConnectionThroughputRead(num_of_buffers, |
| buffer_size)) |
| return throughput |
| |
| def _throughput_test(self, buffer_size, test_name): |
| logging.info('throughput test with buffer_size: %d and testname: %s', |
| buffer_size, test_name) |
| metrics = {} |
| throughput_list = [] |
| num_of_buffers = 1 |
| for _ in range(self.iterations): |
| throughput = self._measure_throughput(num_of_buffers, buffer_size) |
| logging.info('Throughput: %d bytes-per-sec', throughput) |
| throughput_list.append(throughput) |
| |
| metrics['data_transfer_protocol'] = self.data_transfer_type |
| metrics['data_packet_size'] = buffer_size |
| metrics['data_throughput_min_bytes_per_second'] = int( |
| min(throughput_list)) |
| metrics['data_throughput_max_bytes_per_second'] = int( |
| max(throughput_list)) |
| metrics['data_throughput_avg_bytes_per_second'] = int( |
| math.fsum(throughput_list) / float(len(throughput_list))) |
| |
| logging.info('Throughput at large buffer: %s', metrics) |
| |
| asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0, |
| 'Minimum throughput must be greater than 0!') |
| |
| self.metrics.add_test_metrics(metrics) |
| for metric in metrics: |
| self.record_data({ |
| 'Test Name': test_name, |
| 'sponge_properties': { |
| metric: metrics[metric], |
| } |
| }) |
| self.record_data({ |
| 'Test Name': test_name, |
| 'sponge_properties': { |
| 'proto_ascii': |
| self.metrics.proto_message_to_ascii(), |
| 'primary_device_build': |
| self.phone.get_device_info()['android_release_id'] |
| } |
| }) |
| |
| def setup_class(self): |
| """Standard Mobly setup class.""" |
| super(BluetoothThroughputTest, self).setup_class() |
| if len(self.android_devices) < 2: |
| raise TestAbortClass( |
| 'Not enough android phones detected (need at least two)') |
| self.phone = self.android_devices[0] |
| |
| # We treat the secondary phone as a derived_bt_device in order for the |
| # generic script to work with this android phone properly. Data will be sent |
| # from first phone to the second phone. |
| self.derived_bt_device = self.android_devices[1] |
| self.phone.init_setup() |
| self.derived_bt_device.init_setup() |
| self.phone.sl4a_setup() |
| self.derived_bt_device.sl4a_setup() |
| self.set_btsnooplogmode_full(self.phone) |
| self.set_btsnooplogmode_full(self.derived_bt_device) |
| |
| self.metrics = ( |
| metrics_utils.BluetoothMetricLogger( |
| metrics_pb2.BluetoothDataTestResult())) |
| self.metrics.add_primary_device_metrics(self.phone) |
| self.metrics.add_connected_device_metrics(self.derived_bt_device) |
| |
| self.data_transfer_type = metrics_pb2.BluetoothDataTestResult.RFCOMM |
| self.iterations = int(self.user_params.get('iterations', 300)) |
| logging.info('Running Bluetooth throughput test %s times.', self.iterations) |
| logging.info('Successfully found required devices.') |
| |
| def setup_test(self): |
| """Setup for bluetooth latency test.""" |
| logging.info('Setup Test for test_bluetooth_throughput') |
| super(BluetoothThroughputTest, self).setup_test() |
| asserts.assert_true(self.phone.connect_with_rfcomm(self.derived_bt_device), |
| 'Failed to establish RFCOMM connection') |
| |
| def test_bluetooth_throughput_large_buffer(self): |
| """Tests the throughput with large buffer size. |
| |
| Tests the throughput over a series of data transfers with large buffer size. |
| """ |
| large_buffer_size = 300 |
| test_name = 'test_bluetooth_throughput_large_buffer' |
| self._throughput_test(large_buffer_size, test_name) |
| |
| def test_bluetooth_throughput_medium_buffer(self): |
| """Tests the throughput with medium buffer size. |
| |
| Tests the throughput over a series of data transfers with medium buffer |
| size. |
| """ |
| medium_buffer_size = 100 |
| test_name = 'test_bluetooth_throughput_medium_buffer' |
| self._throughput_test(medium_buffer_size, test_name) |
| |
| def test_bluetooth_throughput_small_buffer(self): |
| """Tests the throughput with small buffer size. |
| |
| Tests the throughput over a series of data transfers with small buffer size. |
| """ |
| small_buffer_size = 10 |
| test_name = 'test_bluetooth_throughput_small_buffer' |
| self._throughput_test(small_buffer_size, test_name) |
| |
| def test_maximum_buffer_size(self): |
| """Calculates the maximum allowed buffer size for one packet.""" |
| current_buffer_size = 300 |
| throughput = -1 |
| num_of_buffers = 1 |
| while True: |
| logging.info('Trying buffer size %d', current_buffer_size) |
| try: |
| throughput = self._measure_throughput( |
| num_of_buffers, current_buffer_size) |
| logging.info('The throughput is %d at buffer size of %d', throughput, |
| current_buffer_size) |
| except ApiError: |
| maximum_buffer_size = current_buffer_size - 1 |
| logging.info('Max buffer size: %d bytes', maximum_buffer_size) |
| logging.info('Max throughput: %d bytes-per-second', throughput) |
| self.record_data({ |
| 'Test Name': 'test_maximum_buffer_size', |
| 'sponge_properties': { |
| 'maximum_buffer_size': maximum_buffer_size |
| } |
| }) |
| return True |
| current_buffer_size += 1 |
| |
| def teardown_test(self): |
| self.phone.sl4a.bluetoothSocketConnStop() |
| self.derived_bt_device.sl4a.bluetoothSocketConnStop() |
| |
| def teardown_class(self): |
| self.phone.factory_reset_bluetooth() |
| self.derived_bt_device.factory_reset_bluetooth() |
| logging.info('Factory resetting Bluetooth on devices.') |
| super(BluetoothThroughputTest, self).teardown_class() |
| |
| |
| if __name__ == '__main__': |
| test_runner.main() |