#!/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. from retry import retry import acts_contrib.test_utils.power.PowerBaseTest as PBT from acts_contrib.test_utils.wifi import wifi_test_utils as wutils from acts_contrib.test_utils.wifi import wifi_power_test_utils as wputils from acts_contrib.test_utils.power import plot_utils IPERF_DURATION = 'iperf_duration' INITIAL_ATTEN = [0, 0, 90, 90] IPERF_TAIL = 5 class PowerWiFiBaseTest(PBT.PowerBaseTest): """Base class for WiFi power related tests. Inherited from the PowerBaseTest class """ def setup_class(self): super().setup_class() if hasattr(self, 'access_points'): self.access_point = self.access_points[0] self.access_point_main = self.access_points[0] if len(self.access_points) > 1: self.access_point_aux = self.access_points[1] if hasattr(self, 'attenuators'): self.set_attenuation(INITIAL_ATTEN) if hasattr(self, 'network_file'): self.networks = self.unpack_custom_file(self.network_file, False) self.main_network = self.networks['main_network'] self.aux_network = self.networks['aux_network'] if hasattr(self, 'packet_senders'): self.pkt_sender = self.packet_senders[0] if hasattr(self, 'iperf_servers'): self.iperf_server = self.iperf_servers[0] if self.iperf_duration: self.mon_duration = self.iperf_duration - self.mon_offset - IPERF_TAIL self.mon_info = self.create_monsoon_info() try: self._set_country_code() except Exception as e: self.log.warning('error in set country code with %s', e) country_code = self.dut.droid.wifiGetCountryCode() self.log.info('the country code is %s', country_code) @retry(tries=5, delay=10) def _set_country_code(self): wutils.wifi_toggle_state(self.dut, True) wutils.set_wifi_country_code(self.dut, 'US') def teardown_test(self): """Tear down necessary objects after test case is finished. Bring down the AP interface, delete the bridge interface, stop the packet sender, and reset the ethernet interface for the packet sender """ super().teardown_test() if hasattr(self, 'pkt_sender'): self._safe_teardown('pkt_sender stop sending', self.pkt_sender.stop_sending, ignore_status=True) if hasattr(self, 'brconfigs'): self._safe_teardown('brconfigs', self.access_point.bridge.teardown, self.brconfigs) delattr(self, 'brconfigs') if hasattr(self, 'brconfigs_main'): self._safe_teardown('brconfigs_main', self.access_point_main.bridge.teardown, self.brconfigs_main) delattr(self, 'brconfigs_main') if hasattr(self, 'brconfigs_aux'): self._safe_teardown('brconfigs_aux', self.access_point_aux.bridge.teardown, self.brconfigs_aux) delattr(self, 'brconfigs_aux') if hasattr(self, 'access_points'): for ap in self.access_points: self._safe_teardown('access point {}'.format(ap.identifier), ap.close) if hasattr(self, 'pkt_sender'): self._safe_teardown('pkt_sender reset host interface', wputils.reset_host_interface, self.pkt_sender.interface) if hasattr(self, 'iperf_server'): self._safe_teardown('iperf_server', self.iperf_server.stop); def _safe_teardown(self, attr, teardown_method, *arg, **kwargs): """Teardown the object with try block. Adds a try block for each teardown step to make sure that each teardown step is executed. Args: attr: the teardown attribute description for logging teardown_method: the method for teardown *arg: positional arguments for teardown_method **kwargs: keyword arguments for teardown_method """ try: self.log.info('teardown %s with %s', attr, teardown_method.__name__) teardown_method(*arg, **kwargs) except Exception as e: self.log.warning('teardown of %s fails with %s', attr, e) def teardown_class(self): """Clean up the test class after tests finish running """ super().teardown_class() if hasattr(self, 'access_points'): for ap in self.access_points: ap.close() def setup_ap_connection(self, network, bandwidth=80, connect=True, ap=None, dtim_period=None): """Setup AP and connect DUT to it. Args: network: the network config for the AP to be setup bandwidth: bandwidth of the WiFi network to be setup connect: indicator of if connect dut to the network after setup ap: access point object, default is None to find the main AP dtim_period: the dtim period of access point Returns: self.brconfigs: dict for bridge interface configs """ wutils.wifi_toggle_state(self.dut, True) if not dtim_period: dtim_period = self.ap_dtim_period if not ap: if hasattr(self, 'access_points'): self.brconfigs = wputils.ap_setup( self.access_point, network, bandwidth=bandwidth, dtim_period=dtim_period) else: self.brconfigs = wputils.ap_setup( ap, network, bandwidth=bandwidth, dtim_period=dtim_period) if connect: wutils.wifi_connect(self.dut, network, num_of_tries=3) if ap or (not ap and hasattr(self, 'access_points')): return self.brconfigs def collect_power_data(self): """Measure power, plot and check pass/fail. If IPERF is run, need to pull iperf results and attach it to the plot. """ samples = super().collect_power_data() tag = '' if self.iperf_duration: throughput = self.process_iperf_results() plot_title = ('{0}_{1}_{2}_RSSI_{3:d}dBm_Throughput_{4:.2f}' 'Mbps'.format(self.test_name, self.dut.model, self.dut.build_info['build_id'], self.RSSI, throughput)) plot_utils.current_waveform_plot(samples, self.mon_voltage, self.mon_info.data_path, plot_title) return samples def setup_test(self): """Set up test specific parameters or configs. """ super().setup_test() wutils.reset_wifi(self.dut) wutils.wifi_toggle_state(self.dut, False)