#!/usr/bin/env python3 # # Copyright 2019 - Google # # 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. """ Test Script for Telephony Stress data Test """ import collections import time from acts.test_decorators import test_tracker_info from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest from acts_contrib.test_utils.tel.tel_test_utils import iperf_test_by_adb from acts_contrib.test_utils.tel.tel_test_utils import iperf_udp_test_by_adb from acts.logger import epoch_to_log_line_timestamp from acts.utils import get_current_epoch_time class TelLiveStressDataTest(TelephonyBaseTest): def setup_class(self): super().setup_class() self.ad = self.android_devices[0] self.iperf_server_address = self.user_params.get("iperf_server", '0.0.0.0') self.iperf_tcp_port = int( self.user_params.get("iperf_tcp_port", 0)) self.iperf_udp_port = int( self.user_params.get("iperf_udp_port", 0)) self.iperf_duration = int( self.user_params.get("iperf_duration", 60)) self.iperf_iteration = int( self.user_params.get("iperf_iteration", 10)) self.sleep_time_between_iperf_iterations = int( self.user_params.get("sleep_time_between_iperf_iterations", 2)) def stress_test_upload(self, test_tcp=True): """Start the upload iperf stress test. Args: test_tcp: True for using TCP, using UDP otherwise. Returns: True if success, False if fail. """ fail_count = collections.defaultdict(int) for i in range(1, self.iperf_iteration + 1): msg = "Stress Throughput Test %s Iteration: <%s> / <%s>" % ( self.test_name, i, self.iperf_iteration) begin_time = get_current_epoch_time() self.log.info(msg) iteration_result = True if test_tcp: if not iperf_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_tcp_port, False, self.iperf_duration): fail_count["upload"] += 1 iteration_result = False self.log.error("%s upload failure.", msg) else: if not iperf_udp_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_udp_port, False, self.iperf_duration): fail_count["upload"] += 1 iteration_result = False self.log.error("%s upload failure.", msg) self.log.info("%s %s", msg, iteration_result) if not iteration_result: self._take_bug_report("%s_UploadNo_%s" % (self.test_name, i), begin_time) if self.sleep_time_between_iperf_iterations: self.ad.droid.goToSleepNow() time.sleep(self.sleep_time_between_iperf_iterations) test_result = True for failure, count in fail_count.items(): if count: self.log.error("%s: %s %s failures in %s iterations", self.test_name, count, failure, self.iperf_iteration) test_result = False return test_result def stress_test_download(self, test_tcp=True): """Start the download iperf stress test. Args: test_tcp: True for using TCP, using UDP otherwise. Returns: True if success, False if fail. """ fail_count = collections.defaultdict(int) for i in range(1, self.iperf_iteration + 1): msg = "Stress Throughput Test %s Iteration: <%s> / <%s>" % ( self.test_name, i, self.iperf_iteration) begin_time = get_current_epoch_time() self.log.info(msg) iteration_result = True if test_tcp: if not iperf_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_tcp_port, True, self.iperf_duration): fail_count["download"] += 1 iteration_result = False self.log.error("%s download failure.", msg) else: if not iperf_udp_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_udp_port, True, self.iperf_duration): fail_count["download"] += 1 iteration_result = False self.log.error("%s download failure.", msg) self.log.info("%s %s", msg, iteration_result) if not iteration_result: self._take_bug_report("%s_DownloadNo_%s" % (self.test_name, i), begin_time) if self.sleep_time_between_iperf_iterations: self.ad.droid.goToSleepNow() time.sleep(self.sleep_time_between_iperf_iterations) test_result = True for failure, count in fail_count.items(): if count: self.log.error("%s: %s %s failures in %s iterations", self.test_name, count, failure, self.iperf_iteration) test_result = False return test_result @test_tracker_info(uuid="190fdeb1-541e-455f-9f37-762a8e55c07f") @TelephonyBaseTest.tel_test_wrap def test_tcp_upload_stress(self): return iperf_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_tcp_port, False, self.iperf_duration) @test_tracker_info(uuid="af9805f8-6ed5-4e05-823e-d88dcef45637") @TelephonyBaseTest.tel_test_wrap def test_tcp_download_stress(self): return iperf_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_tcp_port, True, self.iperf_duration) @test_tracker_info(uuid="55bf5e09-dc7b-40bc-843f-31fed076ffe4") @TelephonyBaseTest.tel_test_wrap def test_udp_upload_stress(self): return iperf_udp_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_udp_port, False, self.iperf_duration) @test_tracker_info(uuid="02ae88b2-d597-45df-ab5a-d701d1125a0f") @TelephonyBaseTest.tel_test_wrap def test_udp_download_stress(self): return iperf_udp_test_by_adb(self.log, self.ad, self.iperf_server_address, self.iperf_udp_port, True, self.iperf_duration) @test_tracker_info(uuid="79aaa7ec-5046-4ffe-b27a-ca93e404e9e0") @TelephonyBaseTest.tel_test_wrap def test_tcp_upload_data_stress(self): return self.stress_test_upload() @test_tracker_info(uuid="6a1e5032-9498-4d23-8ae9-db36f1a238c1") @TelephonyBaseTest.tel_test_wrap def test_tcp_download_data_stress(self): return self.stress_test_download() @test_tracker_info(uuid="22400c16-dbbb-41c9-afd0-86b525a0bcee") @TelephonyBaseTest.tel_test_wrap def test_udp_upload_data_stress(self): return self.stress_test_upload(test_tcp=False) @test_tracker_info(uuid="9f3b2818-5265-422e-9e6f-9ee08dfcc696") @TelephonyBaseTest.tel_test_wrap def test_udp_download_data_stress(self): return self.stress_test_download(test_tcp=False)