1#/usr/bin/env python3
2#
3# Copyright (C) 2018 The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may not
6# use this file except in compliance with the License. You may obtain a copy of
7# the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations under
15# the License.
16
17import statistics
18from acts import asserts
19from acts.base_test import BaseTestClass
20from acts.signals import TestPass
21from acts.test_decorators import test_tracker_info
22from acts.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
23from acts.test_utils.bt.bt_test_utils import orchestrate_rfcomm_connection
24from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
25from acts.test_utils.bt.bt_test_utils import verify_server_and_client_connected
26from acts.test_utils.bt.loggers.bluetooth_metric_logger import BluetoothMetricLogger
27from acts.test_utils.bt.loggers.protos import bluetooth_metric_pb2 as proto_module
28from acts.utils import set_location_service
29
30
31class BluetoothThroughputTest(BaseTestClass):
32    """Connects two Android phones and tests the throughput between them.
33
34    Attributes:
35         client_device: An Android device object that will be sending data
36         server_device: An Android device object that will be receiving data
37         bt_logger: The proxy logger instance for each test case
38         data_transfer_type: Data transfer protocol used for the test
39    """
40
41    def setup_class(self):
42        super().setup_class()
43
44        # Sanity check of the devices under test
45        # TODO(b/119051823): Investigate using a config validator to replace this.
46        if len(self.android_devices) < 2:
47            raise ValueError(
48                'Not enough android phones detected (need at least two)')
49
50        # Data will be sent from the client_device to the server_device
51        self.client_device = self.android_devices[0]
52        self.server_device = self.android_devices[1]
53        self.bt_logger = BluetoothMetricLogger.for_test_case()
54        self.data_transfer_type = proto_module.BluetoothDataTestResult.RFCOMM
55        self.log.info('Successfully found required devices.')
56
57    def setup_test(self):
58        setup_multiple_devices_for_bt_test(self.android_devices)
59        self._connect_rfcomm()
60
61    def teardown_test(self):
62        if verify_server_and_client_connected(
63                self.client_device, self.server_device, log=False):
64            self.client_device.droid.bluetoothSocketConnStop()
65            self.server_device.droid.bluetoothSocketConnStop()
66
67    def _connect_rfcomm(self):
68        """Establishes an RFCOMM connection between two phones.
69
70        Connects the client device to the server device given the hardware
71        address of the server device.
72        """
73
74        set_location_service(self.client_device, True)
75        set_location_service(self.server_device, True)
76        server_address = self.server_device.droid.bluetoothGetLocalAddress()
77        self.log.info('Pairing and connecting devices')
78        asserts.assert_true(self.client_device.droid
79                            .bluetoothDiscoverAndBond(server_address),
80                            'Failed to pair and connect devices')
81
82        # Create RFCOMM connection
83        asserts.assert_true(orchestrate_rfcomm_connection
84                            (self.client_device, self.server_device),
85                            'Failed to establish RFCOMM connection')
86
87    def _measure_throughput(self, num_of_buffers, buffer_size):
88        """Measures the throughput of a data transfer.
89
90        Sends data from the client device that is read by the server device.
91        Calculates the throughput for the transfer.
92
93        Args:
94            num_of_buffers: An integer value designating the number of buffers
95                          to be sent.
96            buffer_size: An integer value designating the size of each buffer,
97                       in bytes.
98
99        Returns:
100            The throughput of the transfer in bytes per second.
101        """
102
103        # TODO(b/119638242): Need to fix throughput send/receive methods
104        (self.client_device.droid
105         .bluetoothConnectionThroughputSend(num_of_buffers, buffer_size))
106
107        throughput = (self.server_device.droid
108                      .bluetoothConnectionThroughputRead(num_of_buffers,
109                                                         buffer_size))
110        return throughput
111
112    @BluetoothBaseTest.bt_test_wrap
113    @test_tracker_info(uuid='23afba5b-5801-42c2-8d7a-41510e91a605')
114    def test_bluetooth_throughput_large_buffer(self):
115        """Tests the throughput over a series of data transfers with large
116        buffer size.
117        """
118
119        metrics = {}
120        throughput_list = []
121
122        for transfer in range(300):
123            throughput = self._measure_throughput(1, 300)
124            self.log.info('Throughput: {} bytes-per-sec'.format(throughput))
125            throughput_list.append(throughput)
126
127        metrics['data_transfer_protocol'] = self.data_transfer_type
128        metrics['data_packet_size'] = 300
129        metrics['data_throughput_min_bytes_per_second'] = int(
130            min(throughput_list))
131        metrics['data_throughput_max_bytes_per_second'] = int(
132            max(throughput_list))
133        metrics['data_throughput_avg_bytes_per_second'] = int(statistics.mean(
134            throughput_list))
135
136        proto = self.bt_logger.get_results(metrics,
137                                           self.__class__.__name__,
138                                           self.server_device,
139                                           self.client_device)
140
141        asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0,
142                            'Minimum throughput must be greater than 0!',
143                            extras=proto)
144        raise TestPass('Throughput test (large buffer) completed successfully',
145                       extras=proto)
146
147    @BluetoothBaseTest.bt_test_wrap
148    @test_tracker_info(uuid='5472fe33-891e-4fa1-ba84-78fc6f6a2327')
149    def test_bluetooth_throughput_medium_buffer(self):
150        """Tests the throughput over a series of data transfers with medium
151        buffer size.
152        """
153
154        metrics = {}
155        throughput_list = []
156
157        for transfer in range(300):
158            throughput = self._measure_throughput(1, 100)
159            self.log.info('Throughput: {} bytes-per-sec'.format(throughput))
160            throughput_list.append(throughput)
161
162        metrics['data_transfer_protocol'] = self.data_transfer_type
163        metrics['data_packet_size'] = 100
164        metrics['data_throughput_min_bytes_per_second'] = int(
165            min(throughput_list))
166        metrics['data_throughput_max_bytes_per_second'] = int(
167            max(throughput_list))
168        metrics['data_throughput_avg_bytes_per_second'] = int(statistics.mean(
169            throughput_list))
170
171        proto = self.bt_logger.get_results(metrics,
172                                           self.__class__.__name__,
173                                           self.server_device,
174                                           self.client_device)
175
176        asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0,
177                            'Minimum throughput must be greater than 0!',
178                            extras=proto)
179        raise TestPass('Throughput test (medium buffer) completed successfully',
180                       extras=proto)
181
182    @BluetoothBaseTest.bt_test_wrap
183    @test_tracker_info(uuid='97589280-cefa-4ae4-b3fd-94ec9c1f4104')
184    def test_bluetooth_throughput_small_buffer(self):
185        """Tests the throughput over a series of data transfers with small
186        buffer size.
187        """
188
189        metrics = {}
190        throughput_list = []
191
192        for transfer in range(300):
193            throughput = self._measure_throughput(1, 10)
194            self.log.info('Throughput: {} bytes-per-sec'.format(throughput))
195            throughput_list.append(throughput)
196
197        metrics['data_transfer_protocol'] = self.data_transfer_type
198        metrics['data_packet_size'] = 10
199        metrics['data_throughput_min_bytes_per_second'] = int(
200            min(throughput_list))
201        metrics['data_throughput_max_bytes_per_second'] = int(
202            max(throughput_list))
203        metrics['data_throughput_avg_bytes_per_second'] = int(statistics.mean(
204            throughput_list))
205
206        proto = self.bt_logger.get_results(metrics,
207                                           self.__class__.__name__,
208                                           self.server_device,
209                                           self.client_device)
210
211        asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0,
212                            'Minimum throughput must be greater than 0!',
213                            extras=proto)
214        raise TestPass('Throughput test (small buffer) completed successfully',
215                       extras=proto)
216
217    @BluetoothBaseTest.bt_test_wrap
218    def test_maximum_buffer_size(self):
219        """Calculates the maximum allowed buffer size for one packet."""
220
221        current_buffer_size = 1
222        while True:
223            self.log.info('Trying buffer size {}'.format(current_buffer_size))
224            try:
225                throughput = self._measure_throughput(1, current_buffer_size)
226            except Exception:
227                buffer_msg = ('Max buffer size: {} bytes'.
228                              format(current_buffer_size - 1))
229                throughput_msg = ('Max throughput: {} bytes-per-second'.
230                                  format(throughput))
231                self.log.info(buffer_msg)
232                self.log.info(throughput_msg)
233                return True
234            current_buffer_size += 1
235