#!/usr/bin/env python3.4 # # Copyright 2018 - 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 time import scapy.all as scapy from acts import asserts from acts.test_utils.power import IperfHelper as IPH from acts.test_utils.power import PowerCellularLabBaseTest as PWCEL from acts.test_utils.wifi import wifi_power_test_utils as wputils class PowerTelTrafficTest(PWCEL.PowerCellularLabBaseTest): """ Cellular traffic power test. Inherits from PowerCellularLabBaseTest. Parses config specific to this kind of test. Contains methods to start data traffic between a local instance of iPerf and one running in the dut. """ # Keywords for test name parameters PARAM_DIRECTION = 'direction' PARAM_DIRECTION_UL = 'ul' PARAM_DIRECTION_DL = 'dl' PARAM_DIRECTION_DL_UL = 'dlul' PARAM_BANDWIDTH_LIMIT = 'blimit' # Iperf waiting time IPERF_MARGIN = 10 # Constant used to calculate the tcp window size from total throughput TCP_WINDOW_FRACTION = 40 def __init__(self, controllers): """ Class initialization. Sets test parameters to initial values. """ super().__init__(controllers) # These variables are passed to iPerf when starting data # traffic with the -b parameter to limit throughput on # the application layer. self.bandwidth_limit_dl = None self.bandwidth_limit_ul = None # Throughput obtained from iPerf self.iperf_results = None def setup_test(self): """ Executed before every test case. Parses test configuration from the test name and prepares the simulation for measurement. """ # Call parent method first to setup simulation if not super().setup_test(): return False # Traffic direction values = self.consume_parameter(self.PARAM_DIRECTION, 1) if values: self.traffic_direction = values[1] else: self.log.error("The test name has to include parameter {} " "followed by {}/{}/{}.".format( self.PARAM_DIRECTION, self.PARAM_DIRECTION_UL, self.PARAM_DIRECTION_DL, self.PARAM_DIRECTION_DL_UL)) return False # Bandwidth limit values = self.consume_parameter(self.PARAM_BANDWIDTH_LIMIT, 2) if values: self.bandwidth_limit_dl = values[1] self.bandwidth_limit_ul = values[2] else: self.bandwidth_limit_dl = 0 self.bandwidth_limit_ul = 0 self.log.error( "No bandwidth limit was indicated in the test parameters. " "Setting to default value of 0 (no limit to bandwidth). To set " "a different value include parameter '{}' followed by two " "strings indicating downlink and uplink bandwidth limits for " "iPerf.".format(self.PARAM_BANDWIDTH_LIMIT)) # No errors when parsing parameters return True def teardown_test(self): """Tear down necessary objects after test case is finished. """ for ips in self.iperf_servers: ips.stop() def power_tel_traffic_test(self): """ Measures power and throughput during data transmission. Measurement step in this test. Starts iPerf client in the DUT and then initiates power measurement. After that, DUT is connected again and the result from iPerf is collected. Pass or fail is decided with a threshold value. """ # Start data traffic iperf_helpers = self.start_tel_traffic(self.dut) # Measure power self.collect_power_data() # Wait for iPerf to finish time.sleep(self.IPERF_MARGIN + 2) # Collect throughput measurement self.iperf_results = self.get_iperf_results(self.dut, iperf_helpers) # Check if power measurement is below the required value self.pass_fail_check() return self.test_result, self.iperf_results def get_iperf_results(self, device, iperf_helpers): """ Pulls iperf results from the device. Args: device: the device from which iperf results need to be pulled. Returns: a dictionary containing DL/UL throughput in Mbit/s. """ throughput = {} for iph in iperf_helpers: self.log.info("Getting {} throughput results.".format( iph.traffic_direction)) iperf_result = iph.process_iperf_results( device, self.log, self.iperf_servers, self.test_name) throughput[iph.traffic_direction] = iperf_result return throughput def pass_fail_check(self): """ Checks power consumption and throughput. Uses the base class method to check power consumption. Also, compares the obtained throughput with the expected value provided by the simulation class. """ for direction, throughput in self.iperf_results.items(): try: if direction == "UL": expected_t = self.simulation.maximum_uplink_throughput() elif direction == "DL": expected_t = self.simulation.maximum_downlink_throughput() else: raise RuntimeError("Unexpected traffic direction value.") except NotImplementedError: # Some simulation classes might not have implemented the max # throughput calculation yet. self.log.debug("Expected throughput is not available for the " "current simulation class.") else: self.log.info( "The expected {} throughput is {} Mbit/s.".format( direction, expected_t)) asserts.assert_true( 0.90 < throughput / expected_t < 1.10, "{} throughput differed more than 10% from the expected " "value! ({}/{} = {})".format(direction, round(throughput, 3), round(expected_t, 3), round(throughput / expected_t, 3))) super().pass_fail_check() def start_tel_traffic(self, client_host): """ Starts iPerf in the indicated device and initiates traffic. Starts the required iperf clients and servers according to the traffic pattern config in the current test. Args: client_host: device handler in which to start the iperf client. Returns: A list of iperf helpers. """ # The iPerf server is hosted in this computer self.iperf_server_address = scapy.get_if_addr( self.pkt_sender.interface) # Start iPerf traffic iperf_helpers = [] # Calculate TCP windows as a fraction of the expected throughput # Some simulation classes don't implement this method yed try: dl_tcp_window = (self.simulation.maximum_downlink_throughput() / self.TCP_WINDOW_FRACTION) ul_tcp_window = (self.simulation.maximum_uplink_throughput() / self.TCP_WINDOW_FRACTION) except NotImplementedError: dl_tcp_window = None ul_tcp_window = None if self.traffic_direction in [ self.PARAM_DIRECTION_DL, self.PARAM_DIRECTION_DL_UL ]: # Downlink traffic iperf_helpers.append( self.start_iperf_traffic( client_host, server_idx=len(iperf_helpers), traffic_direction='DL', window=dl_tcp_window, bandwidth=self.bandwidth_limit_dl)) if self.traffic_direction in [ self.PARAM_DIRECTION_UL, self.PARAM_DIRECTION_DL_UL ]: # Uplink traffic iperf_helpers.append( self.start_iperf_traffic( client_host, server_idx=len(iperf_helpers), traffic_direction='UL', window=ul_tcp_window, bandwidth=self.bandwidth_limit_ul)) return iperf_helpers def start_iperf_traffic(self, client_host, server_idx, traffic_direction, bandwidth=0, window=None): """Starts iPerf data traffic. Starts an iperf client in an android device and a server locally. Args: client_host: device handler in which to start the iperf client server_idx: id of the iperf server to connect to traffic_direction: has to be either 'UL' or 'DL' bandwidth: bandwidth limit for data traffic window: the tcp window. if None, no window will be passed to iperf Returns: An IperfHelper object for the started client/server pair. """ config = { 'traffic_type': 'TCP', 'duration': self.mon_duration + self.mon_offset + self.IPERF_MARGIN, 'start_meas_time': 4, 'server_idx': server_idx, 'port': self.iperf_servers[server_idx].port, 'traffic_direction': traffic_direction, 'window': window } # If bandwidth is equal to zero then no bandwith requirements are set if bandwidth > 0: config['bandwidth'] = bandwidth iph = IPH.IperfHelper(config) # Start the server locally self.iperf_servers[server_idx].start() # Start the client in the android device wputils.run_iperf_client_nonblocking( client_host, self.iperf_server_address, iph.iperf_args) return iph class PowerTelRvRTest(PowerTelTrafficTest): """ Gets Range vs Rate curves while measuring power consumption. Uses PowerTelTrafficTest as a base class. """ # Test name configuration keywords PARAM_SWEEP = "sweep" PARAM_SWEEP_UPLINK = "uplink" PARAM_SWEEP_DOWNLINK = "downlink" # Sweep values. Need to be set before starting test by test # function or child class. downlink_power_sweep = None uplink_power_sweep = None def setup_test(self): """ Executed before every test case. Parses test configuration from the test name and prepares the simulation for measurement. """ # Call parent method first to setup simulation if not super().setup_test(): return False # Get which power value to sweep from config try: values = self.consume_parameter(self.PARAM_SWEEP, 1) if values[1] == self.PARAM_SWEEP_UPLINK: self.sweep = self.PARAM_SWEEP_UPLINK elif values[1] == self.PARAM_SWEEP_DOWNLINK: self.sweep = self.PARAM_SWEEP_DOWNLINK else: raise ValueError() except: self.log.error( "The test name has to include parameter {} followed by " "either {} or {}.".format(self.PARAM_SWEEP, self.PARAM_SWEEP_DOWNLINK, self.PARAM_SWEEP_UPLINK)) return False return True def power_tel_rvr_test(self): """ Main function for the RvR test. Produces the RvR curve according to the indicated sweep values. """ if self.sweep == self.PARAM_SWEEP_DOWNLINK: sweep_range = self.downlink_power_sweep elif self.sweep == self.PARAM_SWEEP_UPLINK: sweep_range = self.uplink_power_sweep current = [] throughput = [] for pw in sweep_range: if self.sweep == self.PARAM_SWEEP_DOWNLINK: self.simulation.set_downlink_rx_power(self.simulation.bts1, pw) elif self.sweep == self.PARAM_SWEEP_UPLINK: self.simulation.set_uplink_tx_power(self.simulation.bts1, pw) i, t = self.power_tel_traffic_test() self.log.info("---------------------") self.log.info("{} -- {} --".format(self.sweep, pw)) self.log.info("{} ----- {}".format(i, t[0])) self.log.info("---------------------") current.append(i) throughput.append(t[0]) print(sweep_range) print(current) print(throughput)