1#/usr/bin/env python3.4
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 os
18import threading
19import time
20
21from acts.base_test import BaseTestClass
22from acts.controllers.iperf_client import IPerfClient
23from acts.test_utils.bt.bt_test_utils import disable_bluetooth
24from acts.test_utils.bt.bt_test_utils import enable_bluetooth
25from acts.test_utils.bt.bt_test_utils import setup_multiple_devices_for_bt_test
26from acts.test_utils.bt.bt_test_utils import take_btsnoop_logs
27from acts.test_utils.coex.coex_test_utils import configure_and_start_ap
28from acts.test_utils.coex.coex_test_utils import iperf_result
29from acts.test_utils.coex.coex_test_utils import get_phone_ip
30from acts.test_utils.coex.coex_test_utils import wifi_connection_check
31from acts.test_utils.coex.coex_test_utils import xlsheet
32from acts.test_utils.wifi.wifi_test_utils import reset_wifi
33from acts.test_utils.wifi.wifi_test_utils import wifi_connect
34from acts.test_utils.wifi.wifi_test_utils import wifi_test_device_init
35from acts.test_utils.wifi.wifi_test_utils import wifi_toggle_state
36from acts.utils import create_dir
37from acts.utils import start_standing_subprocess
38from acts.utils import stop_standing_subprocess
39
40TEST_CASE_TOKEN = "[Test Case]"
41RESULT_LINE_TEMPLATE = TEST_CASE_TOKEN + " %s %s"
42IPERF_SERVER_WAIT_TIME = 5
43
44
45class CoexBaseTest(BaseTestClass):
46
47    def __init__(self, controllers):
48        BaseTestClass.__init__(self, controllers)
49        self.pri_ad = self.android_devices[0]
50        if len(self.android_devices) == 2:
51            self.sec_ad = self.android_devices[1]
52        elif len(self.android_devices) == 3:
53            self.third_ad = self.android_devices[2]
54
55    def setup_class(self):
56        self.tag = 0
57        self.iperf_result = {}
58        self.thread_list = []
59        if not setup_multiple_devices_for_bt_test(self.android_devices):
60            self.log.error("Failed to setup devices for bluetooth test")
61            return False
62        req_params = ["network", "iperf"]
63        self.unpack_userparams(req_params)
64        if "RelayDevice" in self.user_params:
65            self.audio_receiver = self.relay_devices[0]
66        else:
67            self.log.warning("Missing Relay config file.")
68        if "music_file" in self.user_params:
69            self.push_music_to_android_device(self.pri_ad)
70        self.path = self.pri_ad.log_path
71        if "AccessPoints" in self.user_params:
72            self.ap = self.access_points[0]
73            configure_and_start_ap(self.ap, self.network)
74        wifi_test_device_init(self.pri_ad)
75        wifi_connect(self.pri_ad, self.network)
76
77    def setup_test(self):
78        self.received = []
79        for a in self.android_devices:
80            a.ed.clear_all_events()
81        if not wifi_connection_check(self.pri_ad, self.network["SSID"]):
82            self.log.error("Wifi connection does not exist")
83            return False
84        if not enable_bluetooth(self.pri_ad.droid, self.pri_ad.ed):
85            self.log.error("Failed to enable bluetooth")
86            return False
87
88    def teardown_test(self):
89        if not disable_bluetooth(self.pri_ad.droid):
90            self.log.info("Failed to disable bluetooth")
91            return False
92        self.teardown_thread()
93
94    def teardown_class(self):
95        if "AccessPoints" in self.user_params:
96            self.ap.close()
97        reset_wifi(self.pri_ad)
98        wifi_toggle_state(self.pri_ad, False)
99        json_result = self.results.json_str()
100        xlsheet(self.pri_ad, json_result)
101
102    def start_iperf_server_on_shell(self, server_port):
103        """Starts iperf server on android device with specified.
104
105        Args:
106            server_port: Port in which server should be started.
107        """
108        log_path = os.path.join(self.pri_ad.log_path, "iPerf{}".format(
109            server_port))
110        iperf_server = "iperf3 -s -p {} -J".format(server_port)
111        log_files = []
112        create_dir(log_path)
113        out_file_name = "IPerfServer,{},{},{}.log".format(
114            server_port, self.tag, log_files)
115        self.tag = self.tag + 1
116        full_out_path = os.path.join(log_path, out_file_name)
117        cmd = "adb -s {} shell {} > {}".format(
118            self.pri_ad.serial, iperf_server, full_out_path)
119        self.iperf_process.append(start_standing_subprocess(cmd))
120        log_files.append(full_out_path)
121        time.sleep(IPERF_SERVER_WAIT_TIME)
122
123    def stop_iperf_server_on_shell(self):
124        """Stops all the instances of iperf server on shell."""
125        try:
126            for process in self.iperf_process:
127                stop_standing_subprocess(process)
128        except Exception:
129            self.log.info("Iperf server already killed")
130
131    def run_iperf_and_get_result(self):
132        """Frames iperf command based on test and starts iperf client on
133        host machine.
134        """
135        self.flag_list = []
136        self.iperf_process = []
137        test_params = self.current_test_name.split("_")
138
139        self.protocol = test_params[-2:-1]
140        self.stream = test_params[-1:]
141
142        if self.protocol[0] == "tcp":
143            self.iperf_args = "-t {} -p {} {} -J".format(
144                self.iperf["duration"], self.iperf["port_1"],
145                self.iperf["tcp_window_size"])
146        else:
147            self.iperf_args = ("-t {} -p {} -u {} --get-server-output -J"
148                               .format(self.iperf["duration"],
149                                       self.iperf["port_1"],
150                                       self.iperf["udp_bandwidth"]))
151
152        if self.stream[0] == "ul":
153            self.iperf_args += " -R"
154
155        if self.protocol[0] == "tcp" and self.stream[0] == "bidirectional":
156            self.bidirectional_args = "-t {} -p {} {} -R -J".format(
157                self.iperf["duration"], self.iperf["port_2"],
158                self.iperf["tcp_window_size"])
159        else:
160            self.bidirectional_args = ("-t {} -p {} -u {} --get-server-output"
161                                       " -J".format(self.iperf["duration"],
162                                                    self.iperf["port_2"],
163                                                    self.iperf["udp_bandwidth"]
164                                                    ))
165
166        if self.stream[0] == "bidirectional":
167            self.start_iperf_server_on_shell(self.iperf["port_2"])
168        self.start_iperf_server_on_shell(self.iperf["port_1"])
169
170        if self.stream[0] == "bidirectional":
171            args = [
172                lambda: self.run_iperf(self.iperf_args, self.iperf["port_1"]),
173                lambda: self.run_iperf(self.bidirectional_args,
174                                       self.iperf["port_2"])
175            ]
176            self.run_thread(args)
177        else:
178            args = [
179                lambda: self.run_iperf(self.iperf_args, self.iperf["port_1"])
180            ]
181            self.run_thread(args)
182
183    def run_iperf(self, iperf_args, server_port):
184        """Gets android device ip and start iperf client from host machine to
185        that ip and parses the iperf result.
186
187        Args:
188            iperf_args: Iperf parameters to run traffic.
189            server_port: Iperf port to start client.
190        """
191        ip = get_phone_ip(self.pri_ad)
192        iperf_client = IPerfClient(server_port, ip, self.pri_ad.log_path)
193        result = iperf_client.start(iperf_args)
194        try:
195            sent, received = iperf_result(result, self.stream)
196            self.received.append(str(round(received, 2)) + "Mb/s")
197            self.log.info("Sent: {} Mb/s, Received: {} Mb/s".format(
198                sent, received))
199            self.flag_list.append(True)
200
201        except TypeError:
202            self.log.error("Iperf failed/stopped.")
203            self.flag_list.append(False)
204            self.received.append("Iperf Failed")
205
206        self.iperf_result[self.current_test_name] = self.received
207
208    def on_fail(self, record, test_name, begin_time):
209        self.log.info(
210            "Test {} failed, Fetching Btsnoop logs and bugreport".format(
211                test_name))
212        take_btsnoop_logs(self.android_devices, self, test_name)
213        self._take_bug_report(test_name, begin_time)
214        record.extras = self.received
215
216    def _on_fail(self, record):
217        """Proxy function to guarantee the base implementation of on_fail is
218        called.
219
220        Args:
221            record: The records.TestResultRecord object for the failed test
222            case.
223        """
224        if record.details:
225            self.log.error(record.details)
226        self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result)
227        self.on_fail(record, record.test_name, record.log_begin_time)
228
229    def _on_pass(self, record):
230        """Proxy function to guarantee the base implementation of on_pass is
231        called.
232
233        Args:
234            record: The records.TestResultRecord object for the passed test
235            case.
236        """
237        msg = record.details
238        if msg:
239            self.log.info(msg)
240        self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result)
241        record.extras = self.received
242
243    def run_thread(self, kwargs):
244        """Convenience function to start thread.
245
246        Args:
247            kwargs: Function object to start in thread.
248        """
249        for function in kwargs:
250            self.thread = threading.Thread(target=function)
251            self.thread_list.append(self.thread)
252            self.thread.start()
253
254    def teardown_result(self):
255        """Convenience function to join thread and fetch iperf result."""
256        for thread_id in self.thread_list:
257            if thread_id.is_alive():
258                thread_id.join()
259        self.stop_iperf_server_on_shell()
260        if False in self.flag_list:
261            return False
262        return True
263
264    def teardown_thread(self):
265        """Convenience function to join thread."""
266        for thread_id in self.thread_list:
267            if thread_id.is_alive():
268                thread_id.join()
269        self.stop_iperf_server_on_shell()
270
271    def push_music_to_android_device(self, ad):
272        """Add music to Android device as specified by the test config
273
274        Args:
275            ad: Android device
276
277        Returns:
278            True on success, False on failure
279        """
280        self.log.info("Pushing music to the Android device")
281        music_path_str = "music_file"
282        android_music_path = "/sdcard/Music/"
283        if music_path_str not in self.user_params:
284            self.log.error("Need music for audio testcases")
285            return False
286        music_path = self.user_params[music_path_str]
287        if type(music_path) is list:
288            self.log.info("Media ready to push as is.")
289        if type(music_path) is list:
290            for item in music_path:
291                self.music_file_to_play = item
292                ad.adb.push("{} {}".format(item, android_music_path))
293        return True
294
295    def avrcp_actions(self):
296        """Performs avrcp controls like volume up, volume down, skip next and
297        skip previous.
298
299        Returns: True if successful, otherwise False.
300        """
301        #TODO: Validate the success state of functionalities performed.
302        self.audio_receiver.volume_up()
303        time.sleep(2)
304        self.audio_receiver.volume_down()
305        time.sleep(2)
306        self.audio_receiver.skip_next()
307        time.sleep(2)
308        self.audio_receiver.skip_previous()
309        time.sleep(2)
310        return True
311
312    def change_volume(self):
313        """Changes volume with HFP call.
314
315        Returns: True if successful, otherwise False.
316        """
317        self.audio_receiver.volume_up()
318        time.sleep(2)
319        self.audio_receiver.volume_down()
320        time.sleep(2)
321        return True
322