1#!/usr/bin/env python3
2#
3# Copyright (C) 2018 The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may not
6# use this file except in compliance with the License. You may obtain a copy of
7# 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, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations under
15# the License.
16
17import json
18import os
19import threading
20import time
21
22from acts.base_test import BaseTestClass
23from acts.controllers import android_device
24from acts.controllers import relay_device_controller
25from acts.test_utils.bt.bt_test_utils import disable_bluetooth
26from acts.test_utils.bt.bt_test_utils import enable_bluetooth
27from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
28from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
29from acts.test_utils.coex.coex_test_utils import A2dpDumpsysParser
30from acts.test_utils.coex.coex_test_utils import (
31    collect_bluetooth_manager_dumpsys_logs)
32from acts.test_utils.coex.coex_test_utils import configure_and_start_ap
33from acts.test_utils.coex.coex_test_utils import iperf_result
34from acts.test_utils.coex.coex_test_utils import parse_fping_results
35from acts.test_utils.coex.coex_test_utils import wifi_connection_check
36from acts.test_utils.wifi import wifi_retail_ap as retail_ap
37from acts.test_utils.wifi.wifi_performance_test_utils import get_iperf_arg_string
38from acts.test_utils.wifi.wifi_power_test_utils import get_phone_ip
39from acts.test_utils.wifi.wifi_test_utils import reset_wifi
40from acts.test_utils.wifi.wifi_test_utils import wifi_connect
41from acts.test_utils.wifi.wifi_test_utils import wifi_test_device_init
42from acts.test_utils.wifi.wifi_test_utils import wifi_toggle_state
43
44AVRCP_WAIT_TIME = 3
45
46
47class CoexBaseTest(BaseTestClass):
48
49    class IperfVariables:
50
51        def __init__(self, current_test_name):
52            self.iperf_started = False
53            self.bidirectional_client_path = None
54            self.bidirectional_server_path = None
55            self.iperf_server_path = None
56            self.iperf_client_path = None
57            self.throughput = []
58            self.protocol = current_test_name.split("_")[-2]
59            self.stream = current_test_name.split("_")[-1]
60            self.is_bidirectional = False
61            if self.stream == 'bidirectional':
62                self.is_bidirectional = True
63
64    def setup_class(self):
65        super().setup_class()
66        self.pri_ad = self.android_devices[0]
67        if len(self.android_devices) == 2:
68            self.sec_ad = self.android_devices[1]
69        elif len(self.android_devices) == 3:
70            self.third_ad = self.android_devices[2]
71        self.ssh_config = None
72
73        self.counter = 0
74        self.thread_list = []
75        if not setup_multiple_devices_for_bt_test(self.android_devices):
76            self.log.error('Failed to setup devices for bluetooth test')
77            return False
78        req_params = ['network', 'iperf']
79        opt_params = [
80            'AccessPoint', 'RetailAccessPoints', 'RelayDevice',
81            'required_devices'
82        ]
83        self.unpack_userparams(req_params, opt_params)
84
85        self.iperf_server = self.iperf_servers[0]
86        self.iperf_client = self.iperf_clients[0]
87
88        if hasattr(self, 'RelayDevice'):
89            self.audio_receiver = self.relay_devices[0]
90            self.audio_receiver.power_on()
91            self.headset_mac_address = self.audio_receiver.mac_address
92        else:
93            self.log.warning('Missing Relay config file.')
94
95        if hasattr(self, 'AccessPoint'):
96            self.ap = self.access_points[0]
97            configure_and_start_ap(self.ap, self.network)
98        elif hasattr(self, 'RetailAccessPoints'):
99            self.retail_access_points = retail_ap.create(
100                self.RetailAccessPoints)
101            self.retail_access_point = self.retail_access_points[0]
102            band = self.retail_access_point.band_lookup_by_channel(
103                self.network['channel'])
104            self.retail_access_point.set_channel(band, self.network['channel'])
105        else:
106            self.log.warning('config file have no access point information')
107
108        wifi_test_device_init(self.pri_ad)
109        wifi_connect(self.pri_ad, self.network, num_of_tries=5)
110
111    def setup_test(self):
112        self.tag = 0
113        self.result = {}
114        self.dev_list = {}
115        self.iperf_variables = self.IperfVariables(self.current_test_name)
116        self.a2dp_dumpsys = A2dpDumpsysParser()
117        self.log_path = os.path.join(self.pri_ad.log_path,
118                self.current_test_name)
119        os.makedirs(self.log_path, exist_ok=True)
120        self.json_file = os.path.join(self.log_path, 'test_results.json')
121        for a in self.android_devices:
122            a.ed.clear_all_events()
123        if not wifi_connection_check(self.pri_ad, self.network['SSID']):
124            self.log.error('Wifi connection does not exist')
125            return False
126        if not enable_bluetooth(self.pri_ad.droid, self.pri_ad.ed):
127            self.log.error('Failed to enable bluetooth')
128            return False
129        if hasattr(self, 'required_devices'):
130            if ('discovery' in self.current_test_name or
131                    'ble' in self.current_test_name):
132                self.create_android_relay_object()
133        else:
134            self.log.warning('required_devices is not given in config file')
135
136    def teardown_test(self):
137        self.parsing_results()
138        with open(self.json_file, 'a') as results_file:
139            json.dump(self.result, results_file, indent=4, sort_keys=True)
140        if not disable_bluetooth(self.pri_ad.droid):
141            self.log.info('Failed to disable bluetooth')
142            return False
143        self.destroy_android_and_relay_object()
144
145    def teardown_class(self):
146        if hasattr(self, 'AccessPoint'):
147            self.ap.close()
148        self.reset_wifi_and_store_results()
149
150    def reset_wifi_and_store_results(self):
151        """Resets wifi and store test results."""
152        reset_wifi(self.pri_ad)
153        wifi_toggle_state(self.pri_ad, False)
154
155    def create_android_relay_object(self):
156        """Creates android device object and relay device object if required
157        devices has android device and relay device."""
158        if 'AndroidDevice' in self.required_devices:
159            self.inquiry_devices = android_device.create(
160                self.required_devices['AndroidDevice'])
161            self.dev_list['AndroidDevice'] = self.inquiry_devices
162        if 'RelayDevice' in self.required_devices:
163            self.relay = relay_device_controller.create(
164                self.required_devices['RelayDevice'])
165            self.dev_list['RelayDevice'] = self.relay
166
167    def destroy_android_and_relay_object(self):
168        """Destroys android device object and relay device object if required
169        devices has android device and relay device."""
170        if hasattr(self, 'required_devices'):
171            if ('discovery' in self.current_test_name or
172                    'ble' in self.current_test_name):
173                if hasattr(self, 'inquiry_devices'):
174                    for device in range(len(self.inquiry_devices)):
175                        inquiry_device = self.inquiry_devices[device]
176                        if not disable_bluetooth(inquiry_device.droid):
177                            self.log.info('Failed to disable bluetooth')
178                    android_device.destroy(self.inquiry_devices)
179                if hasattr(self, 'relay'):
180                    relay_device_controller.destroy(self.relay)
181
182    def parsing_results(self):
183        """Result parser for fping results and a2dp packet drops."""
184        if 'fping' in self.current_test_name:
185            output_path = '{}{}{}'.format(self.pri_ad.log_path, '/Fping/',
186                                          'fping_%s.txt' % self.counter)
187            self.result['fping_loss%'] = parse_fping_results(
188                self.fping_params['fping_drop_tolerance'], output_path)
189            self.counter = +1
190        if 'a2dp_streaming' in self.current_test_name:
191            file_path = collect_bluetooth_manager_dumpsys_logs(
192                self.pri_ad, self.current_test_name)
193            self.result['a2dp_packet_drop'] = (
194                self.a2dp_dumpsys.parse(file_path))
195            if self.result['a2dp_packet_drop'] == 0:
196                self.result['a2dp_packet_drop'] = None
197
198    def run_iperf_and_get_result(self):
199        """Frames iperf command based on test and starts iperf client on
200        host machine.
201
202        Returns:
203            throughput: Throughput of the run.
204        """
205        self.iperf_server.start(tag=self.tag)
206        if self.iperf_variables.stream == 'ul':
207            iperf_args = get_iperf_arg_string(
208                duration=self.iperf['duration'],
209                reverse_direction=1,
210                traffic_type=self.iperf_variables.protocol
211            )
212        elif self.iperf_variables.stream == 'dl':
213            iperf_args = get_iperf_arg_string(
214                duration=self.iperf['duration'],
215                reverse_direction=0,
216                traffic_type=self.iperf_variables.protocol
217            )
218        ip = get_phone_ip(self.pri_ad)
219        self.tag = self.tag + 1
220        self.iperf_variables.iperf_client_path = (
221                self.iperf_client.start(ip, iperf_args, self.tag))
222
223        self.iperf_server.stop()
224        if (self.iperf_variables.protocol == 'udp' and
225                self.iperf_variables.stream == 'ul'):
226            throughput = iperf_result(
227                self.log, self.iperf_variables.protocol,
228                self.iperf_variables.iperf_server_path)
229        else:
230            throughput = iperf_result(self.log,
231                                    self.iperf_variables.protocol,
232                                    self.iperf_variables.iperf_client_path)
233
234        if not throughput:
235            self.log.error('Iperf failed/stopped')
236            self.iperf_variables.throughput.append(0)
237        else:
238            self.iperf_variables.throughput.append(
239                str(round(throughput, 2)) + "Mb/s")
240            self.log.info("Throughput: {} Mb/s".format(throughput))
241        self.result["throughput"] = self.iperf_variables.throughput
242        return throughput
243
244    def on_fail(self, test_name, begin_time):
245        """A function that is executed upon a test case failure.
246
247        Args:
248            test_name: Name of the test that triggered this function.
249            begin_time: Logline format timestamp taken when the test started.
250        """
251        self.log.info('Test {} failed, Fetching Btsnoop logs and bugreport'.
252                      format(test_name))
253        take_btsnoop_logs(self.android_devices, self, test_name)
254        self._take_bug_report(test_name, begin_time)
255
256    def run_thread(self, kwargs):
257        """Convenience function to start thread.
258
259        Args:
260            kwargs: Function object to start in thread.
261        """
262        for function in kwargs:
263            self.thread = threading.Thread(target=function)
264            self.thread_list.append(self.thread)
265            self.thread.start()
266
267    def teardown_thread(self):
268        """Convenience function to join thread."""
269        for thread_id in self.thread_list:
270            if thread_id.is_alive():
271                thread_id.join()
272
273    def get_call_volume(self):
274        """Function to get call volume when bluetooth headset connected.
275
276        Returns:
277            Call volume.
278        """
279        return self.pri_ad.adb.shell(
280            'settings list system|grep volume_bluetooth_sco_bt_sco_hs')
281
282    def change_volume(self):
283        """Changes volume with HFP call.
284
285        Returns: True if successful, otherwise False.
286        """
287        if 'Volume_up' and 'Volume_down' in (
288                self.relay_devices[0].relays.keys()):
289            current_volume = self.get_call_volume()
290            self.audio_receiver.press_volume_down()
291            time.sleep(AVRCP_WAIT_TIME)  # wait till volume_changes
292            if current_volume == self.get_call_volume():
293                self.log.error('Decrease volume failed')
294                return False
295            time.sleep(AVRCP_WAIT_TIME)
296            current_volume = self.get_call_volume()
297            self.audio_receiver.press_volume_up()
298            time.sleep(AVRCP_WAIT_TIME)  # wait till volume_changes
299            if current_volume == self.get_call_volume():
300                self.log.error('Increase volume failed')
301                return False
302        else:
303            self.log.warning(
304                'No volume control pins specified in relay config.')
305        return True
306