1#!/usr/bin/env python3 2# 3# Copyright 2019 - Google 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16""" 17 Test Script for Telephony Stress data Test 18""" 19import collections 20import time 21 22from acts.test_decorators import test_tracker_info 23from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest 24from acts_contrib.test_utils.tel.tel_test_utils import iperf_test_by_adb 25from acts_contrib.test_utils.tel.tel_test_utils import iperf_udp_test_by_adb 26from acts.logger import epoch_to_log_line_timestamp 27from acts.utils import get_current_epoch_time 28 29 30class TelLiveStressDataTest(TelephonyBaseTest): 31 def setup_class(self): 32 super().setup_class() 33 self.ad = self.android_devices[0] 34 self.iperf_server_address = self.user_params.get("iperf_server", 35 '0.0.0.0') 36 self.iperf_tcp_port = int( 37 self.user_params.get("iperf_tcp_port", 0)) 38 self.iperf_udp_port = int( 39 self.user_params.get("iperf_udp_port", 0)) 40 self.iperf_duration = int( 41 self.user_params.get("iperf_duration", 60)) 42 self.iperf_iteration = int( 43 self.user_params.get("iperf_iteration", 10)) 44 self.sleep_time_between_iperf_iterations = int( 45 self.user_params.get("sleep_time_between_iperf_iterations", 2)) 46 47 def stress_test_upload(self, test_tcp=True): 48 """Start the upload iperf stress test. 49 50 Args: 51 test_tcp: True for using TCP, using UDP otherwise. 52 53 Returns: 54 True if success, False if fail. 55 """ 56 fail_count = collections.defaultdict(int) 57 for i in range(1, self.iperf_iteration + 1): 58 msg = "Stress Throughput Test %s Iteration: <%s> / <%s>" % ( 59 self.test_name, i, self.iperf_iteration) 60 begin_time = get_current_epoch_time() 61 self.log.info(msg) 62 iteration_result = True 63 if test_tcp: 64 if not iperf_test_by_adb(self.log, 65 self.ad, 66 self.iperf_server_address, 67 self.iperf_tcp_port, 68 False, 69 self.iperf_duration): 70 fail_count["upload"] += 1 71 iteration_result = False 72 self.log.error("%s upload failure.", msg) 73 else: 74 if not iperf_udp_test_by_adb(self.log, 75 self.ad, 76 self.iperf_server_address, 77 self.iperf_udp_port, 78 False, 79 self.iperf_duration): 80 fail_count["upload"] += 1 81 iteration_result = False 82 self.log.error("%s upload failure.", msg) 83 84 self.log.info("%s %s", msg, iteration_result) 85 if not iteration_result: 86 self._take_bug_report("%s_UploadNo_%s" % (self.test_name, i), 87 begin_time) 88 89 if self.sleep_time_between_iperf_iterations: 90 self.ad.droid.goToSleepNow() 91 time.sleep(self.sleep_time_between_iperf_iterations) 92 93 test_result = True 94 for failure, count in fail_count.items(): 95 if count: 96 self.log.error("%s: %s %s failures in %s iterations", 97 self.test_name, count, failure, 98 self.iperf_iteration) 99 test_result = False 100 return test_result 101 102 def stress_test_download(self, test_tcp=True): 103 """Start the download iperf stress test. 104 105 Args: 106 test_tcp: True for using TCP, using UDP otherwise. 107 108 Returns: 109 True if success, False if fail. 110 """ 111 fail_count = collections.defaultdict(int) 112 for i in range(1, self.iperf_iteration + 1): 113 msg = "Stress Throughput Test %s Iteration: <%s> / <%s>" % ( 114 self.test_name, i, self.iperf_iteration) 115 begin_time = get_current_epoch_time() 116 self.log.info(msg) 117 iteration_result = True 118 if test_tcp: 119 if not iperf_test_by_adb(self.log, 120 self.ad, 121 self.iperf_server_address, 122 self.iperf_tcp_port, 123 True, 124 self.iperf_duration): 125 fail_count["download"] += 1 126 iteration_result = False 127 self.log.error("%s download failure.", msg) 128 else: 129 if not iperf_udp_test_by_adb(self.log, 130 self.ad, 131 self.iperf_server_address, 132 self.iperf_udp_port, 133 True, 134 self.iperf_duration): 135 fail_count["download"] += 1 136 iteration_result = False 137 self.log.error("%s download failure.", msg) 138 139 self.log.info("%s %s", msg, iteration_result) 140 if not iteration_result: 141 self._take_bug_report("%s_DownloadNo_%s" % (self.test_name, i), 142 begin_time) 143 144 if self.sleep_time_between_iperf_iterations: 145 self.ad.droid.goToSleepNow() 146 time.sleep(self.sleep_time_between_iperf_iterations) 147 148 test_result = True 149 for failure, count in fail_count.items(): 150 if count: 151 self.log.error("%s: %s %s failures in %s iterations", 152 self.test_name, count, failure, 153 self.iperf_iteration) 154 test_result = False 155 return test_result 156 157 @test_tracker_info(uuid="190fdeb1-541e-455f-9f37-762a8e55c07f") 158 @TelephonyBaseTest.tel_test_wrap 159 def test_tcp_upload_stress(self): 160 return iperf_test_by_adb(self.log, 161 self.ad, 162 self.iperf_server_address, 163 self.iperf_tcp_port, 164 False, 165 self.iperf_duration) 166 167 @test_tracker_info(uuid="af9805f8-6ed5-4e05-823e-d88dcef45637") 168 @TelephonyBaseTest.tel_test_wrap 169 def test_tcp_download_stress(self): 170 return iperf_test_by_adb(self.log, 171 self.ad, 172 self.iperf_server_address, 173 self.iperf_tcp_port, 174 True, 175 self.iperf_duration) 176 177 @test_tracker_info(uuid="55bf5e09-dc7b-40bc-843f-31fed076ffe4") 178 @TelephonyBaseTest.tel_test_wrap 179 def test_udp_upload_stress(self): 180 return iperf_udp_test_by_adb(self.log, 181 self.ad, 182 self.iperf_server_address, 183 self.iperf_udp_port, 184 False, 185 self.iperf_duration) 186 187 @test_tracker_info(uuid="02ae88b2-d597-45df-ab5a-d701d1125a0f") 188 @TelephonyBaseTest.tel_test_wrap 189 def test_udp_download_stress(self): 190 return iperf_udp_test_by_adb(self.log, 191 self.ad, 192 self.iperf_server_address, 193 self.iperf_udp_port, 194 True, 195 self.iperf_duration) 196 197 @test_tracker_info(uuid="79aaa7ec-5046-4ffe-b27a-ca93e404e9e0") 198 @TelephonyBaseTest.tel_test_wrap 199 def test_tcp_upload_data_stress(self): 200 return self.stress_test_upload() 201 202 @test_tracker_info(uuid="6a1e5032-9498-4d23-8ae9-db36f1a238c1") 203 @TelephonyBaseTest.tel_test_wrap 204 def test_tcp_download_data_stress(self): 205 return self.stress_test_download() 206 207 @test_tracker_info(uuid="22400c16-dbbb-41c9-afd0-86b525a0bcee") 208 @TelephonyBaseTest.tel_test_wrap 209 def test_udp_upload_data_stress(self): 210 return self.stress_test_upload(test_tcp=False) 211 212 @test_tracker_info(uuid="9f3b2818-5265-422e-9e6f-9ee08dfcc696") 213 @TelephonyBaseTest.tel_test_wrap 214 def test_udp_download_data_stress(self): 215 return self.stress_test_download(test_tcp=False)