| #!/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) |
