#!/usr/bin/env python3 # # Copyright 2019 - 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. import re import time from acts import base_test from acts.libs.proc import job from acts import signals from acts.test_decorators import test_tracker_info # Time it takes for the usb tethering IP to show up in ifconfig. IFCONFIG_SETTLE_TIME = 5 USB_CHARGE_MODE = 'svc usb setFunctions' USB_TETHERING_MODE = 'svc usb setFunctions rndis' DEVICE_IP_ADDRESS = 'ip address' class UsbTetheringThroughputTest(base_test.BaseTestClass): """Tests for usb tethering throughput test. Test Bed Requirement: * One Android device. * Sim card with data call. """ def setup_class(self): """ Setup devices for usb tethering """ self.dut = self.android_devices[0] self.ip_server = self.iperf_servers[0] self.port_num = self.ip_server.port req_params = [ 'iperf_usb_3_criteria', 'iperf_usb_2_criteria', 'controller_path', 'iperf_test_sec' ] self.unpack_userparams(req_param_names=req_params) if self.dut.model in self.user_params.get('legacy_projects', []): self.usb_controller_path = self.user_params[ 'legacy_controller_path'] else: self.usb_controller_path = self.controller_path def setup_test(self): self.dut.adb.wait_for_device() self.dut.droid.wakeLockAcquireBright() self.dut.droid.wakeUpNow() self.dut.unlock_screen() def teardown_test(self): self.dut.droid.wakeLockRelease() self.dut.droid.goToSleepNow() # Reboot device to avoid the side effect that cause adb missing after echo usb speed. self.dut.reboot() 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) def enable_usb_tethering(self): """Stop SL4A service and enable usb tethering. Logic steps are 1. Stop SL4A service. 2. Enable usb tethering. 3. Restart SL4A service. 4. Check usb tethering enabled. Raises: Signals.TestFailure is raised by not find rndis string. """ self.dut.stop_services() self.dut.adb.shell(USB_TETHERING_MODE, ignore_status=True) self.dut.adb.wait_for_device() self.dut.start_services() if 'rndis' not in self.dut.adb.shell(DEVICE_IP_ADDRESS): raise signals.TestFailure('Unable to enable USB tethering.') def get_pc_tethering_ip(self): """Check usb tethering IP from PC. Returns: Usb tethering IP from PC. Raises: Signals.TestFailure is raised by not find usb tethering IP. """ time.sleep(IFCONFIG_SETTLE_TIME) check_usb_tethering = job.run('ifconfig').stdout matches = re.findall('inet (\d+.\d+.42.\d+)', check_usb_tethering) if not matches: raise signals.TestFailure( 'Unable to find tethering IP. The device may not be tethered.') else: return matches[0] def get_dut_tethering_ip(self): """Check usb tethering IP from Android device. Returns: Usb tethering IP from Android device. Raises: Signals.TestFailure is raised by not find usb tethering IP. """ time.sleep(IFCONFIG_SETTLE_TIME) check_usb_tethering = self.dut.adb.shell('ifconfig') matches = re.findall('addr:(\d+.\d+.42.\d+)', check_usb_tethering) if not matches: raise signals.TestFailure( 'Unable to find tethering IP. The device may not be tethered.') else: return matches[0] def pc_can_ping(self, count=3): """Run ping traffic from PC side. Logic steps are 1. PC ping the usb server ip. 2. Check the packet loss rate. Args: count : count for ping test. Returns: True: If no packet loss. False: Otherwise. """ ip = self.get_dut_tethering_ip() ping = job.run( 'ping -c {} {}'.format(count, ip), ignore_status=True).stdout self.log.info(ping) return '0% packet loss' in ping def dut_can_ping(self, count=3): """Run ping traffic from Android device side. Logic steps are 1. Android device ping the 8.8.8.8. 2. Check the packet loss rate. Args: count : count for ping test. Returns: True: If no packet loss. False: Otherwise. """ ip = '8.8.8.8' ping = self.dut.adb.shell( 'ping -c {} {}'.format(count, ip), ignore_status=True) self.log.info(ping) return '0% packet loss' in ping def run_iperf_client(self, dut_ip, extra_params=''): """Run iperf client command after iperf server enabled. Args: dut_ip: string which contains the ip address. extra_params: params to be added to the iperf client. Returns: Success: True if the iperf execute. False otherwise. Rate: throughput data. """ self.log.info('Run iperf process.') cmd = "iperf3 -c {} {} -p {}".format(dut_ip, extra_params, self.port_num) self.log.info(cmd) try: result = self.dut.adb.shell(cmd, ignore_status=True) self.log.info(result) except: self.log.error('Fail to execute iperf client.') return False, 0 rate = re.findall('(\d+.\d+) (.)bits/sec.*receiver', result) return True, rate[0][0], rate[0][1] def run_iperf_tx_rx(self, usb_speed, criteria): """Run iperf tx and rx. Args: usb_speed: string which contains the speed info. criteria: usb performance criteria. Raises: Signals.TestFailure is raised by usb tethering under criteria. """ self.enable_usb_tethering() self.dut.adb.wait_for_device() self.ip_server.start() pc_ip = self.get_pc_tethering_ip() tx_success, tx_rate, tx_unit = self.run_iperf_client( pc_ip, '-t{} -i1 -w2M'.format(self.iperf_test_sec)) rx_success, rx_rate, rx_unit = self.run_iperf_client( pc_ip, '-t{} -i1 -w2M -R'.format(self.iperf_test_sec)) self.log.info('TestResult iperf_{}_rx'.format(usb_speed) + rx_rate + ' ' + rx_unit + 'bits/sec') self.log.info('TestResult iperf_{}_tx'.format(usb_speed) + tx_rate + ' ' + tx_unit + 'bits/sec') self.ip_server.stop() if not tx_success or (float(tx_rate) < criteria and tx_unit != "G"): raise signals.TestFailure('Iperf {}_tx test is {} {}bits/sec, ' 'the throughput result failed.'.format( usb_speed, tx_rate, tx_unit)) if not rx_success or (float(rx_rate) < criteria and rx_unit != "G"): raise signals.TestFailure('Iperf {}_rx test is {} {}bits/sec, ' 'the throughput result failed.'.format( usb_speed, rx_rate, rx_unit)) def get_usb_speed(self): """Get current usb speed.""" usb_controller_address = self.dut.adb.shell( 'getprop sys.usb.controller', ignore_status=True) usb_speed = self.dut.adb.shell( 'cat sys/class/udc/{}/current_speed'.format( usb_controller_address)) return usb_speed, usb_controller_address @test_tracker_info(uuid="e7e0dfdc-3d1c-4642-a468-27326c49e4cb") def test_tethering_ping(self): """Enable usb tethering then executing ping test. Steps: 1. Stop SL4A service. 2. Enable usb tethering. 3. Restart SL4A service. 4. Execute ping test from PC and Android Device. 5. Check the ping lost rate. """ self.enable_usb_tethering() if self.pc_can_ping() and self.dut_can_ping(): raise signals.TestPass( 'Ping test is passed. Network is reachable.') raise signals.TestFailure( 'Ping test failed. Maybe network is unreachable.') @test_tracker_info(uuid="8263c880-8a7e-4a68-b47f-e7caba3e9968") def test_usb_tethering_iperf_super_speed(self): """Enable usb tethering then executing iperf test. Steps: 1. Stop SL4A service. 2. Enable usb tethering. 3. Restart SL4A service. 4. Skip test if device not support super-speed. 5. Execute iperf test for usb tethering and get the throughput result. 6. Check the iperf throughput result. """ if (self.get_usb_speed()[0] != 'super-speed'): raise signals.TestSkip( 'USB 3 not available for the device, skip super-speed performance test.' ) self.run_iperf_tx_rx('usb_3', self.iperf_usb_3_criteria) @test_tracker_info(uuid="5d8a22fd-1f9b-4758-a6b4-855d134b348a") def test_usb_tethering_iperf_high_speed(self): """Enable usb tethering then executing iperf test. Steps: 1. Stop SL4A service. 2. Enable usb tethering. 3. Restart SL4A service. 4. Force set usb speed to high-speed if default is super-speed. 5. Execute iperf test for usb tethering and get the throughput result. 6. Check the iperf throughput result. """ if (self.get_usb_speed()[0] != 'high-speed'): self.log.info( 'Default usb speed is USB 3,Force set usb to high-speed.') self.dut.stop_services() self.dut.adb.wait_for_device() self.dut.adb.shell( 'echo high > {}{}.ssusb/speed'.format( self.usb_controller_path, self.get_usb_speed()[1].strip('.dwc3')), ignore_status=True) self.dut.adb.wait_for_device() self.dut.start_services() self.run_iperf_tx_rx('usb_2', self.iperf_usb_2_criteria)