1#!/usr/bin/env python3 2# 3# Copyright 2019 - The Android Open Source Project 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 17import re 18import time 19 20from acts import base_test 21from acts.libs.proc import job 22from acts import signals 23from acts.test_decorators import test_tracker_info 24 25# Time it takes for the usb tethering IP to show up in ifconfig. 26IFCONFIG_SETTLE_TIME = 5 27USB_CHARGE_MODE = 'svc usb setFunctions' 28USB_TETHERING_MODE = 'svc usb setFunctions rndis' 29DEVICE_IP_ADDRESS = 'ip address' 30 31class UsbTetheringThroughputTest(base_test.BaseTestClass): 32 """Tests for usb tethering throughput test. 33 34 Test Bed Requirement: 35 * One Android device. 36 * Sim card with data call. 37 """ 38 39 def setup_class(self): 40 """ Setup devices for usb tethering """ 41 self.dut = self.android_devices[0] 42 self.ip_server = self.iperf_servers[0] 43 self.port_num = self.ip_server.port 44 req_params = [ 45 'iperf_criteria', 46 ] 47 self.unpack_userparams(req_param_names=req_params) 48 49 def setup_test(self): 50 self.dut.droid.wakeLockAcquireBright() 51 self.dut.droid.wakeUpNow() 52 self.dut.unlock_screen() 53 54 def teardown_test(self): 55 self.dut.droid.wakeLockRelease() 56 self.dut.droid.goToSleepNow() 57 self.dut.stop_services() 58 #Set usb function back to charge mode. 59 self.dut.adb.shell(USB_CHARGE_MODE) 60 self.dut.adb.wait_for_device() 61 self.dut.start_services() 62 self.ip_server.stop() 63 64 def on_fail(self, test_name, begin_time): 65 self.dut.take_bug_report(test_name, begin_time) 66 self.dut.cat_adb_log(test_name, begin_time) 67 68 def enable_usb_tethering(self): 69 """Stop SL4A service and enable usb tethering. 70 71 Logic steps are 72 1. Stop SL4A service. 73 2. Enable usb tethering. 74 3. Restart SL4A service. 75 4. Check usb tethering enabled. 76 77 Raises: 78 Signals.TestFailure is raised by not find rndis string. 79 """ 80 self.dut.stop_services() 81 self.dut.adb.shell(USB_TETHERING_MODE, ignore_status=True) 82 self.dut.adb.wait_for_device() 83 self.dut.start_services() 84 if 'rndis' not in self.dut.adb.shell(DEVICE_IP_ADDRESS): 85 raise signals.TestFailure('Unable to enable USB tethering.') 86 87 def get_pc_tethering_ip(self): 88 """Check usb tethering IP from PC. 89 90 Returns: 91 Usb tethering IP from PC. 92 93 Raises: 94 Signals.TestFailure is raised by not find usb tethering IP. 95 """ 96 time.sleep(IFCONFIG_SETTLE_TIME) 97 check_usb_tethering = job.run('ifconfig').stdout 98 matches = re.findall('inet addr:(\d+.\d+.42.\d+)', check_usb_tethering) 99 if not matches: 100 raise signals.TestFailure( 101 'Unable to find tethering IP. The device may not be tethered.') 102 else: 103 return matches[0] 104 105 def get_dut_tethering_ip(self): 106 """Check usb tethering IP from Android device. 107 108 Returns: 109 Usb tethering IP from Android device. 110 111 Raises: 112 Signals.TestFailure is raised by not find usb tethering IP. 113 """ 114 time.sleep(IFCONFIG_SETTLE_TIME) 115 check_usb_tethering = self.dut.adb.shell('ifconfig') 116 matches = re.findall('addr:(\d+.\d+.42.\d+)', check_usb_tethering) 117 if not matches: 118 raise signals.TestFailure( 119 'Unable to find tethering IP. The device may not be tethered.') 120 else: 121 return matches[0] 122 123 def pc_can_ping(self, count=3): 124 """Run ping traffic from PC side. 125 126 Logic steps are 127 1. PC ping the usb server ip. 128 2. Check the packet loss rate. 129 130 Args: 131 count : count for ping test. 132 133 Returns: 134 True: If no packet loss. 135 False: Otherwise. 136 """ 137 ip = self.get_dut_tethering_ip() 138 ping = job.run( 139 'ping -c {} {}'.format(count, ip), ignore_status=True).stdout 140 self.log.info(ping) 141 return '0% packet loss' in ping 142 143 def dut_can_ping(self, count=3): 144 """Run ping traffic from Android device side. 145 146 Logic steps are 147 1. Android device ping the 8.8.8.8. 148 2. Check the packet loss rate. 149 150 Args: 151 count : count for ping test. 152 153 Returns: 154 True: If no packet loss. 155 False: Otherwise. 156 """ 157 ip = '8.8.8.8' 158 ping = self.dut.adb.shell( 159 'ping -c {} {}'.format(count, ip), ignore_status=True) 160 self.log.info(ping) 161 return '0% packet loss' in ping 162 163 def run_iperf_client(self, dut_ip, extra_params=''): 164 """Run iperf client command after iperf server enabled. 165 166 Args: 167 dut_ip: string which contains the ip address. 168 extra_params: params to be added to the iperf client. 169 170 Returns: 171 Success: True if the iperf execute. False otherwise. 172 Rate: throughput data. 173 """ 174 self.log.info('Run iperf process.') 175 cmd = "iperf3 -c {} {} -p {}".format(dut_ip, extra_params, 176 self.port_num) 177 self.log.info(cmd) 178 try: 179 result = self.dut.adb.shell(cmd, ignore_status=True) 180 self.log.info(result) 181 except: 182 self.log.error('Fail to execute iperf client.') 183 return False, 0 184 rate = re.findall('(\d+.\d+) Mbits/sec.*receiver', result) 185 return True, rate[0] 186 187 @test_tracker_info(uuid="e7e0dfdc-3d1c-4642-a468-27326c49e4cb") 188 def test_tethering_ping(self): 189 """Enable usb tethering then executing ping test. 190 191 Steps: 192 1. Stop SL4A service. 193 2. Enable usb tethering. 194 3. Restart SL4A service. 195 4. Execute ping test from PC and Android Device. 196 5. Check the ping lost rate. 197 """ 198 self.enable_usb_tethering() 199 if self.pc_can_ping() and self.dut_can_ping(): 200 raise signals.TestPass('Ping test is passed. Network is reachable.') 201 raise signals.TestFailure( 202 'Ping test failed. Maybe network is unreachable.') 203 204 @test_tracker_info(uuid="8263c880-8a7e-4a68-b47f-e7caba3e9968") 205 def test_usb_tethering_iperf(self): 206 """Enable usb tethering then executing iperf test. 207 208 Steps: 209 1. Stop SL4A service. 210 2. Enable usb tethering. 211 3. Restart SL4A service. 212 4. Execute iperf test for usb tethering and get the throughput result. 213 5. Check the iperf throughput result. 214 """ 215 self.enable_usb_tethering() 216 self.ip_server.start() 217 pc_ip = self.get_pc_tethering_ip() 218 tx_success, tx_rate = self.run_iperf_client(pc_ip, '-t5 -i1 -w2M') 219 rx_success, rx_rate = self.run_iperf_client(pc_ip, '-t5 -i1 -w2M -R') 220 self.log.info('Iperf rx result: ' + rx_rate + ' Mbits/sec') 221 self.log.info('Iperf tx result: ' + tx_rate + ' Mbits/sec') 222 self.ip_server.stop() 223 if not tx_success or float(tx_rate) < self.iperf_criteria: 224 raise signals.TestFailure( 225 'Iperf tx test is {} Mbits/sec, ' 226 'the throughput result failed.'.format(tx_rate)) 227 if not rx_success or float(rx_rate) < self.iperf_criteria: 228 raise signals.TestFailure( 229 'Iperf rx test is {} Mbits/sec, ' 230 'the throughput result failed.'.format(rx_rate)) 231