• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3.4
2#
3#   Copyright 2018 - 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    Connectivity Monitor and Telephony Troubleshooter Tests
18"""
19
20import os
21import re
22import time
23
24from acts import signals
25from acts import utils
26from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
27from acts.test_utils.tel.tel_defines import CAPABILITY_VOLTE
28from acts.test_utils.tel.tel_defines import CAPABILITY_VT
29from acts.test_utils.tel.tel_defines import CAPABILITY_WFC
30from acts.test_utils.tel.tel_defines import MAX_WAIT_TIME_FOR_STATE_CHANGE
31from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED
32from acts.test_utils.tel.tel_defines import VT_STATE_BIDIRECTIONAL
33from acts.test_utils.tel.tel_test_utils import bring_up_connectivity_monitor
34from acts.test_utils.tel.tel_test_utils import call_setup_teardown
35from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected
36from acts.test_utils.tel.tel_test_utils import fastboot_wipe
37from acts.test_utils.tel.tel_test_utils import get_device_epoch_time
38from acts.test_utils.tel.tel_test_utils import get_model_name
39from acts.test_utils.tel.tel_test_utils import get_operator_name
40from acts.test_utils.tel.tel_test_utils import hangup_call
41from acts.test_utils.tel.tel_test_utils import last_call_drop_reason
42from acts.test_utils.tel.tel_test_utils import reboot_device
43from acts.test_utils.tel.tel_test_utils import toggle_airplane_mode
44from acts.test_utils.tel.tel_test_utils import toggle_volte
45from acts.test_utils.tel.tel_test_utils import toggle_wfc
46from acts.test_utils.tel.tel_test_utils import wait_for_wfc_enabled
47from acts.test_utils.tel.tel_test_utils import wifi_toggle_state
48from acts.test_utils.tel.tel_test_utils import trigger_modem_crash
49from acts.test_utils.tel.tel_test_utils import trigger_modem_crash_by_modem
50from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
51from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
52from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb
53from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan
54from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte
55from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g
56from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_3g
57from acts.test_utils.tel.tel_voice_utils import phone_setup_csfb
58from acts.test_utils.tel.tel_voice_utils import phone_setup_iwlan
59from acts.test_utils.tel.tel_voice_utils import phone_setup_volte
60from acts.test_utils.tel.tel_video_utils import video_call_setup_teardown
61from acts.test_utils.tel.tel_video_utils import phone_setup_video
62from acts.test_utils.tel.tel_video_utils import \
63    is_phone_in_call_video_bidirectional
64
65CALL_DROP_CODE_MAPPING = {
66    373: "Radio Internal Error",
67    175: "Invalid Transaction Identifier V02",
68    159: "Temporary Failure",
69    135: "Rejected by Network V02",
70    118: "SS Not Available",
71    115: "Call Barred V02",
72    42: "Access Block V02",
73    41: "Incompatible V02"
74}
75
76CONSECUTIVE_CALL_FAILS = 5
77CALL_TROUBLE_THRESHOLD = 25
78TROUBLES = {
79    1: "WIFI_CALL_DROPS_IN_BAD_WIFI_SIGNAL",
80    2: "WIFI_CALL_DROPS_IN_GOOD_WIFI_SIGNAL_ON_SPECIFIC_WIFI_NETWORK",
81    3: "WIFI_CALL_DROPS_WITH_SPECIFIC_REASON_IN_GOOD_WIFI_SIGNAL",
82    4: "WIFI_CALL_DROPS_WITH_RANDOM_FAILURES_IN_GOOD_WIFI_SIGNAL",
83    5: "VOLTE_CALL_DROPS_IN_BAD_LTE_SIGNAL_AREAS",
84    6: "VOLTE_CALL_DROPS_IN_GOOD_LTE_SIGNAL_AREAS",
85    7: "CS_CALL_DROPS_IMS_DISABLED",
86    8: "CS_CALL_DROPS_WFC_DISABLED",
87    9: "CS_CALL_DROPS_IMS_REGISTRATION_FAILURES",
88    10: "CS_CALL_DROPS_DURING_SRVCC",
89    11: "CS_CALL_DROPS_IN_BAD_RF_CONDITIONS",
90    12: "CS_CALL_DROPS_IN_GOOD_RF_CONDITIONS_WITH_SPECIFIC_REASON",
91    13: "UNABLE_TO_TRIAGE"
92}
93
94ACTIONS = {
95    1: "CHECK_BLUETOOTH",
96    2: "CHECK_HEADSET",
97    3: "SWITCH_FROM_WIFI_PREFERRED_TO_CELLULAR_PREFERRED",
98    4: "SWITCH_FROM_CELLULAR_PREFERRED_TO_WIFI_PREFERRED",
99    5: "ENABLE_ADVANCED_4G_CALLING",
100    6: "DISABLE_ADVANCED_4G_CALLING",
101    7: "TOGGLE_AIRPLANE_MODE_TWICE",
102    8: "REBOOT_THE_PHONE",
103    9: "ENABLE_WIFI_CALLING",
104    10: "DISABLE_WIFI_CALLING",
105    11: "DISABLE_AIRPLANE_MODE",
106    12: "NONE"
107}
108
109IGNORED_CALL_DROP_REASONS = ["Radio Link Lost", "Media Timeout"]
110
111CALL_DATA_LOGS = (
112    "/data/data/com.google.android.connectivitymonitor/databases")
113
114IGNORED_CALL_DROP_TRIGGERS = ["toggle_apm", "toggle_wifi"]
115
116
117class TelLiveConnectivityMonitorBaseTest(TelephonyBaseTest):
118    def __init__(self, controllers):
119        TelephonyBaseTest.__init__(self, controllers)
120        self.user_params["enable_connectivity_metrics"] = False
121        self.user_params["telephony_auto_rerun"] = 0
122        self.consecutive_failure_limit = 5
123
124    def setup_class(self):
125        TelephonyBaseTest.setup_class(self)
126        self.dut = self.android_devices[0]
127        self.ad_reference = self.android_devices[1]
128        self.dut_model = get_model_name(self.dut)
129        self.dut_operator = get_operator_name(self.log, self.dut)
130        self.dut_capabilities = self.dut.telephony.get("capabilities", [])
131        self.dut_wfc_modes = self.dut.telephony.get("wfc_modes", [])
132        self.reference_capabilities = self.ad_reference.telephony.get(
133            "capabilities", [])
134        self.dut.log.info("DUT capabilities: %s", self.dut_capabilities)
135        self.skip_reset_between_cases = False
136        self.user_params["telephony_auto_rerun"] = 0
137        self.number_of_devices = 1
138        self.call_drop_override_code = self.user_params.get(
139            "call_drop_override_code", 373)
140
141    def setup_test(self):
142        TelephonyBaseTest.setup_test(self)
143        bring_up_connectivity_monitor(self.dut)
144        ## Work around for WFC not working issue on 2018 devices
145        if "Permissive" not in self.dut.adb.shell("su root getenforce"):
146            self.dut.adb.shell("su root setenforce 0")
147
148    def on_fail(self, test_name, begin_time):
149        self.dut.log.info("Pulling %s", CALL_DATA_LOGS)
150        log_path = os.path.join(self.dut.log_path, test_name,
151                                "ConnectivityMonitorLogs_%s" % self.dut.serial)
152        utils.create_dir(log_path)
153        self.dut.pull_files([CALL_DATA_LOGS], log_path)
154
155        self._take_bug_report(test_name, begin_time)
156
157    def teardown_test(self):
158        self.set_drop_reason_override(override_code=None)
159        TelephonyBaseTest.teardown_test(self)
160
161    def connect_to_wifi(self):
162        if not ensure_wifi_connected(self.log, self.dut,
163                                     self.wifi_network_ssid,
164                                     self.wifi_network_pass):
165            self.dut.log.error("Fail to connected to WiFi")
166            return False
167        else:
168            self.dut.log.info("Connected to WiFi")
169            return True
170
171    def is_wfc_enabled(self):
172        return wait_for_wfc_enabled(self.log, self.dut)
173
174    def enable_volte(self):
175        if CAPABILITY_VOLTE not in self.dut_capabilities:
176            raise signals.TestSkip("VoLTE is not supported, abort test.")
177        toggle_volte(self.log, self.dut, True)
178
179    def enable_wfc(self):
180        if CAPABILITY_WFC not in self.dut_capabilities:
181            raise signals.TestSkip("WFC is not supported, abort test.")
182        toggle_wfc(self.log, self.dut, True)
183
184    def disable_volte(self):
185        if CAPABILITY_VOLTE not in self.dut_capabilities:
186            raise signals.TestSkip("VoLTE is not supported, abort test.")
187        toggle_volte(self.log, self.dut, False)
188
189    def disable_wfc(self):
190        if CAPABILITY_WFC not in self.dut_capabilities:
191            raise signals.TestSkip("WFC is not supported, abort test.")
192        toggle_wfc(self.log, self.dut, False)
193
194    def setup_wfc_non_apm(self):
195        if CAPABILITY_WFC not in self.dut_capabilities and (
196                WFC_MODE_WIFI_PREFERRED not in self.dut_wfc_modes):
197            raise signals.TestSkip(
198                "WFC in non-APM is not supported, abort test.")
199        if not phone_setup_iwlan(
200                self.log, self.dut, False, WFC_MODE_WIFI_PREFERRED,
201                self.wifi_network_ssid, self.wifi_network_pass):
202            self.dut.log.error("Failed to setup WFC.")
203            raise signals.TestFailure("Failed to setup WFC in non-APM")
204        self.dut.log.info("Phone is in WFC enabled state.")
205        return True
206
207    def setup_wfc_apm(self):
208        if CAPABILITY_WFC not in self.dut_capabilities:
209            raise signals.TestSkip("WFC is not supported, abort test.")
210        if not phone_setup_iwlan(self.log, self.dut, True,
211                                 self.dut_wfc_modes[0], self.wifi_network_ssid,
212                                 self.wifi_network_pass):
213            self.dut.log.error("Failed to setup WFC.")
214            raise signals.TestFailure("Failed to setup WFC in APM")
215        self.dut.log.info("Phone is in WFC enabled state.")
216        return True
217
218    def setup_volte(self):
219        if CAPABILITY_VOLTE not in self.dut_capabilities:
220            raise signals.TestSkip("VoLTE is not supported, abort test.")
221        if not phone_setup_volte(self.log, self.dut):
222            self.dut.log.error("Phone failed to enable VoLTE.")
223            raise signals.TestFailure("Failed to enable VoLTE")
224        self.dut.log.info("Phone VOLTE is enabled successfully.")
225        return True
226
227    def setup_csfb(self):
228        if not phone_setup_csfb(self.log, self.dut):
229            self.dut.log.error("Phone failed to setup CSFB.")
230            raise signals.TestFailure("Failed to setup CSFB")
231        self.dut.log.info("Phone CSFB is enabled successfully.")
232        return True
233
234    def setup_3g(self):
235        if not phone_setup_voice_3g(self.log, self.dut):
236            self.dut.log.error("Phone failed to setup 3G.")
237            raise signals.TestFailure("Faile to setup 3G")
238        self.dut.log.info("Phone RAT 3G is enabled successfully.")
239        return True
240
241    def setup_2g(self):
242        if self.dut_operator not in ("tmo", "uk_ee"):
243            raise signals.TestSkip("2G is not supported, abort test.")
244        if not phone_setup_voice_2g(self.log, self.dut):
245            self.dut.log.error("Phone failed to setup 2G.")
246            raise signals.TestFailure("Failed to setup 2G")
247        self.dut.log.info("RAT 2G is enabled successfully.")
248        return True
249
250    def setup_vt(self):
251        if CAPABILITY_VT not in self.dut_capabilities or (
252                CAPABILITY_VT not in self.reference_capabilities):
253            raise signals.TestSkip("VT is not supported, abort test.")
254        for ad in (self.dut, self.ad_reference):
255            if not phone_setup_video(self.log, ad):
256                ad.log.error("Failed to setup VT.")
257                return False
258            return True
259
260    def set_drop_reason_override(self, override_code=None):
261        if not override_code:
262            if self.dut.adb.shell("getprop vendor.radio.call_end_reason"):
263                self.dut.adb.shell("setprop vendor.radio.call_end_reason ''")
264        else:
265            if self.dut.adb.shell("getprop vendor.radio.call_end_reason"
266                                  ) != str(override_code):
267                cmd = "setprop vendor.radio.call_end_reason %s" \
268                      % override_code
269                self.dut.log.info("====== %s ======", cmd)
270                self.dut.adb.shell(cmd)
271
272    def modem_crash(self):
273        # Modem SSR
274        self.user_params["check_crash"] = False
275        self.dut.log.info("Triggering ModemSSR")
276        try:
277            self.dut.droid.logI("======== Trigger modem crash ========")
278        except Exception:
279            pass
280        if (not self.dut.is_apk_installed("com.google.mdstest")
281            ) or self.dut.adb.getprop("ro.build.version.release")[0] in (
282                "8", "O", "7", "N") or self.dut.model in ("angler", "bullhead",
283                                                          "sailfish",
284                                                          "marlin"):
285            trigger_modem_crash(self.dut)
286        else:
287            trigger_modem_crash_by_modem(self.dut)
288
289    def call_drop_by_modem_crash(self,
290                                 call_verification_function=None,
291                                 vt=False):
292        if vt:
293            if not video_call_setup_teardown(
294                    self.log,
295                    self.dut,
296                    self.ad_reference,
297                    None,
298                    video_state=VT_STATE_BIDIRECTIONAL,
299                    verify_caller_func=is_phone_in_call_video_bidirectional,
300                    verify_callee_func=is_phone_in_call_video_bidirectional):
301                self.dut.log.error("VT Call Failed.")
302                return False
303        else:
304            if not call_setup_teardown(
305                    self.log,
306                    self.dut,
307                    self.ad_reference,
308                    ad_hangup=None,
309                    verify_caller_func=call_verification_function,
310                    wait_time_in_call=10):
311                self.log.error("Call setup failed")
312                return False
313
314        # Modem SSR
315        self.modem_crash()
316
317        try:
318            if self.dut.droid.telecomIsInCall():
319                self.dut.log.info("Still in call after trigger modem crash")
320                return False
321            else:
322                reasons = self.dut.search_logcat(
323                    "qcril_qmi_voice_map_qmi_to_ril_last_call_failure_cause")
324                if reasons:
325                    self.dut.log.info(reasons[-1]["log_message"])
326        except Exception as e:
327            self.dut.log.error(e)
328
329    def toggle_apm(self):
330        toggle_airplane_mode(self.log, self.dut, new_state=None)
331
332    def toggle_wifi(self):
333        wifi_toggle_state(self.log, self.dut, None)
334
335    def drop_reason_override(self):
336        hangup_call(self.log, self.ad_reference)
337
338    def clearn_up_bugreport_database(self):
339        self.dut.adb.shell(
340            "rm /data/data/com.google.android.connectivitymonitor/"
341            "shared_prefs/ConnectivityMonitor_BugReport.xml")
342
343    def clearn_up_troubleshooter_database(self):
344        self.dut.adb.shell(
345            "rm /data/data/com.google.android.connectivitymonitor/"
346            "shared_prefs/ConnectivityMonitor_TroubleshooterResult.xml")
347
348    def parsing_bugreport_database(self):
349        output = self.dut.adb.shell(
350            "cat /data/data/com.google.android.connectivitymonitor/"
351            "shared_prefs/ConnectivityMonitor_BugReport.xml")
352        bugreport_database = re.findall(r">Call Drop:\s+(.*)<", output)
353        self.dut.log.info("bugreport_database = %s", bugreport_database)
354        return bugreport_database
355
356    def parsing_troubleshooter_database(self):
357        output = self.dut.adb.shell(
358            "cat /data/data/com.google.android.connectivitymonitor/"
359            "shared_prefs/ConnectivityMonitor_TroubleshooterResult.xml")
360        results = re.findall(r"name=\"(\S+)\">(\S+)<", output)
361        troubleshooter_database = {}
362        for result in results:
363            if "count" in result[0] or "num_calls" in result[0]:
364                troubleshooter_database[result[0]] = int(result[1])
365            else:
366                troubleshooter_database[result[0]] = result[1]
367        self.dut.log.info("TroubleshooterResult=%s",
368                          sorted(troubleshooter_database.items()))
369        return troubleshooter_database
370
371    def parsing_call_summary(self):
372        call_summary = self.dut.adb.shell(
373            "dumpsys activity service com.google.android.connectivitymonitor/"
374            ".ConnectivityMonitorService")
375        self.dut.log.info(call_summary)
376        call_summary_info = {}
377        results = re.findall(
378            r"(\S+): (\d+) out of (\d+) calls dropped, percentage=(\S+)",
379            call_summary)
380        for result in results:
381            call_summary_info[result[0]] = int(result[2])
382            call_summary_info["%s_dropped" % result[0]] = int(result[1])
383            if result[3] == "NaN":
384                call_summary_info["%s_dropped_percentage" % result[0]] = 0
385            else:
386                call_summary_info["%s_dropped_percentage" % result[0]] = float(
387                    result[3])
388        results = re.findall(r"(\S+): predominant failure reason=(.+)",
389                             call_summary)
390        for result in results:
391            call_summary_info["%s_failure_reason" % result[0]] = result[1]
392        self.dut.log.info("call summary dumpsys = %s",
393                          sorted(call_summary_info.items()))
394        return call_summary_info
395
396    def parsing_call_statistics(self):
397        call_statistics_info = {}
398        call_statistics = self.dut.adb.shell(
399            "content query --uri content://com.google.android."
400            "connectivitymonitor.troubleshooterprovider/call_statistics")
401        self.dut.log.info("troubleshooterprovider call_statistics:\n%s",
402                          call_statistics)
403        results = re.findall(r"KEY=(\S+), VALUE=(\S+)", call_statistics)
404        for result in results:
405            if ("count" in result[0] or "num_calls" in result[0]):
406                if result[1] == "NULL":
407                    call_statistics_info[result[0]] = 0
408                else:
409                    call_statistics_info[result[0]] = int(result[1])
410            else:
411                call_statistics_info[result[0]] = result[1]
412        self.dut.log.info("troubleshooterprovider call_statistics: %s",
413                          sorted(call_statistics_info.items()))
414        return call_statistics_info
415
416    def parsing_diagnostics(self):
417        diagnostics_info = {}
418        diagnostics = self.dut.adb.shell(
419            "content query --uri content://com.google.android."
420            "connectivitymonitor.troubleshooterprovider/diagnostics")
421        self.dut.log.info("troubleshooterprovider diagnostics:\n%s",
422                          diagnostics)
423        results = re.findall(r"KEY=(\S+), VALUE=(\S+)", diagnostics)
424        for result in results:
425            diagnostics_info[result[0]] = result[1]
426        self.dut.log.info("troubleshooterprovider diagnostics: %s",
427                          sorted(diagnostics_info.items()))
428        return diagnostics_info
429
430    def call_setup_and_connectivity_monitor_checking(self,
431                                                     setup=None,
432                                                     handover=None,
433                                                     triggers=[],
434                                                     expected_drop_reason="",
435                                                     expected_trouble=None,
436                                                     expected_action=None):
437
438        call_verification_function = None
439        begin_time = get_device_epoch_time(self.dut)
440        call_data_summary_before = self.parsing_call_summary()
441        call_statistics_before = self.parsing_call_statistics()
442        self.parsing_diagnostics()
443        self.parsing_troubleshooter_database()
444        bugreport_database_before = self.parsing_bugreport_database()
445
446        if expected_drop_reason:
447            expected_drop_reasons = set(expected_drop_reason.split("|"))
448        else:
449            expected_drop_reasons = set()
450        checking_counters = ["Calls"]
451        checking_reasons = []
452        result = True
453        if setup in ("wfc_apm", "wfc_non_apm"):
454            call_verification_function = is_phone_in_call_iwlan
455        elif setup == "volte":
456            call_verification_function = is_phone_in_call_volte
457        elif setup == "csfb":
458            call_verification_function = is_phone_in_call_csfb
459        elif setup == "3g":
460            call_verification_function = is_phone_in_call_3g
461        elif setup == "2g":
462            call_verification_function = is_phone_in_call_2g
463        technology = handover or setup
464        if technology in ("wfc_apm", "wfc_non_apm"):
465            if triggers and triggers[0] not in IGNORED_CALL_DROP_TRIGGERS:
466                checking_counters.extend(
467                    ["Calls_dropped", "VOWIFI", "VOWIFI_dropped"])
468                checking_reasons.append("VOWIFI_failure_reason")
469            elif call_data_summary_before.get("Calls_dropped", 0):
470                checking_counters.append("VOWIFI")
471        elif technology == "volte":
472            if triggers and triggers[0] not in IGNORED_CALL_DROP_TRIGGERS:
473                checking_counters.extend(
474                    ["Calls_dropped", "VOLTE", "VOLTE_dropped"])
475                checking_reasons.append("VOLTE_failure_reason")
476            elif call_data_summary_before.get("Calls_dropped", 0):
477                checking_counters.append("VOLTE")
478        elif technology in ("csfb", "3g", "2g"):
479            if triggers and triggers[0] not in IGNORED_CALL_DROP_TRIGGERS:
480                checking_counters.extend(["Calls_dropped", "CS", "CS_dropped"])
481                checking_reasons.append("CS_failure_reason")
482            elif call_data_summary_before.get("Calls_dropped", 0):
483                checking_counters.append("CS")
484
485        if setup == "vt":
486            if not video_call_setup_teardown(
487                    self.log,
488                    self.dut,
489                    self.ad_reference,
490                    None,
491                    video_state=VT_STATE_BIDIRECTIONAL,
492                    verify_caller_func=is_phone_in_call_video_bidirectional,
493                    verify_callee_func=is_phone_in_call_video_bidirectional):
494                raise signals.TestFailure("VT Call Failed.")
495        else:
496            if not call_setup_teardown(
497                    self.log,
498                    self.dut,
499                    self.ad_reference,
500                    ad_hangup=None,
501                    verify_caller_func=call_verification_function,
502                    wait_time_in_call=10):
503                raise signals.TestFailure("Call Setup Failed.")
504
505        for trigger in triggers:
506            if self.dut.droid.telecomIsInCall():
507                self.dut.log.info("Telecom is in call")
508                self.dut.log.info(
509                    "Voice in RAT %s",
510                    self.dut.droid.telephonyGetCurrentVoiceNetworkType())
511            else:
512                self.dut.log.info("Not in call")
513            # Trigger in-call event
514            if trigger and getattr(self, trigger, None):
515                trigger_func = getattr(self, trigger)
516                trigger_func()
517                time.sleep(MAX_WAIT_TIME_FOR_STATE_CHANGE)
518
519        if self.dut.droid.telecomIsInCall():
520            self.dut.log.info("Telecom is in call")
521            self.dut.log.info(
522                "Voice in RAT %s",
523                self.dut.droid.telephonyGetCurrentVoiceNetworkType())
524        else:
525            self.dut.log.info("Not in call")
526
527        if self.dut.droid.telecomIsInCall():
528            self.dut.log.info("Telecom is in call")
529            self.dut.log.info(
530                "Voice in RAT %s",
531                self.dut.droid.telephonyGetCurrentVoiceNetworkType())
532        else:
533            self.dut.log.info("Not in call")
534
535        drop_reason = last_call_drop_reason(self.dut, begin_time)
536        drop_reason = drop_reason.title()
537        if drop_reason:
538            expected_drop_reasons.add(drop_reason)
539        for ad in (self.ad_reference, self.dut):
540            try:
541                if ad.droid.telecomIsInCall():
542                    if triggers:
543                        ad.log.info("Still in call after triggers %s",
544                                    triggers)
545                        result = False
546                    hangup_call(self.log, ad)
547                    time.sleep(MAX_WAIT_TIME_FOR_STATE_CHANGE)
548            except Exception as e:
549                ad.log.error(e)
550
551        call_data_summary_after = self.parsing_call_summary()
552        call_statistics_after = self.parsing_call_statistics()
553        diagnostics_after = self.parsing_diagnostics()
554        ts_database_after = self.parsing_troubleshooter_database()
555
556        for counter in checking_counters:
557            if call_data_summary_after.get(
558                    counter,
559                    0) != call_data_summary_before.get(counter, 0) + 1:
560                self.dut.log.error("Counter %s did not increase", counter)
561                result = False
562            else:
563                self.dut.log.info("Counter %s increased", counter)
564            if counter == "Calls":
565                if call_statistics_after.get("num_calls",
566                                             0) - call_statistics_before.get(
567                                                 "num_calls", 0) < 1:
568                    self.dut.log.warning(
569                        "call_statistics num_calls didn't increase")
570                    # result = False
571                else:
572                    self.dut.log.info("call_statistics num_calls increased")
573            if "_dropped" in counter and counter != "Calls_dropped":
574                desc = counter.split("_")[0]
575                if desc == "VOWIFI":
576                    stat_key = "recent_wfc_fail_count"
577                else:
578                    stat_key = "recent_%s_fail_count" % desc.lower()
579                before = call_statistics_after.get(stat_key, 0)
580                after = call_statistics_after.get(stat_key, 0)
581                most_failure_call_type = call_statistics_after.get(
582                    "call_type_with_most_failures")
583                diagnosis = diagnostics_after.get("diagnosis")
584                actions = diagnostics_after.get("actions")
585                if after - before < 1:
586                    self.dut.log.warning("call_statistics %s didn't increase, "
587                                         "before %s, after %s" %
588                                         (stat_key, before, after))
589                    # result = False
590                else:
591                    self.dut.log.info("call_statistics %s increased", stat_key)
592                if most_failure_call_type != desc:
593                    self.dut.log.warning(
594                        "call_statistics call_type_with_most_failures "
595                        "is %s, not %s", most_failure_call_type, desc)
596                else:
597                    self.dut.log.info(
598                        "call_statistics call_type_with_most_failures is %s",
599                        most_failure_call_type)
600                dropped = call_data_summary_after.get("%s_dropped" % desc, 0)
601                drop_percentage = call_data_summary_after.get(
602                    "%s_dropped_percentage" % desc, 0)
603                self.dut.log.info("%s_dropped = %s, percentage = %s", desc,
604                                  dropped, drop_percentage)
605                if expected_trouble and expected_trouble != diagnosis:
606                    self.dut.log.warning("diagnoisis = %s, expecting %s",
607                                         diagnosis, expected_trouble)
608                if expected_action and expected_action != actions:
609                    self.dut.log.error("actions = %s, expecting %s", actions,
610                                       expected_action)
611                    result = False
612                if drop_percentage > CALL_TROUBLE_THRESHOLD and (
613                        dropped > CONSECUTIVE_CALL_FAILS):
614                    if diagnosis == "UNABLE_TO_TRIAGE":
615                        self.dut.log.error(
616                            "troubleshooter diagnosis is %s with %s dropped "
617                            "and %s drop_percentage", diagnosis, dropped,
618                            drop_percentage)
619                        result = False
620                    if actions == "NONE":
621                        self.dut.log.error(
622                            "troubleshooter failed to provide suggestion, "
623                            "actions = %s", actions)
624                        result = False
625        if expected_drop_reasons:
626            expected_drop_reason = "|".join(expected_drop_reasons)
627        for reason_key in checking_reasons:
628            if call_data_summary_after.get(reason_key, None):
629                drop_reason = call_data_summary_after[reason_key]
630                if expected_drop_reason and drop_reason not in expected_drop_reason:
631                    self.dut.log.error("%s is: %s, expecting %s", reason_key,
632                                       drop_reason, expected_drop_reason)
633                    result = False
634                else:
635                    self.dut.log.info("%s is: %s as expected", reason_key,
636                                      drop_reason)
637            else:
638                self.dut.log.error("%s is not provided in summary report",
639                                   reason_key)
640                result = False
641
642        if not triggers or triggers[0] in IGNORED_CALL_DROP_TRIGGERS:
643            return result
644        if drop_reason in bugreport_database_before:
645            self.dut.log.info("%s is in bugreport database %s before call",
646                              drop_reason, bugreport_database_before)
647            return result
648        else:
649            self.dut.log.info("%s is not in bugreport database %s before call",
650                              drop_reason, bugreport_database_before)
651        if drop_reason in IGNORED_CALL_DROP_REASONS:
652            self.dut.log.info(
653                "Call drop with reason %s will skip bugreport notification",
654                drop_reason)
655            return result
656        else:
657            self.dut.log.info(
658                "Call drop %s should generate bugreport notification",
659                drop_reason)
660        # Parse logcat for UI notification only for the first failure
661        if self.dut.search_logcat("Bugreport notification title Call Drop:",
662                                  begin_time):
663            self.dut.log.info(
664                "Bugreport notification title Call Drop is seen in logcat")
665            return result
666        else:
667            self.dut.log.error(
668                "Bugreport notification title Call Drop is not seen in logcat")
669            return False
670
671    def call_drop_test(self,
672                       setup=None,
673                       handover=None,
674                       count=CONSECUTIVE_CALL_FAILS,
675                       triggers=[],
676                       expected_drop_reason=None,
677                       expected_trouble=None,
678                       expected_action=None):
679        if not triggers:
680            if self.dut.model in ("marlin", "sailfish", "walleye", "taimen"):
681                triggers = ["modem_crash"]
682                expected_drop_reason = "Error Unspecified"
683            else:
684                triggers = ["drop_reason_override"]
685        if "drop_reason_override" in triggers:
686            self.set_drop_reason_override(
687                override_code=self.call_drop_override_code)
688            expected_drop_reason = CALL_DROP_CODE_MAPPING[int(
689                self.call_drop_override_code)]
690        for iter in range(count):
691            self.dut.log.info("===== %s_iter_%s =====", self.test_name,
692                              iter + 1)
693            if iter < count - 1:
694                action = None
695                trouble = None
696            else:
697                action = expected_action
698                trouble = expected_trouble
699            if not self.call_setup_and_connectivity_monitor_checking(
700                    setup=setup,
701                    handover=handover,
702                    triggers=triggers,
703                    expected_drop_reason=expected_drop_reason,
704                    expected_trouble=trouble,
705                    expected_action=action):
706                return False
707        return True
708
709    def call_drop_triggered_suggestion_test(self,
710                                            setup=None,
711                                            handover=None,
712                                            triggers=[],
713                                            expected_drop_reason=None,
714                                            expected_trouble=None,
715                                            expected_action=None):
716        call_summary = self.parsing_call_summary()
717        diagnostics = self.parsing_diagnostics()
718        diagnosis = diagnostics.get("diagnosis")
719        actions = diagnostics.get("actions")
720        self.dut.log.info("Expected trouble = %s, action = %s",
721                          expected_trouble, expected_action)
722        if expected_trouble and diagnosis == expected_trouble and not handover:
723            self.dut.log.info("Diagnosis is the expected %s", trouble)
724            if expected_action and expected_action != actions:
725                self.dut.log.error("Action is %s, expecting %s", actions,
726                                   expected_action)
727                result = False
728            if setup in ("wfc_apm", "wfc_non_apm"):
729                desc = "VOWIFI"
730            elif setup == "volte":
731                desc = "VOLTE"
732            elif setup in ("csfb", "3g", "2g"):
733                desc = "CS"
734            drops = call_summary.get("%s_dropped" % desc, 0)
735            drop_percentage = call_summary.get("%s_dropped_percentage" % desc,
736                                               0)
737            if drops < CONSECUTIVE_CALL_FAILS or drop_percentage < 25:
738                self.dut.log.error(
739                    "Should NOT get %s for %s %s_dropped and %s %s_dropped_percentage",
740                    trouble, drops, desc, drop_percentage, desc)
741                return False
742            else:
743                return True
744        else:
745            self.dut.log.info("Generate %s consecutive call drops",
746                              CONSECUTIVE_CALL_FAILS)
747            return self.call_drop_test(
748                setup=setup,
749                handover=handover,
750                count=CONSECUTIVE_CALL_FAILS,
751                triggers=triggers,
752                expected_drop_reason=expected_drop_reason,
753                expected_trouble=expected_trouble,
754                expected_action=expected_action)
755
756    def healthy_call_test(self,
757                          setup=None,
758                          handover=None,
759                          count=1,
760                          triggers=[],
761                          expected_trouble=None,
762                          expected_action=None):
763        if self.dut.model not in ("marlin", "sailfish", "walleye", "taimen"):
764            self.set_drop_reason_override(override_code=25)
765        result = True
766        for iter in range(count):
767            if not self.call_setup_and_connectivity_monitor_checking(
768                    setup=setup,
769                    handover=handover,
770                    triggers=triggers,
771                    expected_trouble=expected_trouble,
772                    expected_action=expected_action):
773                return False
774        return True
775
776    def forced_call_drop_test(self,
777                              setup=None,
778                              handover=None,
779                              triggers=None,
780                              expected_drop_reason=None):
781        expected_trouble = None
782        expected_action = None
783        technology = handover or setup
784        if setup:
785            setup_func = getattr(self, "setup_%s" % setup)
786            if not setup_func(): return False
787            if technology == "volte":
788                expected_trouble = TROUBLES[6],
789                expected_action = ACTIONS[6]
790            elif technology == "csfb":
791                if CAPABILITY_VOLTE in self.dut_capabilities:
792                    expected_action = ACTIONS[5]
793                else:
794                    expected_action = ACTIONS[7]
795                expected_trouble = TROUBLES[7]
796            elif technology == "3g":
797                if CAPABILITY_VOLTE in self.dut_capabilities:
798                    expected_action = ACTIONS[5]
799                else:
800                    expected_action = ACTIONS[7]
801                expected_trouble = TROUBLES[7]
802            elif technology == "2g":
803                if CAPABILITY_VOLTE in self.dut_capabilities:
804                    expected_action = ACTIONS[5]
805                else:
806                    expected_action = ACTIONS[7]
807                expected_trouble = TROUBLES[7]
808            elif technology == "wfc_apm":
809                expected_trouble = TROUBLES[3]
810                expected_action = ACTIONS[11]
811            elif technology == "wfc_non_apm":
812                expected_trouble = TROUBLES[3]
813                expected_action = ACTIONS[3]
814
815        return self.call_drop_triggered_suggestion_test(
816            setup=setup,
817            handover=handover,
818            triggers=triggers,
819            expected_drop_reason=expected_drop_reason,
820            expected_trouble=expected_trouble,
821            expected_action=expected_action)
822
823    def call_drop_test_after_wipe(self, setup=None):
824        if setup:
825            setup_func = getattr(self, "setup_%s" % setup)
826            if not setup_func(): return False
827        fastboot_wipe(self.dut)
828        bring_up_connectivity_monitor(self.dut)
829        return self.forced_call_drop_test(setup=setup)
830
831    def call_drop_test_after_reboot(self, setup=None):
832        self.forced_call_drop_test(setup=setup)
833        self.healthy_call_test(setup=setup, count=1)
834        reboot_device(self.dut)
835        return self.forced_call_drop_test(setup=setup)
836