• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3.4
2#
3#   Copyright 2018 - 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 acts_contrib.test_utils.power.PowerBaseTest as PBT
18from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
19from acts_contrib.test_utils.wifi import wifi_power_test_utils as wputils
20from acts_contrib.test_utils.power import plot_utils
21
22IPERF_DURATION = 'iperf_duration'
23INITIAL_ATTEN = [0, 0, 90, 90]
24IPERF_TAIL = 5
25
26
27class PowerWiFiBaseTest(PBT.PowerBaseTest):
28    """Base class for WiFi power related tests.
29
30    Inherited from the PowerBaseTest class
31    """
32    def setup_class(self):
33
34        super().setup_class()
35        if hasattr(self, 'access_points'):
36            self.access_point = self.access_points[0]
37            self.access_point_main = self.access_points[0]
38            if len(self.access_points) > 1:
39                self.access_point_aux = self.access_points[1]
40        if hasattr(self, 'attenuators'):
41            self.set_attenuation(INITIAL_ATTEN)
42        if hasattr(self, 'network_file'):
43            self.networks = self.unpack_custom_file(self.network_file, False)
44            self.main_network = self.networks['main_network']
45            self.aux_network = self.networks['aux_network']
46        if hasattr(self, 'packet_senders'):
47            self.pkt_sender = self.packet_senders[0]
48        if hasattr(self, 'iperf_servers'):
49            self.iperf_server = self.iperf_servers[0]
50        if self.iperf_duration:
51            self.mon_duration = self.iperf_duration - self.mon_offset - IPERF_TAIL
52            self.create_monsoon_info()
53
54    def teardown_test(self):
55        """Tear down necessary objects after test case is finished.
56
57        Bring down the AP interface, delete the bridge interface, stop the
58        packet sender, and reset the ethernet interface for the packet sender
59        """
60        super().teardown_test()
61        if hasattr(self, 'pkt_sender'):
62            self._safe_teardown('pkt_sender stop sending',
63                                self.pkt_sender.stop_sending,
64                                ignore_status=True)
65        if hasattr(self, 'brconfigs'):
66            self._safe_teardown('brconfigs', self.access_point.bridge.teardown,
67                                self.brconfigs)
68            delattr(self, 'brconfigs')
69        if hasattr(self, 'brconfigs_main'):
70            self._safe_teardown('brconfigs_main',
71                                self.access_point_main.bridge.teardown,
72                                self.brconfigs_main)
73            delattr(self, 'brconfigs_main')
74        if hasattr(self, 'brconfigs_aux'):
75            self._safe_teardown('brconfigs_aux',
76                                self.access_point_aux.bridge.teardown,
77                                self.brconfigs_aux)
78            delattr(self, 'brconfigs_aux')
79        if hasattr(self, 'access_points'):
80            for ap in self.access_points:
81                self._safe_teardown('access point {}'.format(ap.identifier),
82                                    ap.close)
83        if hasattr(self, 'pkt_sender'):
84            self._safe_teardown('pkt_sender reset host interface',
85                                wputils.reset_host_interface,
86                                self.pkt_sender.interface)
87        if hasattr(self, 'iperf_server'):
88            self._safe_teardown('iperf_server', self.iperf_server.stop);
89
90    def _safe_teardown(self, attr, teardown_method, *arg, **kwargs):
91        """Teardown the object with try block.
92
93        Adds a try block for each teardown step to make sure that each
94        teardown step is executed.
95
96        Args:
97            attr: the teardown attribute description for logging
98            teardown_method: the method for teardown
99            *arg: positional arguments for teardown_method
100            **kwargs: keyword arguments for teardown_method
101        """
102        try:
103            self.log.info('teardown %s with %s', attr, teardown_method.__name__)
104            teardown_method(*arg, **kwargs)
105        except Exception as e:
106            self.log.warning('teardown of %s fails with %s', attr, e)
107
108    def teardown_class(self):
109        """Clean up the test class after tests finish running
110
111        """
112        super().teardown_class()
113        if hasattr(self, 'access_points'):
114            for ap in self.access_points:
115                ap.close()
116
117    def setup_ap_connection(self,
118                            network,
119                            bandwidth=80,
120                            connect=True,
121                            ap=None):
122        """Setup AP and connect DUT to it.
123
124        Args:
125            network: the network config for the AP to be setup
126            bandwidth: bandwidth of the WiFi network to be setup
127            connect: indicator of if connect dut to the network after setup
128            ap: access point object, default is None to find the main AP
129        Returns:
130            self.brconfigs: dict for bridge interface configs
131        """
132        wutils.wifi_toggle_state(self.dut, True)
133        if not ap:
134            if hasattr(self, 'access_points'):
135                self.brconfigs = wputils.ap_setup(self.access_point,
136                                                  network,
137                                                  bandwidth=bandwidth)
138        else:
139            self.brconfigs = wputils.ap_setup(ap, network, bandwidth=bandwidth)
140        if connect:
141            wutils.wifi_connect(self.dut, network, num_of_tries=3)
142
143        if ap or (not ap and hasattr(self, 'access_points')):
144            return self.brconfigs
145
146    def collect_power_data(self):
147        """Measure power, plot and check pass/fail.
148
149        If IPERF is run, need to pull iperf results and attach it to the plot.
150        """
151        samples = super().collect_power_data()
152        tag = ''
153        if self.iperf_duration:
154            throughput = self.process_iperf_results()
155            plot_title = '{}_{}_{}_RSSI_{0:d}dBm_Throughput_{1:.2f}Mbps'.format(
156                self.test_name, self.dut.model,
157                self.dut.build_info['build_id'], self.RSSI, throughput)
158            plot_utils.current_waveform_plot(samples, self.mon_voltage,
159                                             self.mon_info.data_path,
160                                             plot_title)
161        return samples
162