Make changes to run pixel as a2dp sink.
This CL also removes the deprecated apollo tests.
Bug: None
Change-Id: I5e68cac76fec658dbeada9643ad40864793e49e5
diff --git a/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py b/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py
index cc8bde9..9624602 100644
--- a/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py
+++ b/acts/framework/acts/test_utils/coex/CoexPerformanceBaseTest.py
@@ -21,6 +21,8 @@
from acts.metrics.loggers.blackbox import BlackboxMetricLogger
from acts.test_utils.bt.bt_test_utils import disable_bluetooth
+from acts.test_utils.coex.audio_test_utils import AudioCaptureResult
+from acts.test_utils.coex.audio_test_utils import get_audio_capture_device
from acts.test_utils.coex.CoexBaseTest import CoexBaseTest
from acts.test_utils.coex.coex_test_utils import bokeh_chart_plot
from acts.test_utils.coex.coex_test_utils import collect_bluetooth_manager_dumpsys_logs
@@ -30,8 +32,8 @@
from acts.test_utils.wifi.wifi_test_utils import wifi_test_device_init
from acts.utils import get_current_epoch_time
-RSSI_POLL_RESULTS = "Monitoring , Handle: 0x0003, POLL"
-RSSI_RESULTS = "Monitoring , Handle: 0x0003, "
+RSSI_POLL_RESULTS = 'Monitoring , Handle: 0x0003, POLL'
+RSSI_RESULTS = 'Monitoring , Handle: 0x0003, '
def get_atten_range(start, stop, step):
@@ -70,12 +72,13 @@
metric_name='wifi_range_metric')
def setup_class(self):
- req_params = ["test_params", "Attenuator"]
- self.unpack_userparams(req_params)
- if hasattr(self, "Attenuator"):
+ req_params = ['test_params', 'Attenuator']
+ opt_params = ['audio_params']
+ self.unpack_userparams(req_params, opt_params)
+ if hasattr(self, 'Attenuator'):
self.num_atten = self.attenuators[0].instrument.num_atten
else:
- self.log.error("Attenuator should be connected to run tests.")
+ self.log.error('Attenuator should be connected to run tests.')
return False
for i in range(self.num_atten):
self.attenuators[i].set_atten(0)
@@ -97,7 +100,9 @@
self.test_params["attenuation_step"]))
def setup_test(self):
- if "a2dp_streaming" in self.current_test_name:
+ if ('a2dp_streaming' in self.current_test_name and
+ hasattr(self, 'audio_params')):
+ self.audio = get_audio_capture_device(self.sec_ad, self.audio_params)
self.a2dp_streaming = True
for i in range(self.num_atten):
self.attenuators[i].set_atten(0)
@@ -212,6 +217,8 @@
return self.iperf_received, self.a2dp_dropped_list, False
time.sleep(5) # Time for attenuation to set.
begin_time = get_current_epoch_time()
+ if self.a2dp_streaming:
+ self.audio.start()
if called_func:
if not multithread_func(self.log, called_func):
self.iperf_received.append(float(str(
@@ -230,7 +237,9 @@
self.log.info('Android device: {}'.format((
adb_rssi_results[-1]['log_message']).split(',')[5]))
if self.a2dp_streaming:
- analysis_path = self.audio.audio_quality_analysis(self.log_path)
+ self.path = self.audio.stop()
+ analysis_path = AudioCaptureResult(
+ self.path).audio_quality_analysis(self.audio_params)
with open(analysis_path) as f:
self.rvr[bt_atten]["audio_artifacts"][atten] = f.readline()
content = json.loads(self.rvr[bt_atten]["audio_artifacts"][atten])
diff --git a/acts/framework/acts/test_utils/coex/coex_test_utils.py b/acts/framework/acts/test_utils/coex/coex_test_utils.py
index ae4c399..27b1584 100644
--- a/acts/framework/acts/test_utils/coex/coex_test_utils.py
+++ b/acts/framework/acts/test_utils/coex/coex_test_utils.py
@@ -73,41 +73,38 @@
AVRCP_WAIT_TIME = 3
-def avrcp_actions(pri_ad, audio_receiver):
+def avrcp_actions(pri_ad, bt_device):
"""Performs avrcp controls like volume up, volume down, skip next and
skip previous.
Args:
pri_ad: Android device.
- audio_receiver: Relay instance.
+ bt_device: bt device instance.
Returns:
True if successful, otherwise False.
"""
- if "Volume_up" and "Volume_down" in (audio_receiver.relays.keys()):
- current_volume = pri_ad.droid.getMediaVolume()
- audio_receiver.press_volume_up()
+ current_volume = pri_ad.droid.getMediaVolume()
+ for _ in range(5):
+ bt_device.volume_up()
time.sleep(AVRCP_WAIT_TIME)
- if current_volume == pri_ad.droid.getMediaVolume():
- pri_ad.log.error("Increase volume failed")
- return False
+ if current_volume == pri_ad.droid.getMediaVolume():
+ pri_ad.log.error("Increase volume failed")
+ return False
+ time.sleep(AVRCP_WAIT_TIME)
+ current_volume = pri_ad.droid.getMediaVolume()
+ for _ in range(5):
+ bt_device.volume_down()
time.sleep(AVRCP_WAIT_TIME)
- current_volume = pri_ad.droid.getMediaVolume()
- audio_receiver.press_volume_down()
- time.sleep(AVRCP_WAIT_TIME)
- if current_volume == pri_ad.droid.getMediaVolume():
- pri_ad.log.error("Decrease volume failed")
- return False
- else:
- pri_ad.log.warning("No volume control pins specfied in relay config.")
+ if current_volume == pri_ad.droid.getMediaVolume():
+ pri_ad.log.error("Decrease volume failed")
+ return False
- if "Next" and "Previous" in audio_receiver.relays.keys():
- audio_receiver.press_next()
- time.sleep(AVRCP_WAIT_TIME)
- audio_receiver.press_previous()
- time.sleep(AVRCP_WAIT_TIME)
- else:
- pri_ad.log.warning("No track change pins specfied in relay config.")
+ #TODO: (sairamganesh) validate next and previous calls.
+ bt_device.next()
+ time.sleep(AVRCP_WAIT_TIME)
+ bt_device.previous()
+ time.sleep(AVRCP_WAIT_TIME)
return True
diff --git a/acts/tests/google/coex/apollo_tests/ApolloBasicPerformanceTest.py b/acts/tests/google/coex/apollo_tests/ApolloBasicPerformanceTest.py
deleted file mode 100755
index 9ae508f..0000000
--- a/acts/tests/google/coex/apollo_tests/ApolloBasicPerformanceTest.py
+++ /dev/null
@@ -1,189 +0,0 @@
-#!/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.
-
-from acts.controllers.buds_lib.apollo_utils import get_serial_object
-from acts.test_utils.bt import BtEnum
-from acts.test_utils.coex.CoexPerformanceBaseTest import CoexPerformanceBaseTest
-from acts.test_utils.coex.coex_test_utils import perform_classic_discovery
-from acts.test_utils.coex.coex_test_utils import pair_and_connect_headset
-
-
-class ApolloBasicPerformanceTest(CoexPerformanceBaseTest):
- """Test suite to check A2DP Functionality with Wlan.
-
- Test Setup:
-
- Two Android device.
- One apollo board.
- """
- def setup_class(self):
- super().setup_class()
- req_params = ["serial_device"]
- self.unpack_userparams(req_params)
- self.buds_device = get_serial_object(self.pri_ad, self.serial_device)
- self.headset_mac_address = self.buds_device.bluetooth_address
-
- def perform_classic_discovery_with_iperf(self):
- """Wrapper function to start iperf traffic and classic discovery.
-
- Returns:
- True if successful, False otherwise.
- """
- tasks = [(perform_classic_discovery, (self.pri_ad,
- self.iperf['duration'],
- self.json_file,
- self.dev_list)),
- (self.run_iperf_and_get_result, ())]
- return self.set_attenuation_and_run_iperf(tasks)
-
- def connect_headset(self):
- """Connect to apollo headset.
-
- Returns:
- True if successful, False otherwise.
- """
- self.buds_device.send("ResetPair\n")
- self.buds_device.set_pairing_mode()
- if not pair_and_connect_headset(
- self.pri_ad, self.headset_mac_address,
- set([BtEnum.BluetoothProfile.A2DP.value])):
- self.log.error("Failed to pair and connect to headset")
- return False
- self.buds_device.set_stay_connected(1)
-
- def test_performance_with_bluetooth_discovery_tcp_ul(self):
- """Check throughput when bluetooth discovery is ongoing.
-
- This test is to start TCP-uplink traffic between host machine and
- android device and bluetooth discovery and checks throughput.
-
- Steps:
- 1. Start TCP-uplink traffic and bluetooth discovery parallelly.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_classic_discovery_with_iperf()
-
- def test_performance_with_bluetooth_discovery_tcp_dl(self):
- """Check throughput when bluetooth discovery is ongoing.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and bluetooth discovery and checks throughput.
-
- Steps:
- 1. Start TCP-downlink traffic and bluetooth discovery parallelly.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_classic_discovery_with_iperf()
-
- def test_performance_with_bluetooth_discovery_udp_ul(self):
- """Check throughput when bluetooth discovery is ongoing.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and bluetooth discovery and checks throughput.
-
- Steps:
- 1. Start UDP-uplink traffic and bluetooth discovery parallelly.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_classic_discovery_with_iperf()
-
- def test_performance_with_bluetooth_discovery_udp_dl(self):
- """Check throughput when bluetooth discovery is ongoing.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and bluetooth discovery and checks throughput.
-
- Steps:
- 1. Start UDP-downlink traffic and bluetooth discovery parallelly.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_classic_discovery_with_iperf()
-
- def test_inquiry_after_headset_connection_with_tcp_ul(self):
- """Starts TCP-uplink traffic, start inquiry after bluetooth connection.
-
- This test is to start TCP-uplink traffic between host machine and
- android device and test functional behaviour of bluetooth discovery
- after connecting to headset.
-
- Steps:
- 1. Run TCP-uplink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- self.connect_headset()
- return self.perform_classic_discovery_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_tcp_dl(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run TCP-downlink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- self.connect_headset()
- return self.perform_classic_discovery_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_udp_ul(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run UDP-uplink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- self.connect_headset()
- return self.perform_classic_discovery_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_udp_dl(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run UDP-downlink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- self.connect_headset()
- return self.perform_classic_discovery_with_iperf()
diff --git a/acts/tests/google/coex/apollo_tests/ApolloWithA2dpPerformanceTest.py b/acts/tests/google/coex/apollo_tests/ApolloWithA2dpPerformanceTest.py
deleted file mode 100755
index 87f56bf..0000000
--- a/acts/tests/google/coex/apollo_tests/ApolloWithA2dpPerformanceTest.py
+++ /dev/null
@@ -1,288 +0,0 @@
-# /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.
-
-from acts.controllers.buds_lib.apollo_utils import avrcp_actions
-from acts.controllers.buds_lib.apollo_utils import get_serial_object
-from acts.test_utils.bt import BtEnum
-from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
-from acts.test_utils.coex.audio_test_utils import SshAudioCapture
-from acts.test_utils.coex.CoexPerformanceBaseTest import CoexPerformanceBaseTest
-from acts.test_utils.coex.coex_test_utils import music_play_and_check
-from acts.test_utils.coex.coex_test_utils import pair_and_connect_headset
-from acts.test_utils.coex.coex_test_utils import perform_classic_discovery
-from acts.test_utils.coex.coex_test_utils import push_music_to_android_device
-
-AVRCP_WAIT_TIME = 5
-
-
-class ApolloWithA2dpPerformanceTest(CoexPerformanceBaseTest):
-
- def setup_class(self):
- super().setup_class()
- req_params = ["serial_device", "audio_params"]
- self.unpack_userparams(req_params)
- self.buds_device = get_serial_object(self.pri_ad, self.serial_device)
- self.headset_mac_address = self.buds_device.bluetooth_address
- self.music_file_to_play = push_music_to_android_device(
- self.pri_ad, self.audio_params)
-
- def setup_test(self):
- super().setup_test()
- if "a2dp_streaming" in self.current_test_name:
- self.audio = SshAudioCapture(self.audio_params, self.log_path)
- self.buds_device.send("ResetPair\n")
- self.buds_device.set_pairing_mode()
- if not pair_and_connect_headset(
- self.pri_ad, self.headset_mac_address,
- set([BtEnum.BluetoothProfile.A2DP.value])):
- self.log.error("Failed to pair and connect to headset")
- return False
- self.buds_device.set_stay_connected(1)
-
- def teardown_test(self):
- if "a2dp_streaming" in self.current_test_name:
- self.audio.terminate_and_store_audio_results()
- clear_bonded_devices(self.pri_ad)
- super().teardown_test()
-
- def teardown_class(self):
- super().teardown_class()
-
- def initiate_music_streaming_to_headset_with_iperf(self):
- """Initiate music streaming to headset and start iperf traffic."""
- tasks = [(self.audio.capture_audio, ()),
- (music_play_and_check, (self.pri_ad,
- self.headset_mac_address,
- self.music_file_to_play,
- self.audio_params[
- 'music_play_time'])),
- (self.run_iperf_and_get_result, ())]
- return self.set_attenuation_and_run_iperf(tasks)
-
- def perform_discovery_with_iperf(self):
- """Starts iperf traffic based on test and perform bluetooth classic
- discovery.
- """
- tasks = [(self.run_iperf_and_get_result, ()),
- (perform_classic_discovery, (self.pri_ad,
- self.iperf['duration'],
- self.json_file,
- self.dev_list))]
- return self.set_attenuation_and_run_iperf(tasks)
-
- def music_streaming_and_avrcp_controls_with_iperf(self):
- """Starts iperf traffic based on test and initiate music streaming and
- check for avrcp controls.
- """
- tasks = [(self.audio.capture_audio, ()),
- (music_play_and_check, (self.pri_ad,
- self.headset_mac_address,
- self.music_file_to_play,
- self.iperf['duration'])),
- (self.run_iperf_and_get_result, ()),
- (avrcp_actions, (self.pri_ad,
- self.buds_device))]
- return self.set_attenuation_and_run_iperf(tasks)
-
- def test_performance_a2dp_streaming_tcp_ul(self):
- """Performance test to check throughput when streaming music.
- This test is to start TCP-uplink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start TCP-uplink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.initiate_music_streaming_to_headset_with_iperf()
-
- def test_performance_a2dp_streaming_tcp_dl(self):
- """Performance test to check throughput when streaming music.
- This test is to start TCP-downlink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start TCP-downlink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.initiate_music_streaming_to_headset_with_iperf()
-
- def test_performance_a2dp_streaming_udp_ul(self):
- """Performance test to check throughput when streaming music.
- This test is to start UDP-uplink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.initiate_music_streaming_to_headset_with_iperf()
-
- def test_performance_a2dp_streaming_udp_dl(self):
- """Performance test to check throughput when streaming music.
- This test is to start UDP-downlink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start UDP-downlink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.initiate_music_streaming_to_headset_with_iperf()
-
- def test_performance_a2dp_streaming_avrcp_controls_with_tcp_ul(self):
- """Performance test to check throughput when music streaming.
- This test is to start TCP-uplink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start TCP-uplink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.music_streaming_and_avrcp_controls_with_iperf()
-
- def test_performance_a2dp_streaming_avrcp_controls_with_tcp_dl(self):
- """Performance test to check throughput when music streaming.
- This test is to start TCP-downlink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start TCP-downlink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.music_streaming_and_avrcp_controls_with_iperf()
-
- def test_performance_a2dp_streaming_avrcp_controls_with_udp_ul(self):
- """Performance test to check throughput when music streaming.
- This test is to start UDP-uplink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.music_streaming_and_avrcp_controls_with_iperf()
-
- def test_performance_a2dp_streaming_avrcp_controls_with_udp_dl(self):
- """Performance test to check throughput when music streaming.
- This test is to start UDP-uplink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start UDP-downlink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.music_streaming_and_avrcp_controls_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_tcp_ul(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start TCP-uplink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run TCP-uplink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_discovery_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_tcp_dl(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run TCP-downlink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_030
- """
- return self.perform_discovery_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_udp_ul(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run UDP-uplink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_discovery_with_iperf()
-
- def test_performance_inquiry_after_headset_connection_with_udp_dl(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run UDP-downlink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
- """
- return self.perform_discovery_with_iperf()
diff --git a/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py b/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py
index f3f838d..e06b4c6 100644
--- a/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py
+++ b/acts/tests/google/coex/performance_tests/WlanWithA2dpPerformanceTest.py
@@ -13,83 +13,66 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
-"""
-Test suite to check Wlan performance with A2DP.
-Test Setup:
+import itertools
-One Android deivce.
-One A2DP Headset connected to Relay.
-"""
-import time
-
-from acts import asserts
+from acts.test_utils.abstract_devices.bluetooth_handsfree_abstract_device import BluetoothHandsfreeAbstractDeviceFactory as bf
from acts.test_utils.bt import BtEnum
from acts.test_utils.bt.bt_test_utils import clear_bonded_devices
from acts.test_utils.coex.CoexPerformanceBaseTest import CoexPerformanceBaseTest
-from acts.test_utils.coex.audio_test_utils import SshAudioCapture
from acts.test_utils.coex.coex_test_utils import avrcp_actions
from acts.test_utils.coex.coex_test_utils import music_play_and_check
from acts.test_utils.coex.coex_test_utils import pair_and_connect_headset
from acts.test_utils.coex.coex_test_utils import perform_classic_discovery
from acts.test_utils.coex.coex_test_utils import push_music_to_android_device
+
class WlanWithA2dpPerformanceTest(CoexPerformanceBaseTest):
+ def __init__(self, controllers):
+ super().__init__(controllers)
+ req_params = ['standalone_params', 'dut', 'music_file']
+ self.unpack_userparams(req_params)
+ self.tests = self.generate_test_cases([
+ 'a2dp_streaming_on_bt', 'perform_discovery_with_headset_connected',
+ 'a2dp_streaming_and_avrcp'])
+
def setup_class(self):
super().setup_class()
-
- req_params = ["iterations", "fping_params", "headset_mac_address",
- "audio_params"]
- self.unpack_userparams(req_params)
- if hasattr(self, "audio_params"):
- if self.audio_params["music_file"]:
- self.music_file_to_play = push_music_to_android_device(
- self.pri_ad, self.audio_params)
- if not self.music_file_to_play:
- self.log.error("Music file push failed.")
- return False
- else:
- self.log.warning("No Music files pushed to play.")
+ self.music_file_to_play = push_music_to_android_device(
+ self.pri_ad, self.music_file)
+ attr, idx = self.dut.split(':')
+ self.dut_controller = getattr(self, attr)[int(idx)]
+ self.bt_device = bf().generate(self.dut_controller)
def setup_test(self):
- if hasattr(self, "RelayDevice"):
- self.audio_receiver.enter_pairing_mode()
- time.sleep(5) # Wait until device goes into pairing mode.
- elif (not hasattr(self, "RelayDevice") and
- "avrcp" in self.current_test_name):
- asserts.skip("Relay device not connected,"
- "Hence avrcp tests can't be run")
super().setup_test()
- if "a2dp_streaming" in self.current_test_name:
- self.audio = SshAudioCapture(self.audio_params, self.log_path)
+ self.bt_device.power_on()
+ self.headset_mac_address = self.bt_device.mac_address
+ self.bt_device.enter_pairing_mode()
+ self.pri_ad.droid.bluetoothStartPairingHelper(True)
+ self.pri_ad.droid.bluetoothMakeDiscoverable()
if not pair_and_connect_headset(
self.pri_ad, self.headset_mac_address,
set([BtEnum.BluetoothProfile.A2DP.value])):
- self.log.error("Failed to pair and connect to headset")
+ self.log.error('Failed to pair and connect to headset')
return False
def teardown_test(self):
- if "a2dp_streaming" in self.current_test_name:
- self.audio.terminate_and_store_audio_results()
clear_bonded_devices(self.pri_ad)
- if hasattr(self, "RelayDevice"):
- self.audio_receiver.clean_up()
super().teardown_test()
- def initiate_music_streaming_to_headset_with_iperf(self):
+ def a2dp_streaming_on_bt(self):
"""Initiate music streaming to headset and start iperf traffic."""
- tasks = [(self.audio.capture_audio, ()),
+ tasks = [
(music_play_and_check,
(self.pri_ad, self.headset_mac_address,
self.music_file_to_play,
self.audio_params["music_play_time"])),
(self.run_iperf_and_get_result, ())]
- if not self.set_attenuation_and_run_iperf(tasks):
- return False
- return self.teardown_result()
+ return self.set_attenuation_and_run_iperf(tasks)
- def perform_discovery_with_iperf(self):
+ def perform_discovery_with_headset_connected(self):
"""Starts iperf traffic based on test and perform bluetooth classic
discovery.
"""
@@ -97,265 +80,30 @@
(perform_classic_discovery,
(self.pri_ad, self.iperf["duration"], self.json_file,
self.dev_list))]
- if not self.set_attenuation_and_run_iperf(tasks):
- return False
- return self.teardown_result()
+ return self.set_attenuation_and_run_iperf(tasks)
- def music_streaming_and_avrcp_controls_with_iperf(self):
+ def a2dp_streaming_and_avrcp(self):
"""Starts iperf traffic based on test and initiate music streaming and
check for avrcp controls.
"""
- tasks = [(self.audio.capture_audio, ()),
- (music_play_and_check,
+ tasks = [(music_play_and_check,
(self.pri_ad, self.headset_mac_address,
self.music_file_to_play,
self.audio_params["music_play_time"])),
(self.run_iperf_and_get_result, ()),
- (avrcp_actions, (self.pri_ad, self.audio_receiver))]
- if not self.set_attenuation_and_run_iperf(tasks):
- return False
- return self.teardown_result()
+ (avrcp_actions, (self.pri_ad, self.bt_device))]
+ return self.set_attenuation_and_run_iperf(tasks)
- def test_performance_a2dp_streaming_tcp_ul(self):
- """Performance test to check throughput when streaming music.
+ def generate_test_cases(self, test_types):
+ test_cases = []
+ for protocol, stream, test_type in itertools.product(
+ self.standalone_params['protocol'],
+ self.standalone_params['stream'], test_types):
- This test is to start TCP-uplink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
+ test_name = 'test_performance_with_{}_{}_{}'.format(
+ test_type, protocol, stream)
- Steps:
- 1. Start TCP-uplink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_013
- """
- if not self.initiate_music_streaming_to_headset_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_tcp_dl(self):
- """Performance test to check throughput when streaming music.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start TCP-downlink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_014
- """
- if not self.initiate_music_streaming_to_headset_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_udp_ul(self):
- """Performance test to check throughput when streaming music.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_015
- """
- if not self.initiate_music_streaming_to_headset_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_udp_dl(self):
- """Performance test to check throughput when streaming music.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and test the performance when music streamed to a2dp
- headset.
-
- Steps:
- 1. Start UDP-downlink traffic.
- 2. Start music streaming to a2dp headset.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_016
- """
- if not self.initiate_music_streaming_to_headset_with_iperf():
- return False
- return True
-
- def test_performance_discovery_after_headset_connection_with_tcp_ul(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start TCP-uplink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run TCP-uplink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_029
- """
- if not self.perform_discovery_with_iperf():
- return False
- return True
-
- def test_performance_discovery_after_headset_connection_with_tcp_dl(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run TCP-downlink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_030
- """
- if not self.perform_discovery_with_iperf():
- return False
- return True
-
- def test_performance_discovery_after_headset_connection_with_udp_ul(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run UDP-uplink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_031
- """
- if not self.perform_discovery_with_iperf():
- return False
- return True
-
- def test_performance_discovery_after_headset_connection_with_udp_dl(self):
- """Performance test to check throughput when bluetooth discovery.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and test the performance when bluetooth discovery is
- performed after connecting to headset.
-
- Steps:
- 1. Run UDP-downlink traffic.
- 2. Start bluetooth discovery when headset is connected.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_032
- """
- if not self.perform_discovery_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_avrcp_controls_with_tcp_ul(self):
- """Performance test to check throughput when music streaming.
-
- This test is to start TCP-uplink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start TCP-uplink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_033
- """
- if not self.music_streaming_and_avrcp_controls_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_avrcp_controls_with_tcp_dl(self):
- """Performance test to check throughput when music streaming.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start TCP-downlink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_034
- """
- if not self.music_streaming_and_avrcp_controls_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_avrcp_controls_with_udp_ul(self):
- """Performance test to check throughput when music streaming.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_035
- """
- if not self.music_streaming_and_avrcp_controls_with_iperf():
- return False
- return True
-
- def test_performance_a2dp_streaming_avrcp_controls_with_udp_dl(self):
- """Performance test to check throughput when music streaming.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the wlan throughput when perfroming a2dp music
- streaming and avrcp controls.
-
- Steps:
- 1. Start UDP-downlink traffic.
- 2. Start media streaming to a2dp headset.
- 3. Check all avrcp related controls.
-
- Returns:
- True if successful, False otherwise.
-
- Test Id: Bt_CoEx_036
- """
- if not self.music_streaming_and_avrcp_controls_with_iperf():
- return False
- return True
+ test_function = getattr(self, test_type)
+ setattr(self, test_name, test_function)
+ test_cases.append(test_name)
+ return test_cases