Merge "New test cases test_browsing_4g & test_browsing_wifi"
diff --git a/PREUPLOAD.cfg b/PREUPLOAD.cfg
index 0e638ac..a9a2123 100644
--- a/PREUPLOAD.cfg
+++ b/PREUPLOAD.cfg
@@ -1,11 +1,10 @@
[Hook Scripts]
yapf_hook = ./tools/yapf_checker.py
+proto_check = ./tools/proto_check.py
create_virtualenv = ./tools/create_virtualenv.sh
acts_unittests = /tmp/acts_preupload_virtualenv/bin/python3 ./acts/tests/meta/ActsUnitTest.py
destroy_virtualenv = rm -rf /tmp/acts_preupload_virtualenv/
-proto_check = ./tools/proto_check.py
-
[Builtin Hooks]
commit_msg_bug_field = true
jsonlint = true
diff --git a/acts/framework/acts/controllers/ap_lib/hostapd_config.py b/acts/framework/acts/controllers/ap_lib/hostapd_config.py
index a6d71c2..47afb2e 100644
--- a/acts/framework/acts/controllers/ap_lib/hostapd_config.py
+++ b/acts/framework/acts/controllers/ap_lib/hostapd_config.py
@@ -463,11 +463,13 @@
logging.warning(
'No channel bandwidth specified. Using 80MHz for 11ac.')
self._vht_oper_chwidth = 1
- if not vht_channel_width == 20 and vht_center_channel is None:
+ if vht_center_channel is not None:
+ self._vht_oper_centr_freq_seg0_idx = vht_center_channel
+ elif vht_channel_width == 20:
+ self._vht_oper_centr_freq_seg0_idx = channel
+ else:
self._vht_oper_centr_freq_seg0_idx = self._get_11ac_center_channel_from_channel(
self.channel)
- else:
- self._vht_oper_centr_freq_seg0_idx = vht_center_channel
self._ac_capabilities = set(ac_capabilities)
self._beacon_footer = beacon_footer
self._spectrum_mgmt_required = spectrum_mgmt_required
diff --git a/acts/framework/acts/controllers/ap_lib/hostapd_constants.py b/acts/framework/acts/controllers/ap_lib/hostapd_constants.py
index 7f70dfe..588fbc2 100755
--- a/acts/framework/acts/controllers/ap_lib/hostapd_constants.py
+++ b/acts/framework/acts/controllers/ap_lib/hostapd_constants.py
@@ -270,11 +270,11 @@
HT40_ALLOW_MAP = {
N_CAPABILITY_HT40_MINUS_CHANNELS:
tuple(
- itertools.chain(range(6, 14), range(40, 65, 8), range(104, 137, 8),
+ itertools.chain(range(6, 14), range(40, 65, 8), range(104, 145, 8),
[153, 161])),
N_CAPABILITY_HT40_PLUS_CHANNELS:
tuple(
- itertools.chain(range(1, 8), range(36, 61, 8), range(100, 133, 8),
+ itertools.chain(range(1, 8), range(36, 61, 8), range(100, 141, 8),
[149, 157]))
}
diff --git a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/asus.py b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/asus.py
index af489d6..bb13118 100644
--- a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/asus.py
+++ b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/asus.py
@@ -87,6 +87,7 @@
# Common Parameters
rates = hostapd_constants.CCK_AND_OFDM_DATA_RATES
+ vht_channel_width = 20
n_capabilities = [
hostapd_constants.N_CAPABILITY_LDPC,
hostapd_constants.N_CAPABILITY_TX_STBC,
@@ -110,8 +111,6 @@
rates.update(hostapd_constants.CCK_AND_OFDM_BASIC_RATES)
mode = hostapd_constants.MODE_11N_MIXED
ac_capabilities = None
- vht_channel_width = None
- vht_center_channel = None
# 5GHz
else:
@@ -126,8 +125,6 @@
hostapd_constants.AC_CAPABILITY_MAX_MPDU_11454,
hostapd_constants.AC_CAPABILITY_MAX_A_MPDU_LEN_EXP7
]
- vht_channel_width = 40
- vht_center_channel = 36
additional_params = utils.merge_dicts(rates, vendor_elements,
hostapd_constants.UAPSD_ENABLED)
@@ -146,7 +143,6 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
additional_parameters=additional_params)
return config
@@ -313,6 +309,7 @@
# Common Parameters
rates = hostapd_constants.CCK_AND_OFDM_DATA_RATES
+ vht_channel_width = 20
qbss = {'bss_load_update_period': 50, 'chan_util_avg_period': 600}
n_capabilities = [
hostapd_constants.N_CAPABILITY_LDPC,
@@ -320,6 +317,7 @@
hostapd_constants.N_CAPABILITY_RX_STBC1,
hostapd_constants.N_CAPABILITY_SGI20
]
+
# Broadcom IE
vendor_elements = {'vendor_elements': 'dd090010180200009c0000'}
@@ -334,8 +332,6 @@
'2fd437509c30b3d7f5cf5754fb125aed3b8507045aed3b85' \
'dd1e00904c0418bf0cb2798b0faaff0000aaff0000c0050001000000c3020002'
ac_capabilities = None
- vht_channel_width = None
- vht_center_channel = None
# 5GHz
else:
@@ -352,8 +348,6 @@
hostapd_constants.AC_CAPABILITY_MAX_MPDU_11454,
hostapd_constants.AC_CAPABILITY_MAX_A_MPDU_LEN_EXP7
]
- vht_channel_width = 40
- vht_center_channel = 36
additional_params = utils.merge_dicts(rates, qbss, vendor_elements,
hostapd_constants.UAPSD_ENABLED)
@@ -372,7 +366,6 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
additional_parameters=additional_params)
return config
diff --git a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/linksys.py b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/linksys.py
index 1ee4869..8bb2841 100644
--- a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/linksys.py
+++ b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/linksys.py
@@ -258,8 +258,7 @@
hostapd_constants.AC_CAPABILITY_TX_ANTENNA_PATTERN,
hostapd_constants.AC_CAPABILITY_MAX_A_MPDU_LEN_EXP7
]
- vht_channel_width = 40
- vht_center_channel = 0
+ vht_channel_width = 20
# Epigram, Inc. HT Capabilities IE
# Epigram, Inc. HT Additional Capabilities IE
# Marvell Semiconductor IE
@@ -308,7 +307,6 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
spectrum_mgmt_required=spectrum_mgmt,
additional_parameters=additional_params)
return config
diff --git a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/netgear.py b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/netgear.py
index 331dcdf..922d51a 100644
--- a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/netgear.py
+++ b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/netgear.py
@@ -91,6 +91,7 @@
# Common Parameters
rates = hostapd_constants.CCK_AND_OFDM_DATA_RATES
+ vht_channel_width = 80
n_capabilities = [
hostapd_constants.N_CAPABILITY_LDPC,
hostapd_constants.N_CAPABILITY_TX_STBC,
@@ -119,8 +120,6 @@
mode = hostapd_constants.MODE_11N_MIXED
obss_interval = 300
ac_capabilities = None
- vht_channel_width = None
- vht_center_channel = None
# 5GHz
else:
@@ -129,8 +128,13 @@
mode = hostapd_constants.MODE_11AC_MIXED
n_capabilities += [
hostapd_constants.N_CAPABILITY_SGI40,
- hostapd_constants.N_CAPABILITY_HT40_PLUS
]
+
+ if hostapd_config.ht40_plus_allowed(channel):
+ n_capabilities.append(hostapd_constants.N_CAPABILITY_HT40_PLUS)
+ elif hostapd_config.ht40_minus_allowed(channel):
+ n_capabilities.append(hostapd_constants.N_CAPABILITY_HT40_MINUS)
+
obss_interval = None
ac_capabilities = [
hostapd_constants.AC_CAPABILITY_RXLDPC,
@@ -140,8 +144,6 @@
hostapd_constants.AC_CAPABILITY_MAX_MPDU_11454,
hostapd_constants.AC_CAPABILITY_MAX_A_MPDU_LEN_EXP7
]
- vht_channel_width = 80
- vht_center_channel = 42
additional_params = utils.merge_dicts(
rates, vendor_elements, qbss,
@@ -164,7 +166,6 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
additional_parameters=additional_params)
return config
diff --git a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/tplink.py b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/tplink.py
index 15961ba..e3e9894 100644
--- a/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/tplink.py
+++ b/acts/framework/acts/controllers/ap_lib/third_party_ap_profiles/tplink.py
@@ -85,6 +85,7 @@
# Common Parameters
rates = hostapd_constants.CCK_AND_OFDM_DATA_RATES
+ vht_channel_width = 20
n_capabilities = [
hostapd_constants.N_CAPABILITY_SGI20,
hostapd_constants.N_CAPABILITY_TX_STBC,
@@ -109,8 +110,6 @@
mode = hostapd_constants.MODE_11N_MIXED
n_capabilities.append(hostapd_constants.N_CAPABILITY_DSSS_CCK_40)
ac_capabilities = None
- vht_channel_width = None
- vht_center_channel = None
# 5GHz
else:
@@ -127,8 +126,6 @@
hostapd_constants.AC_CAPABILITY_RX_STBC_1,
hostapd_constants.AC_CAPABILITY_MAX_A_MPDU_LEN_EXP7,
]
- vht_channel_width = 40
- vht_center_channel = 36
additional_params = utils.merge_dicts(
rates, vendor_elements, qbss,
@@ -150,7 +147,6 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
additional_parameters=additional_params)
return config
@@ -199,6 +195,7 @@
# Common Parameters
rates = hostapd_constants.CCK_AND_OFDM_DATA_RATES
+ vht_channel_width = 80
n_capabilities = [
hostapd_constants.N_CAPABILITY_LDPC,
hostapd_constants.N_CAPABILITY_SGI20,
@@ -223,7 +220,6 @@
pwr_constraint = {}
ac_capabilities = None
vht_channel_width = None
- vht_center_channel = None
# 5GHz
else:
@@ -239,10 +235,15 @@
'8801178c011795011e99011e9d011ea1011ea5011e')
pwr_constraint = {'local_pwr_constraint': 3}
n_capabilities += [
- hostapd_constants.N_CAPABILITY_HT40_PLUS,
hostapd_constants.N_CAPABILITY_SGI40,
hostapd_constants.N_CAPABILITY_MAX_AMSDU_7935
]
+
+ if hostapd_config.ht40_plus_allowed(channel):
+ n_capabilities.append(hostapd_constants.N_CAPABILITY_HT40_PLUS)
+ elif hostapd_config.ht40_minus_allowed(channel):
+ n_capabilities.append(hostapd_constants.N_CAPABILITY_HT40_MINUS)
+
ac_capabilities = [
hostapd_constants.AC_CAPABILITY_MAX_MPDU_11454,
hostapd_constants.AC_CAPABILITY_RXLDPC,
@@ -253,8 +254,6 @@
hostapd_constants.AC_CAPABILITY_RX_ANTENNA_PATTERN,
hostapd_constants.AC_CAPABILITY_TX_ANTENNA_PATTERN
]
- vht_channel_width = 80
- vht_center_channel = 42
additional_params = utils.merge_dicts(rates, vendor_elements,
hostapd_constants.UAPSD_ENABLED,
@@ -274,7 +273,6 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
spectrum_mgmt_required=spectrum_mgmt,
additional_parameters=additional_params)
return config
@@ -334,6 +332,7 @@
# Common Parameters
rates = hostapd_constants.CCK_AND_OFDM_DATA_RATES
+ vht_channel_width = 20
n_capabilities = [
hostapd_constants.N_CAPABILITY_SGI20,
hostapd_constants.N_CAPABILITY_TX_STBC,
@@ -356,8 +355,6 @@
short_preamble = True
mode = hostapd_constants.MODE_11N_MIXED
ac_capabilities = None
- vht_channel_width = None
- vht_center_channel = None
# 5GHz
else:
@@ -374,8 +371,6 @@
hostapd_constants.AC_CAPABILITY_RX_STBC_1,
hostapd_constants.AC_CAPABILITY_MAX_A_MPDU_LEN_EXP7,
]
- vht_channel_width = 40
- vht_center_channel = 36
additional_params = utils.merge_dicts(
rates, vendor_elements, hostapd_constants.ENABLE_RRM_BEACON_REPORT,
@@ -396,12 +391,13 @@
n_capabilities=n_capabilities,
ac_capabilities=ac_capabilities,
vht_channel_width=vht_channel_width,
- vht_center_channel=vht_center_channel,
additional_parameters=additional_params)
return config
-def tplink_tlwr940n(iface_wlan_2g=None, channel=None, security=None,
+def tplink_tlwr940n(iface_wlan_2g=None,
+ channel=None,
+ security=None,
ssid=None):
# TODO(b/143104825): Permit RIFS once it is supported
"""A simulated implementation of an TPLink TLWR940N AP.
diff --git a/acts/framework/acts/test_utils/bt/bt_test_utils.py b/acts/framework/acts/test_utils/bt/bt_test_utils.py
index aea9fb2..83b74e1 100644
--- a/acts/framework/acts/test_utils/bt/bt_test_utils.py
+++ b/acts/framework/acts/test_utils/bt/bt_test_utils.py
@@ -24,6 +24,7 @@
from queue import Empty
from subprocess import call
+from acts import asserts
from acts.test_utils.bt.bt_constants import adv_fail
from acts.test_utils.bt.bt_constants import adv_succ
from acts.test_utils.bt.bt_constants import batch_scan_not_supported_list
@@ -58,7 +59,6 @@
from acts.test_utils.tel.tel_test_utils import verify_http_connection
from acts.utils import exe_cmd
-from acts import context
from acts import utils
log = logging
@@ -1745,3 +1745,55 @@
log.error("Mismatch! Read: {}, Expected: {}".format(read_msg, msg))
return False
return True
+
+
+class MediaControlOverSl4a(object):
+ """Media control using sl4a facade for general purpose.
+
+ """
+ def __init__(self, android_device, music_file):
+ """Initialize the media_control class.
+
+ Args:
+ android_dut: android_device object
+ music_file: location of the music file
+ """
+ self.android_device = android_device
+ self.music_file = music_file
+
+ def play(self):
+ """Play media.
+
+ """
+ self.android_device.droid.mediaPlayOpen('file://%s' % self.music_file,
+ 'default', True)
+ playing = self.android_device.droid.mediaIsPlaying()
+ asserts.assert_true(playing,
+ 'Failed to play music %s' % self.music_file)
+
+ def pause(self):
+ """Pause media.
+
+ """
+ self.android_device.droid.mediaPlayPause('default')
+ paused = not self.android_device.droid.mediaIsPlaying()
+ asserts.assert_true(paused,
+ 'Failed to pause music %s' % self.music_file)
+
+ def resume(self):
+ """Resume media.
+
+ """
+ self.android_device.droid.mediaPlayStart('default')
+ playing = self.android_device.droid.mediaIsPlaying()
+ asserts.assert_true(playing,
+ 'Failed to play music %s' % self.music_file)
+
+ def stop(self):
+ """Stop media.
+
+ """
+ self.android_device.droid.mediaPlayStop('default')
+ stopped = not self.android_device.droid.mediaIsPlaying()
+ asserts.assert_true(stopped,
+ 'Failed to stop music %s' % self.music_file)
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/audio_test_utils.py b/acts/framework/acts/test_utils/coex/audio_test_utils.py
index 02b99ca..f4dc403 100644
--- a/acts/framework/acts/test_utils/coex/audio_test_utils.py
+++ b/acts/framework/acts/test_utils/coex/audio_test_utils.py
@@ -15,7 +15,9 @@
# the License.
import logging
+import numpy
import os
+import scipy.io.wavfile as sciwav
from acts.test_utils.coex.audio_capture_device import AudioCaptureBase
from acts.test_utils.coex.audio_capture_device import CaptureAudioOverAdb
@@ -65,8 +67,7 @@
class AudioCaptureResult(AudioCaptureBase):
-
- def __init__(self, path):
+ def __init__(self, path, audio_params=None):
"""Initializes Audio Capture Result class.
Args:
@@ -74,7 +75,11 @@
"""
super().__init__()
self.path = path
- self.analysis_path = os.path.join(self.log_path, ANALYSIS_FILE_TEMPLATE)
+ self.audio_params = audio_params
+ self.analysis_path = os.path.join(self.log_path,
+ ANALYSIS_FILE_TEMPLATE)
+ if self.audio_params:
+ self._trim_wave_file()
def THDN(self, win_size=None, step_size=None, q=1, freq=None):
"""Calculate THD+N value for most recently recorded file.
@@ -104,7 +109,8 @@
q=q,
freq=freq)
- def detect_anomalies(self, freq=None,
+ def detect_anomalies(self,
+ freq=None,
block_size=ANOMALY_DETECTION_BLOCK_SIZE,
threshold=PATTERN_MATCHING_THRESHOLD,
tolerance=ANOMALY_GROUPING_TOLERANCE):
@@ -127,12 +133,11 @@
channel_results (list): anomaly durations for each channel's signal.
List index corresponds to channel index.
"""
- return audio_analysis.get_file_anomaly_durations(
- filename=self.path,
- freq=freq,
- block_size=block_size,
- threshold=threshold,
- tolerance=tolerance)
+ return audio_analysis.get_file_anomaly_durations(filename=self.path,
+ freq=freq,
+ block_size=block_size,
+ threshold=threshold,
+ tolerance=tolerance)
@property
def analysis_fileno(self):
@@ -142,12 +147,9 @@
counter += 1
return counter
- def audio_quality_analysis(self, audio_params):
+ def audio_quality_analysis(self):
"""Measures audio quality based on the audio file given as input.
- Args:
- path: Log path
-
Returns:
analysis_path on success.
"""
@@ -155,14 +157,35 @@
if not os.path.exists(self.path):
raise FileNotFound("Recorded file not found")
try:
- quality_analysis(
- filename=self.path,
- output_file=analysis_path,
- bit_width=bits_per_sample,
- rate=audio_params["sample_rate"],
- channel=audio_params["channel"],
- spectral_only=False)
+ quality_analysis(filename=self.path,
+ output_file=analysis_path,
+ bit_width=bits_per_sample,
+ rate=self.audio_params["sample_rate"],
+ channel=self.audio_params["channel"],
+ spectral_only=False)
except Exception as err:
logging.exception("Failed to analyze raw audio: %s" % err)
return analysis_path
+ def _trim_wave_file(self):
+ """Trim wave files.
+
+ """
+ original_record_file_name = 'original_' + os.path.basename(self.path)
+ original_record_file_path = os.path.join(os.path.dirname(self.path),
+ original_record_file_name)
+ os.rename(self.path, original_record_file_path)
+ fs, data = sciwav.read(original_record_file_path)
+ trim_start = self.audio_params['trim_start']
+ trim_end = self.audio_params['trim_end']
+ trim = numpy.array([[trim_start, trim_end]])
+ trim = trim * fs
+ new_wave_file_list = []
+ for elem in trim:
+ # To check start and end doesn't exceed raw data dimension
+ start_read = min(elem[0], data.shape[0] - 1)
+ end_read = min(elem[1], data.shape[0] - 1)
+ new_wave_file_list.extend(data[start_read:end_read])
+ new_wave_file = numpy.array(new_wave_file_list)
+
+ sciwav.write(self.path, fs, new_wave_file)
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/framework/acts/test_utils/instrumentation/app_installer.py b/acts/framework/acts/test_utils/instrumentation/app_installer.py
index c27d7e7..5c74f8c 100644
--- a/acts/framework/acts/test_utils/instrumentation/app_installer.py
+++ b/acts/framework/acts/test_utils/instrumentation/app_installer.py
@@ -24,81 +24,96 @@
class AppInstaller(object):
- """Class for installing apps on an Android device."""
- def __init__(self, device):
- self.ad = device
- self._pkgs = {}
-
- def install(self, apk_path, *extra_args):
- """Installs an apk on the device.
+ """Class that represents an app on an Android device. Includes methods
+ for install, uninstall, and getting info.
+ """
+ def __init__(self, ad, apk_path):
+ """Initializes an AppInstaller.
Args:
- apk_path: Path to the apk to install
+ ad: device to install the apk
+ apk_path: path to the apk
+ """
+ self._ad = ad
+ self._apk_path = apk_path
+ self._pkg_name = None
+
+ @staticmethod
+ def pull_from_device(ad, pkg_name, dest):
+ """Initializes an AppInstaller by pulling the apk file from the device,
+ given the package name
+
+ Args:
+ ad: device on which the apk is installed
+ pkg_name: package name
+ dest: destination directory
+ (Note: If path represents a directory, it must already exist as
+ a directory)
+
+ Returns: AppInstaller object representing the pulled apk, or None if
+ package not installed
+ """
+ if not ad.is_apk_installed(pkg_name):
+ ad.log.warning('Unable to find package %s on device. Pull aborted.'
+ % pkg_name)
+ return None
+ path_on_device = re.compile(PM_PATH_PATTERN).search(
+ ad.adb.shell('pm path %s' % pkg_name)).group('apk_path')
+ ad.pull_files(path_on_device, dest)
+ if os.path.isdir(dest):
+ dest = os.path.join(dest, os.path.basename(path_on_device))
+ return AppInstaller(ad, dest)
+
+ @property
+ def apk_path(self):
+ return self._apk_path
+
+ @property
+ def pkg_name(self):
+ """Get the package name corresponding to the apk from aapt
+
+ Returns: The package name, or empty string if not found.
+ """
+ if self._pkg_name is None:
+ dump = job.run(
+ 'aapt dump badging %s' % self.apk_path,
+ ignore_status=True).stdout
+ match = re.compile(PKG_NAME_PATTERN).search(dump)
+ self._pkg_name = match.group('pkg_name') if match else ''
+ return self._pkg_name
+
+ def install(self, *extra_args):
+ """Installs the apk on the device.
+
+ Args:
extra_args: Additional flags to the ADB install command.
Note that '-r' is included by default.
"""
- self.ad.log.info('Installing app %s' % apk_path)
- self.ad.ensure_screen_on()
+ self._ad.log.info('Installing app %s' % self.apk_path)
+ self._ad.ensure_screen_on()
args = '-r %s' % ' '.join(extra_args)
- self.ad.adb.install('%s %s' % (args, apk_path))
+ self._ad.adb.install('%s %s' % (args, self.apk_path))
- def uninstall(self, apk_path, *extra_args):
- """Finds the package corresponding to the apk and uninstalls it from the
- device.
+ def uninstall(self, *extra_args):
+ """Uninstalls the apk from the device.
Args:
- apk_path: Path to the apk
extra_args: Additional flags to the uninstall command.
"""
- if self.is_installed(apk_path):
- pkg_name = self.get_package_name(apk_path)
- self.ad.log.info('Uninstalling app %s' % pkg_name)
- self.ad.adb.shell(
- 'pm uninstall %s %s' % (' '.join(extra_args), pkg_name))
+ self._ad.log.info('Uninstalling app %s' % self.pkg_name)
+ if not self.is_installed():
+ self._ad.log.warning('Unable to uninstall app %s. App is not '
+ 'installed.' % self.pkg_name)
+ return
+ self._ad.adb.shell(
+ 'pm uninstall %s %s' % (' '.join(extra_args), self.pkg_name))
- def is_installed(self, apk_path):
- """Verifies that an apk is installed on the device.
-
- Args:
- apk_path: Path to the apk
+ def is_installed(self):
+ """Verifies that the apk is installed on the device.
Returns: True if the apk is installed on the device.
"""
- pkg_name = self.get_package_name(apk_path)
- if not pkg_name:
- self.ad.log.warning('No package name found for %s' % apk_path)
+ if not self.pkg_name:
+ self._ad.log.warning('No package name found for %s' % self.apk_path)
return False
- return self.ad.is_apk_installed(pkg_name)
-
- def get_package_name(self, apk_path):
- """Get the package name corresponding to the apk from aapt
-
- Args:
- apk_path: Path to the apk
-
- Returns: The package name
- """
- if apk_path not in self._pkgs:
- dump = job.run(
- 'aapt dump badging %s' % apk_path, ignore_status=True).stdout
- match = re.compile(PKG_NAME_PATTERN).search(dump)
- self._pkgs[apk_path] = match.group('pkg_name') if match else ''
- return self._pkgs[apk_path]
-
- def pull_apk(self, package_name, dest):
- """Pull the corresponding apk file from device given the package name
-
- Args:
- package_name: Package name
- dest: Destination directory
-
- Returns: Path to the pulled apk, or None if package not installed
- """
- if not self.ad.is_apk_installed(package_name):
- self.ad.log.warning('Unable to find package %s on device. Pull '
- 'aborted.' % package_name)
- return None
- apk_path = re.compile(PM_PATH_PATTERN).search(
- self.ad.adb.shell('pm path %s' % package_name)).group('apk_path')
- self.ad.pull_files(apk_path, dest)
- return os.path.join(dest, os.path.basename(apk_path))
+ return self._ad.is_apk_installed(self.pkg_name)
diff --git a/acts/framework/acts/test_utils/instrumentation/instrumentation_base_test.py b/acts/framework/acts/test_utils/instrumentation/instrumentation_base_test.py
index dbf8360..6b85a87 100644
--- a/acts/framework/acts/test_utils/instrumentation/instrumentation_base_test.py
+++ b/acts/framework/acts/test_utils/instrumentation/instrumentation_base_test.py
@@ -22,10 +22,10 @@
from acts import context
from acts import utils
from acts.keys import Config
-from acts.test_utils.instrumentation import app_installer
from acts.test_utils.instrumentation import instrumentation_proto_parser \
as proto_parser
from acts.test_utils.instrumentation.adb_commands import common
+from acts.test_utils.instrumentation.app_installer import AppInstaller
from acts.test_utils.instrumentation.config_wrapper import ConfigWrapper
from acts.test_utils.instrumentation.instrumentation_command_builder import \
InstrumentationCommandBuilder
@@ -126,7 +126,6 @@
def setup_class(self):
"""Class setup"""
self.ad_dut = self.android_devices[0]
- self.ad_apps = app_installer.AppInstaller(self.ad_dut)
self._prepare_device()
def teardown_class(self):
@@ -274,15 +273,15 @@
# Install PermissionUtils.apk
permissions_apk_path = self._instrumentation_config.get_file(
'permissions_apk')
- self.ad_apps.install(permissions_apk_path)
- if not self.ad_apps.is_installed(permissions_apk_path):
+ permission_utils = AppInstaller(self.ad_dut, permissions_apk_path)
+ permission_utils.install()
+ if not permission_utils.is_installed():
raise InstrumentationTestError(
'Failed to install PermissionUtils.apk, abort!')
- package_name = self.ad_apps.get_package_name(permissions_apk_path)
# Run the instrumentation command
cmd_builder = InstrumentationCommandBuilder()
- cmd_builder.set_manifest_package(package_name)
+ cmd_builder.set_manifest_package(permission_utils.pkg_name)
cmd_builder.set_runner('.PermissionInstrumentation')
cmd_builder.add_flag('-w')
cmd_builder.add_flag('-r')
@@ -292,4 +291,4 @@
self.adb_run(cmd)
# Uninstall PermissionUtils.apk
- self.ad_apps.uninstall(permissions_apk_path)
+ permission_utils.uninstall()
diff --git a/acts/framework/acts/test_utils/power/PowerBTBaseTest.py b/acts/framework/acts/test_utils/power/PowerBTBaseTest.py
index bd7dcfb..8979822 100644
--- a/acts/framework/acts/test_utils/power/PowerBTBaseTest.py
+++ b/acts/framework/acts/test_utils/power/PowerBTBaseTest.py
@@ -37,14 +37,14 @@
obj_atten: attenuator object, a single port attenuator
attenuation_target: target attenuation level to reach to.
"""
- attenuation_step_max = 5
+ attenuation_step_max = 20
sign = lambda x: copysign(1, x)
attenuation_delta = obj_atten.get_atten() - attenuation_target
while abs(attenuation_delta) > attenuation_step_max:
attenuation_intermediate = obj_atten.get_atten(
) - sign(attenuation_delta) * attenuation_step_max
obj_atten.set_atten(attenuation_intermediate)
- time.sleep(2)
+ time.sleep(5)
attenuation_delta = obj_atten.get_atten() - attenuation_target
obj_atten.set_atten(attenuation_target)
diff --git a/acts/framework/acts/test_utils/wifi/WifiBaseTest.py b/acts/framework/acts/test_utils/wifi/WifiBaseTest.py
index cb7ab80..82d655f 100644
--- a/acts/framework/acts/test_utils/wifi/WifiBaseTest.py
+++ b/acts/framework/acts/test_utils/wifi/WifiBaseTest.py
@@ -24,6 +24,7 @@
import acts.controllers.access_point as ap
from acts import asserts
+from acts import signals
from acts import utils
from acts.base_test import BaseTestClass
from acts.signals import TestSignal
@@ -608,3 +609,46 @@
hostapd_constants.BAND_5G, channel_5g)
if not result:
raise ValueError("Failed to configure channel for 5G band.")
+
+ @staticmethod
+ def wifi_test_wrap(fn):
+ def _safe_wrap_test_case(self, *args, **kwargs):
+ test_id = "%s:%s:%s" % (self.__class__.__name__, self.test_name,
+ self.log_begin_time.replace(' ', '-'))
+ self.test_id = test_id
+ self.result_detail = ""
+ tries = int(self.user_params.get("wifi_auto_rerun", 3))
+ for ad in self.android_devices:
+ ad.log_path = self.log_path
+ for i in range(tries + 1):
+ result = True
+ if i > 0:
+ log_string = "[Test Case] RETRY:%s %s" % (i, self.test_name)
+ self.log.info(log_string)
+ self._teardown_test(self.test_name)
+ self._setup_test(self.test_name)
+ try:
+ result = fn(self, *args, **kwargs)
+ except signals.TestFailure as e:
+ self.log.warn("Error msg: %s" % e)
+ if self.result_detail:
+ signal.details = self.result_detail
+ result = False
+ except signals.TestSignal:
+ if self.result_detail:
+ signal.details = self.result_detail
+ raise
+ except Exception as e:
+ self.log.exception(e)
+ asserts.fail(self.result_detail)
+ if result is False:
+ if i < tries:
+ continue
+ else:
+ break
+ if result is not False:
+ asserts.explicit_pass(self.result_detail)
+ else:
+ asserts.fail(self.result_detail)
+
+ return _safe_wrap_test_case
diff --git a/acts/framework/setup.py b/acts/framework/setup.py
index 59210f9..2aa136c 100755
--- a/acts/framework/setup.py
+++ b/acts/framework/setup.py
@@ -66,9 +66,14 @@
self.test_suite = True
def run_tests(self):
- import pytest
- errno = pytest.main(self.test_args)
- sys.exit(errno)
+ test_path = os.path.join(os.path.dirname(__file__),
+ '../tests/meta/ActsUnitTest.py')
+ result = subprocess.Popen('python3 %s' % test_path,
+ stdout=sys.stdout,
+ stderr=sys.stderr,
+ shell=True)
+ result.communicate()
+ sys.exit(result.returncode)
class ActsInstallDependencies(cmd.Command):
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
diff --git a/acts/tests/google/coex/performance_tests/WlanWithBlePerformanceTest.py b/acts/tests/google/coex/performance_tests/WlanWithBlePerformanceTest.py
index 50825c9..23d7423 100644
--- a/acts/tests/google/coex/performance_tests/WlanWithBlePerformanceTest.py
+++ b/acts/tests/google/coex/performance_tests/WlanWithBlePerformanceTest.py
@@ -14,6 +14,7 @@
# License for the specific language governing permissions and limitations under
# the License.
+import itertools
import time
from acts.test_utils.bt.bt_gatt_utils import close_gatt_client
@@ -30,9 +31,17 @@
bluetooth_gatt_list = []
gatt_server_list = []
- def setup_class(self):
- super().setup_class()
-
+ def __init__(self, controllers):
+ super().__init__(controllers)
+ req_params = [
+ # A dict containing:
+ # protocol: A list containing TCP/UDP. Ex: protocol: ['tcp'].
+ # stream: A list containing ul/dl. Ex: stream: ['ul']
+ 'standalone_params'
+ ]
+ self.unpack_userparams(req_params)
+ self.tests = self.generate_test_cases(['start_stop_ble_scan',
+ 'ble_gatt_connection'])
def setup_test(self):
super().setup_test()
@@ -118,12 +127,12 @@
True if successful, False otherwise.
"""
start_time = time.time()
- while((time.time()) < (start_time + self.iperf["duration"])):
+ while(time.time()) < (start_time + self.iperf["duration"]):
self.pri_ad.droid.bluetoothEnableBLE()
- gatt_server_cb = \
- self.sec_ad.droid.gattServerCreateGattServerCallback()
- gatt_server = \
- self.sec_ad.droid.gattServerOpenGattServer(gatt_server_cb)
+ gatt_server_cb = (
+ self.sec_ad.droid.gattServerCreateGattServerCallback())
+ gatt_server = (
+ self.sec_ad.droid.gattServerOpenGattServer(gatt_server_cb))
self.gatt_server_list.append(gatt_server)
try:
bluetooth_gatt, gatt_callback, adv_callback = (
@@ -135,174 +144,27 @@
return False
self.adv_instances.append(adv_callback)
return self._orchestrate_gatt_disconnection(bluetooth_gatt,
- gatt_callback)
+ gatt_callback)
- def ble_start_stop_scan_with_iperf(self):
+ def start_stop_ble_scan(self):
tasks = [(self.ble_start_stop_scan, ()),
(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 ble_gatt_connection_with_iperf(self):
+ def ble_gatt_connection(self):
tasks = [(self.initiate_ble_gatt_connection, ()),
(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 test_performance_ble_scan_tcp_ul(self):
- """Test performance with ble scan.
+ 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):
+ test_name = 'test_performance_with_{}_{}_{}'.format(
+ test_type, protocol, stream)
- This test is to start TCP-uplink traffic between host machine and
- android device and test the wlan throughput when performing ble scan.
-
- Steps:
- 1. Start TCP-uplink traffic.
- 2. Start and stop BLE scan.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_021
- """
- if not self.ble_start_stop_scan_with_iperf():
- return False
- return True
-
- def test_performance_ble_scan_tcp_dl(self):
- """Test performance with ble scan.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the wlan throughput when performing ble scan.
-
- Steps:
- 1. Start TCP-downlink traffic.
- 2. Start and stop BLE scan.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_022
- """
- if not self.ble_start_stop_scan_with_iperf():
- return False
- return True
-
- def test_performance_ble_scan_udp_ul(self):
- """Test performance with ble scan.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the wlan throughput when performing ble scan.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Start and stop BLE scan.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_023
- """
- if not self.ble_start_stop_scan_with_iperf():
- return False
- return True
-
- def test_performance_ble_scan_udp_dl(self):
- """Test performance with ble scan.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and test the wlan throughput when performing ble scan.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Start and stop BLE scan.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_024
- """
- if not self.ble_start_stop_scan_with_iperf():
- return False
- return True
-
- def test_performance_ble_connect_tcp_ul(self):
- """Test performance with ble gatt connection.
-
- This test is to start TCP-uplink traffic between host machine and
- android device and test the wlan throughput when ble gatt connection
- is established.
-
- Steps:
- 1. Start TCP-uplink traffic.
- 2. Initiate gatt connection.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_025
- """
- if not self.ble_gatt_connection_with_iperf():
- return False
- return True
-
- def test_performance_ble_connect_tcp_dl(self):
- """Test performance with ble gatt connection.
-
- This test is to start TCP-downlink traffic between host machine and
- android device and test the wlan throughput when ble gatt connection
- is established.
-
- Steps:
- 1. Start TCP-downlink traffic.
- 2. Initiate gatt connection.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_026
- """
- if not self.ble_gatt_connection_with_iperf():
- return False
- return True
-
- def test_performance_ble_connect_udp_ul(self):
- """Test performance with ble gatt connection.
-
- This test is to start UDP-uplink traffic between host machine and
- android device and test the wlan throughput when ble gatt connection
- is established.
-
- Steps:
- 1. Start UDP-uplink traffic.
- 2. Initiate gatt connection.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_027
- """
- if not self.ble_gatt_connection_with_iperf():
- return False
- return True
-
- def test_performance_ble_connect_udp_dl(self):
- """Test performance with ble gatt connection.
-
- This test is to start UDP-downlink traffic between host machine and
- android device and test the wlan throughput when ble gatt connection
- is established.
-
- Steps:
- 1. Start UDP-downlink traffic.
- 2. Initiate gatt connection.
-
- Returns:
- True if pass, False otherwise.
-
- Test Id: Bt_CoEx_Kpi_028
- """
- if not self.ble_gatt_connection_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
diff --git a/acts/tests/google/wifi/WifiDppTest.py b/acts/tests/google/wifi/WifiDppTest.py
index d7788db..42591b0 100644
--- a/acts/tests/google/wifi/WifiDppTest.py
+++ b/acts/tests/google/wifi/WifiDppTest.py
@@ -19,14 +19,14 @@
import time
from acts import asserts
-from acts import base_test
from acts import utils
from acts.test_decorators import test_tracker_info
from acts.test_utils.wifi import wifi_constants
from acts.test_utils.wifi import wifi_test_utils as wutils
+from acts.test_utils.wifi.WifiBaseTest import WifiBaseTest
from acts.test_utils.wifi.aware import aware_test_utils as autils
-class WifiDppTest(base_test.BaseTestClass):
+class WifiDppTest(WifiBaseTest):
"""This class tests the DPP API surface.
Attributes: The tests in this class require one DUT and one helper phone
@@ -84,8 +84,8 @@
wutils.reset_wifi(self.dut)
def on_fail(self, test_name, begin_time):
- self.dut.take_bug_report(test_name, begin_time)
- self.dut.cat_adb_log(test_name, begin_time)
+ self.dut.take_bug_report(test_name, begin_time)
+ self.dut.cat_adb_log(test_name, begin_time)
def create_and_save_wifi_network_config(self, security):
""" Create a config with random SSID and password.
@@ -625,6 +625,7 @@
""" Tests Begin """
@test_tracker_info(uuid="30893d51-2069-4e1c-8917-c8a840f91b59")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_5G(self):
asserts.skip_if(not self.dut.droid.wifiIs5GHzBandSupported() or
not self.helper_dev.droid.wifiIs5GHzBandSupported(),
@@ -634,6 +635,7 @@
use_mac=True)
@test_tracker_info(uuid="54d1d19a-aece-459c-b819-9d4b1ae63f77")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_5G_broadcast(self):
asserts.skip_if(not self.dut.droid.wifiIs5GHzBandSupported() or
not self.helper_dev.droid.wifiIs5GHzBandSupported(),
@@ -643,6 +645,7 @@
use_mac=False)
@test_tracker_info(uuid="18270a69-300c-4f54-87fd-c19073a2854e ")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_no_chan_in_uri_listen_on_5745_broadcast(self):
asserts.skip_if(not self.dut.droid.wifiIs5GHzBandSupported() or
not self.helper_dev.droid.wifiIs5GHzBandSupported(),
@@ -651,6 +654,7 @@
security=self.DPP_TEST_SECURITY_PSK, responder_chan=None, responder_freq=5745, use_mac=False)
@test_tracker_info(uuid="fbdd687c-954a-400b-9da3-2d17e28b0798")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_no_chan_in_uri_listen_on_5745(self):
asserts.skip_if(not self.dut.droid.wifiIs5GHzBandSupported() or
not self.helper_dev.droid.wifiIs5GHzBandSupported(),
@@ -658,42 +662,50 @@
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK, responder_chan=None, responder_freq=5745, use_mac=True)
- @test_tracker_info(uuid="570f499f-ab12-4405-af14-c9ed36da2e01 ")
+ @test_tracker_info(uuid="570f499f-ab12-4405-af14-c9ed36da2e01")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_no_chan_in_uri_listen_on_2462_broadcast(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK, responder_chan=None, responder_freq=2462, use_mac=False)
@test_tracker_info(uuid="e1f083e0-0878-4c49-8ac5-d7c6bba24625")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_no_chan_in_uri_listen_on_2462(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK, responder_chan=None, responder_freq=2462, use_mac=True)
@test_tracker_info(uuid="d2a526f5-4269-493d-bd79-4e6d1b7b00f0")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK, use_mac=True)
@test_tracker_info(uuid="6ead218c-222b-45b8-8aad-fe7d883ed631")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_sae(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_SAE, use_mac=True)
@test_tracker_info(uuid="1686adb5-1b3c-4e6d-a969-6b007bdd990d")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_passphrase(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE, use_mac=True)
@test_tracker_info(uuid="3958feb5-1a0c-4487-9741-ac06f04c55a2")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_sae_broadcast(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_SAE, use_mac=False)
@test_tracker_info(uuid="fe6d66f5-73a1-46e9-8f49-73b8f332cc8c")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_passphrase_broadcast(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE, use_mac=False)
@test_tracker_info(uuid="9edd372d-e2f1-4545-8d04-6a1636fcbc4b")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_sae_for_ap(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_SAE,
@@ -701,6 +713,7 @@
net_role=self.DPP_TEST_NETWORK_ROLE_AP)
@test_tracker_info(uuid="e9eec912-d665-4926-beac-859cb13dc17b")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_with_psk_passphrase_for_ap(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
@@ -708,26 +721,31 @@
net_role=self.DPP_TEST_NETWORK_ROLE_AP)
@test_tracker_info(uuid="8055694f-606f-41dd-9826-3ea1e9b007f8")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_enrollee_with_sae(self):
self.start_dpp_as_initiator_enrollee(
security=self.DPP_TEST_SECURITY_SAE, use_mac=True)
@test_tracker_info(uuid="c1e9f605-b5c0-4e53-8a08-1b0087a667fa")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_enrollee_with_psk_passphrase(self):
self.start_dpp_as_initiator_enrollee(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE, use_mac=True)
@test_tracker_info(uuid="1d7f30ad-2f9a-427a-8059-651dc8827ae2")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_enrollee_with_sae_broadcast(self):
self.start_dpp_as_initiator_enrollee(
security=self.DPP_TEST_SECURITY_SAE, use_mac=False)
@test_tracker_info(uuid="0cfc2645-600e-4f2b-ab5c-fcee6d363a9a")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_enrollee_with_psk_passphrase_broadcast(self):
self.start_dpp_as_initiator_enrollee(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE, use_mac=False)
@test_tracker_info(uuid="2e26b248-65dd-41f6-977b-e223d72b2de9")
+ @WifiBaseTest.wifi_test_wrap
def test_start_dpp_as_initiator_enrollee_receive_invalid_config(self):
self.start_dpp_as_initiator_enrollee(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
@@ -735,6 +753,7 @@
invalid_config=True)
@test_tracker_info(uuid="ed189661-d1c1-4626-9f01-3b7bb8a417fe")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_fail_authentication(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
@@ -742,6 +761,7 @@
fail_authentication=True)
@test_tracker_info(uuid="5a8c6587-fbb4-4a27-9cba-af6f8935833a")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_fail_unicast_timeout(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
@@ -749,6 +769,7 @@
cause_timeout=True)
@test_tracker_info(uuid="b12353ac-1a04-4036-81a4-2d2d0c653dbb")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_fail_broadcast_timeout(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
@@ -756,6 +777,7 @@
cause_timeout=True)
@test_tracker_info(uuid="eeff91be-09ce-4a33-8b4f-ece40eb51c76")
+ @WifiBaseTest.wifi_test_wrap
def test_dpp_as_initiator_configurator_invalid_uri(self):
self.start_dpp_as_initiator_configurator(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
@@ -763,6 +785,7 @@
invalid_uri=True)
@test_tracker_info(uuid="1fa25f58-0d0e-40bd-8714-ab78957514d9")
+ @WifiBaseTest.wifi_test_wrap
def test_start_dpp_as_initiator_enrollee_fail_timeout(self):
self.start_dpp_as_initiator_enrollee(
security=self.DPP_TEST_SECURITY_PSK_PASSPHRASE,
diff --git a/acts/tests/google/wifi/WifiTetheringTest.py b/acts/tests/google/wifi/WifiTetheringTest.py
index c968b66..d5d197a 100644
--- a/acts/tests/google/wifi/WifiTetheringTest.py
+++ b/acts/tests/google/wifi/WifiTetheringTest.py
@@ -50,14 +50,23 @@
self.network = {"SSID": "hotspot_%s" % utils.rand_ascii_str(6),
"password": "pass_%s" % utils.rand_ascii_str(6)}
self.new_ssid = "wifi_tethering_test2"
+ self.tcpdump_pid=[]
nutils.verify_lte_data_and_tethering_supported(self.hotspot_device)
for ad in self.tethered_devices:
wutils.wifi_test_device_init(ad)
+ def setup_test(self):
+ for ad in self.android_devices:
+ self.tcpdump_pid.append(nutils.start_tcpdump(ad, self.test_name))
+
def teardown_test(self):
if self.hotspot_device.droid.wifiIsApEnabled():
wutils.stop_wifi_tethering(self.hotspot_device)
+ for ad, pid in zip(self.android_devices, self.tcpdump_pid):
+ nutils.stop_tcpdump(ad, pid, self.test_name)
+ self.tcpdump_pid = []
+
def teardown_class(self):
""" Reset devices """