235 lines
9.9 KiB
Python
235 lines
9.9 KiB
Python
#/usr/bin/env python3
|
|
#
|
|
# Copyright (C) 2018 The Android Open Source Project
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
|
|
# use this file except in compliance with the License. You may obtain a copy of
|
|
# the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations under
|
|
# the License.
|
|
|
|
import statistics
|
|
from acts import asserts
|
|
from acts.base_test import BaseTestClass
|
|
from acts.signals import TestPass
|
|
from acts.test_decorators import test_tracker_info
|
|
from acts_contrib.test_utils.bt.BluetoothBaseTest import BluetoothBaseTest
|
|
from acts_contrib.test_utils.bt.bt_test_utils import orchestrate_rfcomm_connection
|
|
from acts_contrib.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
|
|
from acts_contrib.test_utils.bt.bt_test_utils import verify_server_and_client_connected
|
|
from acts_contrib.test_utils.bt.loggers.bluetooth_metric_logger import BluetoothMetricLogger
|
|
from acts_contrib.test_utils.bt.loggers.protos import bluetooth_metric_pb2 as proto_module
|
|
from acts.utils import set_location_service
|
|
|
|
|
|
class BluetoothThroughputTest(BaseTestClass):
|
|
"""Connects two Android phones and tests the throughput between them.
|
|
|
|
Attributes:
|
|
client_device: An Android device object that will be sending data
|
|
server_device: An Android device object that will be receiving data
|
|
bt_logger: The proxy logger instance for each test case
|
|
data_transfer_type: Data transfer protocol used for the test
|
|
"""
|
|
|
|
def setup_class(self):
|
|
super().setup_class()
|
|
|
|
# Sanity check of the devices under test
|
|
# TODO(b/119051823): Investigate using a config validator to replace this.
|
|
if len(self.android_devices) < 2:
|
|
raise ValueError(
|
|
'Not enough android phones detected (need at least two)')
|
|
|
|
# Data will be sent from the client_device to the server_device
|
|
self.client_device = self.android_devices[0]
|
|
self.server_device = self.android_devices[1]
|
|
self.bt_logger = BluetoothMetricLogger.for_test_case()
|
|
self.data_transfer_type = proto_module.BluetoothDataTestResult.RFCOMM
|
|
self.log.info('Successfully found required devices.')
|
|
|
|
def setup_test(self):
|
|
setup_multiple_devices_for_bt_test(self.android_devices)
|
|
self._connect_rfcomm()
|
|
|
|
def teardown_test(self):
|
|
if verify_server_and_client_connected(
|
|
self.client_device, self.server_device, log=False):
|
|
self.client_device.droid.bluetoothSocketConnStop()
|
|
self.server_device.droid.bluetoothSocketConnStop()
|
|
|
|
def _connect_rfcomm(self):
|
|
"""Establishes an RFCOMM connection between two phones.
|
|
|
|
Connects the client device to the server device given the hardware
|
|
address of the server device.
|
|
"""
|
|
|
|
set_location_service(self.client_device, True)
|
|
set_location_service(self.server_device, True)
|
|
server_address = self.server_device.droid.bluetoothGetLocalAddress()
|
|
self.log.info('Pairing and connecting devices')
|
|
asserts.assert_true(self.client_device.droid
|
|
.bluetoothDiscoverAndBond(server_address),
|
|
'Failed to pair and connect devices')
|
|
|
|
# Create RFCOMM connection
|
|
asserts.assert_true(orchestrate_rfcomm_connection
|
|
(self.client_device, self.server_device),
|
|
'Failed to establish RFCOMM connection')
|
|
|
|
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(b/119638242): Need to fix throughput send/receive methods
|
|
(self.client_device.droid
|
|
.bluetoothConnectionThroughputSend(num_of_buffers, buffer_size))
|
|
|
|
throughput = (self.server_device.droid
|
|
.bluetoothConnectionThroughputRead(num_of_buffers,
|
|
buffer_size))
|
|
return throughput
|
|
|
|
@BluetoothBaseTest.bt_test_wrap
|
|
@test_tracker_info(uuid='23afba5b-5801-42c2-8d7a-41510e91a605')
|
|
def test_bluetooth_throughput_large_buffer(self):
|
|
"""Tests the throughput over a series of data transfers with large
|
|
buffer size.
|
|
"""
|
|
|
|
metrics = {}
|
|
throughput_list = []
|
|
|
|
for transfer in range(300):
|
|
throughput = self._measure_throughput(1, 300)
|
|
self.log.info('Throughput: {} bytes-per-sec'.format(throughput))
|
|
throughput_list.append(throughput)
|
|
|
|
metrics['data_transfer_protocol'] = self.data_transfer_type
|
|
metrics['data_packet_size'] = 300
|
|
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(statistics.mean(
|
|
throughput_list))
|
|
|
|
proto = self.bt_logger.get_results(metrics,
|
|
self.__class__.__name__,
|
|
self.server_device,
|
|
self.client_device)
|
|
|
|
asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0,
|
|
'Minimum throughput must be greater than 0!',
|
|
extras=proto)
|
|
raise TestPass('Throughput test (large buffer) completed successfully',
|
|
extras=proto)
|
|
|
|
@BluetoothBaseTest.bt_test_wrap
|
|
@test_tracker_info(uuid='5472fe33-891e-4fa1-ba84-78fc6f6a2327')
|
|
def test_bluetooth_throughput_medium_buffer(self):
|
|
"""Tests the throughput over a series of data transfers with medium
|
|
buffer size.
|
|
"""
|
|
|
|
metrics = {}
|
|
throughput_list = []
|
|
|
|
for transfer in range(300):
|
|
throughput = self._measure_throughput(1, 100)
|
|
self.log.info('Throughput: {} bytes-per-sec'.format(throughput))
|
|
throughput_list.append(throughput)
|
|
|
|
metrics['data_transfer_protocol'] = self.data_transfer_type
|
|
metrics['data_packet_size'] = 100
|
|
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(statistics.mean(
|
|
throughput_list))
|
|
|
|
proto = self.bt_logger.get_results(metrics,
|
|
self.__class__.__name__,
|
|
self.server_device,
|
|
self.client_device)
|
|
|
|
asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0,
|
|
'Minimum throughput must be greater than 0!',
|
|
extras=proto)
|
|
raise TestPass('Throughput test (medium buffer) completed successfully',
|
|
extras=proto)
|
|
|
|
@BluetoothBaseTest.bt_test_wrap
|
|
@test_tracker_info(uuid='97589280-cefa-4ae4-b3fd-94ec9c1f4104')
|
|
def test_bluetooth_throughput_small_buffer(self):
|
|
"""Tests the throughput over a series of data transfers with small
|
|
buffer size.
|
|
"""
|
|
|
|
metrics = {}
|
|
throughput_list = []
|
|
|
|
for transfer in range(300):
|
|
throughput = self._measure_throughput(1, 10)
|
|
self.log.info('Throughput: {} bytes-per-sec'.format(throughput))
|
|
throughput_list.append(throughput)
|
|
|
|
metrics['data_transfer_protocol'] = self.data_transfer_type
|
|
metrics['data_packet_size'] = 10
|
|
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(statistics.mean(
|
|
throughput_list))
|
|
|
|
proto = self.bt_logger.get_results(metrics,
|
|
self.__class__.__name__,
|
|
self.server_device,
|
|
self.client_device)
|
|
|
|
asserts.assert_true(metrics['data_throughput_min_bytes_per_second'] > 0,
|
|
'Minimum throughput must be greater than 0!',
|
|
extras=proto)
|
|
raise TestPass('Throughput test (small buffer) completed successfully',
|
|
extras=proto)
|
|
|
|
@BluetoothBaseTest.bt_test_wrap
|
|
def test_maximum_buffer_size(self):
|
|
"""Calculates the maximum allowed buffer size for one packet."""
|
|
|
|
current_buffer_size = 1
|
|
while True:
|
|
self.log.info('Trying buffer size {}'.format(current_buffer_size))
|
|
try:
|
|
throughput = self._measure_throughput(1, current_buffer_size)
|
|
except Exception:
|
|
buffer_msg = ('Max buffer size: {} bytes'.
|
|
format(current_buffer_size - 1))
|
|
throughput_msg = ('Max throughput: {} bytes-per-second'.
|
|
format(throughput))
|
|
self.log.info(buffer_msg)
|
|
self.log.info(throughput_msg)
|
|
return True
|
|
current_buffer_size += 1
|