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.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
26from acts.test_utils.tel.tel_defines import CAPABILITY_VOLTE
27from acts.test_utils.tel.tel_defines import CAPABILITY_VT
28from acts.test_utils.tel.tel_defines import CAPABILITY_WFC
29from acts.test_utils.tel.tel_defines import MAX_WAIT_TIME_FOR_STATE_CHANGE
30from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED
31from acts.test_utils.tel.tel_defines import VT_STATE_BIDIRECTIONAL
32from acts.test_utils.tel.tel_test_utils import bring_up_connectivity_monitor
33from acts.test_utils.tel.tel_test_utils import call_setup_teardown
34from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected
35from acts.test_utils.tel.tel_test_utils import fastboot_wipe
36from acts.test_utils.tel.tel_test_utils import get_device_epoch_time
37from acts.test_utils.tel.tel_test_utils import get_model_name
38from acts.test_utils.tel.tel_test_utils import get_operator_name
39from acts.test_utils.tel.tel_test_utils import get_outgoing_voice_sub_id
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 setup_class(self):
119        TelephonyBaseTest.setup_class(self)
120        self.user_params["enable_connectivity_metrics"] = False
121        self.user_params["telephony_auto_rerun"] = 0
122        self.consecutive_failure_limit = 5
123
124        self.dut = self.android_devices[0]
125        self.ad_reference = self.android_devices[1]
126        self.dut_model = get_model_name(self.dut)
127        self.dut_operator = get_operator_name(self.log, self.dut)
128        self.dut_subID = get_outgoing_voice_sub_id(self.dut)
129        self.dut_capabilities = self.dut.telephony["subscription"][self.dut_subID].get("capabilities", [])
130        self.dut_wfc_modes = self.dut.telephony["subscription"][self.dut_subID].get("wfc_modes", [])
131        self.ad_reference_subID = get_outgoing_voice_sub_id(self.ad_reference)
132        self.reference_capabilities = self.ad_reference.telephony["subscription"][self.ad_reference_subID].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        os.makedirs(log_path, exist_ok=True)
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