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 itertools 18 19from acts_contrib.test_utils.abstract_devices.bluetooth_handsfree_abstract_device import BluetoothHandsfreeAbstractDeviceFactory as bf 20from acts_contrib.test_utils.bt import BtEnum 21from acts_contrib.test_utils.bt.bt_test_utils import clear_bonded_devices 22from acts_contrib.test_utils.coex.CoexPerformanceBaseTest import CoexPerformanceBaseTest 23from acts_contrib.test_utils.coex.coex_test_utils import avrcp_actions 24from acts_contrib.test_utils.coex.coex_test_utils import music_play_and_check 25from acts_contrib.test_utils.coex.coex_test_utils import pair_and_connect_headset 26from acts_contrib.test_utils.coex.coex_test_utils import perform_classic_discovery 27from acts_contrib.test_utils.coex.coex_test_utils import push_music_to_android_device 28 29 30class WlanWithA2dpPerformanceTest(CoexPerformanceBaseTest): 31 32 def __init__(self, controllers): 33 super().__init__(controllers) 34 req_params = ['standalone_params', 'dut', 'music_file'] 35 self.unpack_userparams(req_params) 36 self.tests = self.generate_test_cases([ 37 'a2dp_streaming_on_bt', 'perform_discovery_with_headset_connected', 38 'a2dp_streaming_and_avrcp']) 39 40 def setup_class(self): 41 super().setup_class() 42 self.music_file_to_play = push_music_to_android_device( 43 self.pri_ad, self.music_file) 44 attr, idx = self.dut.split(':') 45 self.dut_controller = getattr(self, attr)[int(idx)] 46 self.bt_device = bf().generate(self.dut_controller) 47 48 def setup_test(self): 49 super().setup_test() 50 self.bt_device.power_on() 51 self.headset_mac_address = self.bt_device.mac_address 52 self.bt_device.enter_pairing_mode() 53 self.pri_ad.droid.bluetoothStartPairingHelper(True) 54 self.pri_ad.droid.bluetoothMakeDiscoverable() 55 if not pair_and_connect_headset( 56 self.pri_ad, self.headset_mac_address, 57 set([BtEnum.BluetoothProfile.A2DP.value])): 58 self.log.error('Failed to pair and connect to headset') 59 return False 60 61 def teardown_test(self): 62 clear_bonded_devices(self.pri_ad) 63 super().teardown_test() 64 65 def a2dp_streaming_on_bt(self): 66 """Initiate music streaming to headset and start iperf traffic.""" 67 tasks = [ 68 (music_play_and_check, 69 (self.pri_ad, self.headset_mac_address, 70 self.music_file_to_play, 71 self.audio_params["music_play_time"])), 72 (self.run_iperf_and_get_result, ())] 73 return self.set_attenuation_and_run_iperf(tasks) 74 75 def perform_discovery_with_headset_connected(self): 76 """Starts iperf traffic based on test and perform bluetooth classic 77 discovery. 78 """ 79 tasks = [(self.run_iperf_and_get_result, ()), 80 (perform_classic_discovery, 81 (self.pri_ad, self.iperf["duration"], self.json_file, 82 self.dev_list))] 83 return self.set_attenuation_and_run_iperf(tasks) 84 85 def a2dp_streaming_and_avrcp(self): 86 """Starts iperf traffic based on test and initiate music streaming and 87 check for avrcp controls. 88 """ 89 tasks = [(music_play_and_check, 90 (self.pri_ad, self.headset_mac_address, 91 self.music_file_to_play, 92 self.audio_params["music_play_time"])), 93 (self.run_iperf_and_get_result, ()), 94 (avrcp_actions, (self.pri_ad, self.bt_device))] 95 return self.set_attenuation_and_run_iperf(tasks) 96 97 def generate_test_cases(self, test_types): 98 test_cases = [] 99 for protocol, stream, test_type in itertools.product( 100 self.standalone_params['protocol'], 101 self.standalone_params['stream'], test_types): 102 103 test_name = 'test_performance_with_{}_{}_{}'.format( 104 test_type, protocol, stream) 105 106 test_function = getattr(self, test_type) 107 setattr(self, test_name, test_function) 108 test_cases.append(test_name) 109 return test_cases 110