1#!/usr/bin/env python3.4
2#
3#   Copyright 2016 - Google
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"""
17    Base Class for Defining Common Telephony Test Functionality
18"""
19
20import logging
21import os
22import re
23import shutil
24import traceback
25
26import acts.controllers.diag_logger
27
28from acts import asserts
29from acts import logger as acts_logger
30from acts.base_test import BaseTestClass
31from acts.controllers.android_device import DEFAULT_QXDM_LOG_PATH
32from acts.keys import Config
33from acts.signals import TestSignal
34from acts.signals import TestAbortClass
35from acts.signals import TestAbortAll
36from acts.signals import TestBlocked
37from acts import records
38from acts import utils
39
40from acts.test_utils.tel.tel_subscription_utils import \
41    initial_set_up_for_subid_infomation
42from acts.test_utils.tel.tel_test_utils import abort_all_tests
43from acts.test_utils.tel.tel_test_utils import ensure_phones_default_state
44from acts.test_utils.tel.tel_test_utils import ensure_phones_idle
45from acts.test_utils.tel.tel_test_utils import print_radio_info
46from acts.test_utils.tel.tel_test_utils import reboot_device
47from acts.test_utils.tel.tel_test_utils import refresh_sl4a_session
48from acts.test_utils.tel.tel_test_utils import run_multithread_func
49from acts.test_utils.tel.tel_test_utils import setup_droid_properties
50from acts.test_utils.tel.tel_test_utils import set_phone_screen_on
51from acts.test_utils.tel.tel_test_utils import set_phone_silent_mode
52from acts.test_utils.tel.tel_test_utils import set_qxdm_logger_command
53from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers
54from acts.test_utils.tel.tel_test_utils import stop_qxdm_loggers
55from acts.test_utils.tel.tel_test_utils import unlock_sim
56from acts.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND
57from acts.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND
58from acts.test_utils.tel.tel_defines import PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING
59from acts.test_utils.tel.tel_defines import WIFI_VERBOSE_LOGGING_ENABLED
60from acts.test_utils.tel.tel_defines import WIFI_VERBOSE_LOGGING_DISABLED
61
62
63class TelephonyBaseTest(BaseTestClass):
64    def __init__(self, controllers):
65
66        BaseTestClass.__init__(self, controllers)
67        self.logger_sessions = []
68
69        self.log_path = getattr(logging, "log_path", None)
70        self.qxdm_log = self.user_params.get("qxdm_log", True)
71        qxdm_log_mask_cfg = self.user_params.get("qxdm_log_mask_cfg", None)
72        if isinstance(qxdm_log_mask_cfg, list):
73            qxdm_log_mask_cfg = qxdm_log_mask_cfg[0]
74        if qxdm_log_mask_cfg and "dev/null" in qxdm_log_mask_cfg:
75            qxdm_log_mask_cfg = None
76        stop_qxdm_loggers(self.log, self.android_devices)
77        for ad in self.android_devices:
78            try:
79                ad.adb.shell("killall -9 tcpdump")
80            except AdbError:
81                ad.log.warn("Killing existing tcpdump processes failed")
82            if not hasattr(ad, "init_log_path"):
83                ad.init_log_path = ad.log_path
84            ad.log_path = self.log_path
85            print_radio_info(ad)
86            if not unlock_sim(ad):
87                abort_all_tests(ad.log, "unable to unlock SIM")
88            ad.wakeup_screen()
89            ad.adb.shell("input keyevent 82")
90            ad.qxdm_log = getattr(ad, "qxdm_log", self.qxdm_log)
91            if ad.qxdm_log:
92                qxdm_log_mask = getattr(ad, "qxdm_log_mask", None)
93                if qxdm_log_mask_cfg:
94                    qxdm_mask_path = DEFAULT_QXDM_LOG_PATH
95                    ad.adb.shell("mkdir %s" % qxdm_mask_path)
96                    ad.log.info("Push %s to %s", qxdm_log_mask_cfg,
97                                qxdm_mask_path)
98                    ad.adb.push("%s %s" % (qxdm_log_mask_cfg, qxdm_mask_path))
99                    mask_file_name = os.path.split(qxdm_log_mask_cfg)[-1]
100                    qxdm_log_mask = os.path.join(qxdm_mask_path,
101                                                 mask_file_name)
102                set_qxdm_logger_command(ad, mask=qxdm_log_mask)
103                ad.adb.pull(
104                    "/firmware/image/qdsp6m.qdb %s" % ad.init_log_path,
105                    ignore_status=True)
106
107        start_qxdm_loggers(self.log, self.android_devices,
108                           utils.get_current_epoch_time())
109        self.skip_reset_between_cases = self.user_params.get(
110            "skip_reset_between_cases", True)
111
112    # Use for logging in the test cases to facilitate
113    # faster log lookup and reduce ambiguity in logging.
114    @staticmethod
115    def tel_test_wrap(fn):
116        def _safe_wrap_test_case(self, *args, **kwargs):
117            test_id = "%s:%s:%s" % (self.__class__.__name__, self.test_name,
118                                    self.log_begin_time.replace(' ', '-'))
119            self.test_id = test_id
120            self.result_detail = ""
121            tries = 2 if self.user_params.get("telephony_auto_rerun") else 1
122            for i in range(tries):
123                result = True
124                log_string = "[Test ID] %s" % test_id
125                if i > 1:
126                    log_string = "[Rerun]%s" % log_string
127                    self.teardown_test()
128                    self.setup_test()
129                self.log.info(log_string)
130                for ad in self.android_devices:
131                    ad.log_path = self.log_path
132                    try:
133                        ad.droid.logI("Started %s" % log_string)
134                    except Exception as e:
135                        ad.log.warning(e)
136                        refresh_sl4a_session(ad)
137                try:
138                    result = fn(self, *args, **kwargs)
139                except (TestSignal, TestAbortClass, TestAbortAll) as signal:
140                    if self.result_detail:
141                        signal.details = self.result_detail
142                    raise
143                except Exception as e:
144                    self.log.error(traceback.format_exc())
145                    asserts.fail(self.result_detail)
146                for ad in self.android_devices:
147                    try:
148                        ad.droid.logI("Finished %s" % log_string)
149                    except Exception as e:
150                        ad.log.warning(e)
151                        refresh_sl4a_session(ad)
152                if result: break
153            if self.user_params.get("check_crash", True):
154                new_crash = ad.check_crash_report(self.test_name,
155                                                  self.begin_time, True)
156                if new_crash:
157                    msg = "Find new crash reports %s" % new_crash
158                    ad.log.error(msg)
159                    self.result_detail = "%s %s %s" % (self.result_detail,
160                                                       ad.serial, msg)
161                    result = False
162            if result:
163                asserts.explicit_pass(self.result_detail)
164            else:
165                asserts.fail(self.result_detail)
166
167        return _safe_wrap_test_case
168
169    def setup_class(self):
170        sim_conf_file = self.user_params.get("sim_conf_file")
171        if not sim_conf_file:
172            self.log.info("\"sim_conf_file\" is not provided test bed config!")
173        else:
174            if isinstance(sim_conf_file, list):
175                sim_conf_file = sim_conf_file[0]
176            # If the sim_conf_file is not a full path, attempt to find it
177            # relative to the config file.
178            if not os.path.isfile(sim_conf_file):
179                sim_conf_file = os.path.join(
180                    self.user_params[Config.key_config_path], sim_conf_file)
181                if not os.path.isfile(sim_conf_file):
182                    self.log.error("Unable to load user config %s ",
183                                   sim_conf_file)
184
185        setattr(self, "diag_logger",
186                self.register_controller(
187                    acts.controllers.diag_logger, required=False))
188
189        if not self.user_params.get("Attenuator"):
190            ensure_phones_default_state(self.log, self.android_devices)
191        else:
192            ensure_phones_idle(self.log, self.android_devices)
193        for ad in self.android_devices:
194            setup_droid_properties(self.log, ad, sim_conf_file)
195
196            # Setup VoWiFi MDN for Verizon. b/33187374
197            build_id = ad.build_info["build_id"]
198            if "vzw" in [
199                    sub["operator"] for sub in ad.cfg["subscription"].values()
200            ] and ad.is_apk_installed("com.google.android.wfcactivation"):
201                ad.log.info("setup VoWiFi MDN per b/33187374")
202                ad.adb.shell("setprop dbg.vzw.force_wfc_nv_enabled true")
203                ad.adb.shell("am start --ei EXTRA_LAUNCH_CARRIER_APP 0 -n "
204                             "\"com.google.android.wfcactivation/"
205                             ".VzwEmergencyAddressActivity\"")
206            # Sub ID setup
207            initial_set_up_for_subid_infomation(self.log, ad)
208            if "enable_wifi_verbose_logging" in self.user_params:
209                ad.droid.wifiEnableVerboseLogging(WIFI_VERBOSE_LOGGING_ENABLED)
210            # If device is setup already, skip the following setup procedures
211            if getattr(ad, "telephony_test_setup", None):
212                continue
213            # Disable Emergency alerts
214            # Set chrome browser start with no-first-run verification and
215            # disable-fre. Give permission to read from and write to storage.
216            for cmd in (
217                    "pm disable com.android.cellbroadcastreceiver",
218                    "pm grant com.android.chrome "
219                    "android.permission.READ_EXTERNAL_STORAGE",
220                    "pm grant com.android.chrome "
221                    "android.permission.WRITE_EXTERNAL_STORAGE",
222                    "rm /data/local/chrome-command-line",
223                    "am set-debug-app --persistent com.android.chrome",
224                    'echo "chrome --no-default-browser-check --no-first-run '
225                    '--disable-fre" > /data/local/tmp/chrome-command-line'):
226                ad.adb.shell(cmd)
227
228            # Curl for 2016/7 devices
229            try:
230                if int(ad.adb.getprop("ro.product.first_api_level")) >= 25:
231                    out = ad.adb.shell("/data/curl --version")
232                    if not out or "not found" in out:
233                        tel_data = self.user_params.get("tel_data", "tel_data")
234                        if isinstance(tel_data, list):
235                            tel_data = tel_data[0]
236                        curl_file_path = os.path.join(tel_data, "curl")
237                        if not os.path.isfile(curl_file_path):
238                            curl_file_path = os.path.join(
239                                self.user_params[Config.key_config_path],
240                                curl_file_path)
241                        if os.path.isfile(curl_file_path):
242                            ad.log.info("Pushing Curl to /data dir")
243                            ad.adb.push("%s /data" % (curl_file_path))
244                            ad.adb.shell(
245                                "chmod 777 /data/curl", ignore_status=True)
246            except Exception:
247                ad.log.info("Failed to push curl on this device")
248
249            # Ensure that a test class starts from a consistent state that
250            # improves chances of valid network selection and facilitates
251            # logging.
252            try:
253                if not set_phone_screen_on(self.log, ad):
254                    self.log.error("Failed to set phone screen-on time.")
255                    return False
256                if not set_phone_silent_mode(self.log, ad):
257                    self.log.error("Failed to set phone silent mode.")
258                    return False
259                ad.droid.telephonyAdjustPreciseCallStateListenLevel(
260                    PRECISE_CALL_STATE_LISTEN_LEVEL_FOREGROUND, True)
261                ad.droid.telephonyAdjustPreciseCallStateListenLevel(
262                    PRECISE_CALL_STATE_LISTEN_LEVEL_RINGING, True)
263                ad.droid.telephonyAdjustPreciseCallStateListenLevel(
264                    PRECISE_CALL_STATE_LISTEN_LEVEL_BACKGROUND, True)
265            except Exception as e:
266                self.log.error("Failure with %s", e)
267            setattr(ad, "telephony_test_setup", True)
268
269        return True
270
271    def teardown_class(self):
272        stop_qxdm_loggers(self.log, self.android_devices)
273        ensure_phones_default_state(self.log, self.android_devices)
274        try:
275            for ad in self.android_devices:
276                ad.droid.disableDevicePassword()
277                if "enable_wifi_verbose_logging" in self.user_params:
278                    ad.droid.wifiEnableVerboseLogging(
279                        WIFI_VERBOSE_LOGGING_DISABLED)
280                if hasattr(ad, "init_log_path"):
281                    ad.log_path = ad.init_log_path
282            return True
283        except Exception as e:
284            self.log.error("Failure with %s", e)
285
286    def setup_test(self):
287        for ad in self.android_devices:
288            ad.ed.clear_all_events()
289            output = ad.adb.logcat("-t 1")
290            match = re.search(r"\d+-\d+\s\d+:\d+:\d+.\d+", output)
291            if match:
292                ad.test_log_begin_time = match.group(0)
293        if getattr(self, "qxdm_log", True):
294            start_qxdm_loggers(self.log, self.android_devices, self.begin_time)
295        if getattr(self, "diag_logger", None):
296            for logger in self.diag_logger:
297                self.log.info("Starting a diagnostic session %s", logger)
298                self.logger_sessions.append((logger, logger.start()))
299        if self.skip_reset_between_cases:
300            ensure_phones_idle(self.log, self.android_devices)
301        else:
302            ensure_phones_default_state(self.log, self.android_devices)
303
304    def on_exception(self, test_name, begin_time):
305        self._pull_diag_logs(test_name, begin_time)
306        self._take_bug_report(test_name, begin_time)
307        self._cleanup_logger_sessions()
308
309    def on_fail(self, test_name, begin_time):
310        self._pull_diag_logs(test_name, begin_time)
311        self._take_bug_report(test_name, begin_time)
312        self._cleanup_logger_sessions()
313
314    def on_blocked(self, test_name, begin_time):
315        self.on_fail(test_name, begin_time)
316
317    def _ad_take_extra_logs(self, ad, test_name, begin_time):
318        extra_qxdm_logs_in_seconds = self.user_params.get(
319            "extra_qxdm_logs_in_seconds", 60 * 3)
320        result = True
321        if getattr(ad, "qxdm_log", True):
322            # Gather qxdm log modified 3 minutes earlier than test start time
323            if begin_time:
324                qxdm_begin_time = begin_time - 1000 * extra_qxdm_logs_in_seconds
325            else:
326                qxdm_begin_time = None
327            try:
328                ad.get_qxdm_logs(test_name, qxdm_begin_time)
329            except Exception as e:
330                ad.log.error("Failed to get QXDM log for %s with error %s",
331                             test_name, e)
332                result = False
333
334        try:
335            ad.check_crash_report(test_name, begin_time, log_crash_report=True)
336        except Exception as e:
337            ad.log.error("Failed to check crash report for %s with error %s",
338                         test_name, e)
339            result = False
340
341        log_begin_time = getattr(
342            ad, "test_log_begin_time", None
343        ) or acts_logger.epoch_to_log_line_timestamp(begin_time - 1000 * 60)
344        log_path = os.path.join(self.log_path, test_name,
345                                "%s_%s.logcat" % (ad.serial, begin_time))
346        try:
347            ad.adb.logcat(
348                'b all -d -v year -t "%s" > %s' % (log_begin_time, log_path),
349                timeout=120)
350        except Exception as e:
351            ad.log.error("Failed to get logcat with error %s", e)
352            result = False
353        return result
354
355    def _take_bug_report(self, test_name, begin_time):
356        if self._skip_bug_report():
357            return
358        dev_num = getattr(self, "number_of_devices", None) or len(
359            self.android_devices)
360        tasks = [(self._ad_take_bugreport, (ad, test_name, begin_time))
361                 for ad in self.android_devices[:dev_num]]
362        tasks.extend([(self._ad_take_extra_logs, (ad, test_name, begin_time))
363                      for ad in self.android_devices[:dev_num]])
364        run_multithread_func(self.log, tasks)
365        for ad in self.android_devices[:dev_num]:
366            if getattr(ad, "reboot_to_recover", False):
367                reboot_device(ad)
368                ad.reboot_to_recover = False
369        if not self.user_params.get("zip_log", False): return
370        src_dir = os.path.join(self.log_path, test_name)
371        file_name = "%s_%s" % (src_dir, begin_time)
372        self.log.info("Zip folder %s to %s.zip", src_dir, file_name)
373        shutil.make_archive(file_name, "zip", src_dir)
374        shutil.rmtree(src_dir)
375
376    def _block_all_test_cases(self, tests):
377        """Over-write _block_all_test_case in BaseTestClass."""
378        for (i, (test_name, test_func)) in enumerate(tests):
379            signal = TestBlocked("Failed class setup")
380            record = records.TestResultRecord(test_name, self.TAG)
381            record.test_begin()
382            # mark all test cases as FAIL
383            record.test_fail(signal)
384            self.results.add_record(record)
385            # only gather bug report for the first test case
386            if i == 0:
387                self.on_fail(test_name, record.begin_time)
388
389    def on_pass(self, test_name, begin_time):
390        self._cleanup_logger_sessions()
391
392    def get_stress_test_number(self):
393        """Gets the stress_test_number param from user params.
394
395        Gets the stress_test_number param. If absent, returns default 100.
396        """
397        return int(self.user_params.get("stress_test_number", 100))
398