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    Test Script for Telephony Pre Check In Sanity
18"""
19
20import collections
21import time
22import os
23import re
24
25from acts import signals
26from acts.utils import unzip_maintain_permissions
27from acts.utils import exe_cmd
28from acts.controllers.android_device import SL4A_APK_NAME
29from acts.controllers.android_device import list_adb_devices
30from acts.controllers.android_device import list_fastboot_devices
31from acts.test_decorators import test_tracker_info
32from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
33from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_FOR_BOOT_COMPLETE
34from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_FOR_CARRIERCONFIG_CHANGE
35from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_FOR_CARRIERID_CHANGE
36from acts_contrib.test_utils.tel.tel_defines import VZW_CARRIER_CONFIG_VERSION
37from acts_contrib.test_utils.tel.tel_defines import ATT_CARRIER_CONFIG_VERSION
38from acts_contrib.test_utils.tel.tel_defines import CARRIER_ID_METADATA_URL
39from acts_contrib.test_utils.tel.tel_defines import CARRIER_ID_CONTENT_URL
40from acts_contrib.test_utils.tel.tel_defines import CARRIER_ID_VERSION
41from acts_contrib.test_utils.tel.tel_defines import ER_DB_ID_VERSION
42from acts_contrib.test_utils.tel.tel_defines import WAIT_TIME_FOR_ER_DB_CHANGE
43from acts_contrib.test_utils.tel.tel_defines import CARRIER_ID_METADATA_URL_P
44from acts_contrib.test_utils.tel.tel_defines import CARRIER_ID_CONTENT_URL_P
45from acts_contrib.test_utils.tel.tel_defines import CARRIER_ID_VERSION_P
46from acts_contrib.test_utils.tel.tel_lookup_tables import device_capabilities
47from acts_contrib.test_utils.tel.tel_lookup_tables import operator_capabilities
48from acts_contrib.test_utils.tel.tel_test_utils import lock_lte_band_by_mds
49from acts_contrib.test_utils.tel.tel_test_utils import get_model_name
50from acts_contrib.test_utils.tel.tel_test_utils import get_operator_name
51from acts_contrib.test_utils.tel.tel_test_utils import reboot_device
52from acts_contrib.test_utils.tel.tel_test_utils import toggle_airplane_mode
53from acts_contrib.test_utils.tel.tel_test_utils import trigger_modem_crash_by_modem
54from acts_contrib.test_utils.tel.tel_test_utils import bring_up_sl4a
55from acts_contrib.test_utils.tel.tel_test_utils import fastboot_wipe
56from acts_contrib.test_utils.tel.tel_test_utils import get_carrier_config_version
57from acts_contrib.test_utils.tel.tel_test_utils import get_carrier_id_version
58from acts_contrib.test_utils.tel.tel_test_utils import get_er_db_id_version
59from acts_contrib.test_utils.tel.tel_test_utils import get_database_content
60from acts_contrib.test_utils.tel.tel_test_utils import install_googleaccountutil_apk
61from acts_contrib.test_utils.tel.tel_test_utils import add_whitelisted_account
62from acts_contrib.test_utils.tel.tel_test_utils import adb_disable_verity
63from acts_contrib.test_utils.tel.tel_test_utils import install_carriersettings_apk
64from acts_contrib.test_utils.tel.tel_test_utils import ensure_wifi_connected
65from acts_contrib.test_utils.tel.tel_test_utils import cleanup_configupdater
66from acts_contrib.test_utils.tel.tel_test_utils import pull_carrier_id_files
67from acts_contrib.test_utils.tel.tel_test_utils import wifi_toggle_state
68from acts_contrib.test_utils.tel.tel_voice_utils import phone_setup_volte
69from acts_contrib.test_utils.tel.tel_subscription_utils import get_cbrs_and_default_sub_id
70from acts.utils import get_current_epoch_time
71from acts.keys import Config
72
73class TelLiveNoQXDMLogTest(TelephonyBaseTest):
74    def setup_class(self):
75        super().setup_class()
76        self.dut = self.android_devices[0]
77        if len(self.android_devices) > 1:
78            self.ad_reference = self.android_devices[1]
79            setattr(self.ad_reference, "qxdm_log", False)
80        else:
81            self.ad_reference = None
82        setattr(self.dut, "qxdm_log", False)
83        self.stress_test_number = int(
84            self.user_params.get("stress_test_number", 5))
85        self.skip_reset_between_cases = False
86        self.dut_model = get_model_name(self.dut)
87        self.dut_operator = get_operator_name(self.log, self.dut)
88        self.dut_capabilities = set(
89            device_capabilities.get(
90                self.dut_model, device_capabilities["default"])) & set(
91                    operator_capabilities.get(
92                        self.dut_operator, operator_capabilities["default"]))
93        self.dut.log.info("DUT capabilities: %s", self.dut_capabilities)
94        self.user_params["check_crash"] = False
95        self.skip_reset_between_cases = False
96        self.path = self.user_params['account_util'][0]
97        self.user_account = "commsteltest1"
98        self.user_password = "tellivetest2"
99
100    def _get_list_average(self, input_list):
101        total_sum = float(sum(input_list))
102        total_count = float(len(input_list))
103        if input_list == []:
104            return False
105        return float(total_sum / total_count)
106
107    def _telephony_bootup_time_test(self):
108        """Telephony Bootup Perf Test
109
110        Arguments:
111            check_lte_data: whether to check the LTE data.
112            check_volte: whether to check Voice over LTE.
113            check_wfc: whether to check Wifi Calling.
114
115        Expected Results:
116            Time
117
118        Returns:
119            True is pass, False if fail.
120        """
121        self.number_of_devices = 1
122        ad = self.dut
123        toggle_airplane_mode(self.log, ad, False)
124        if not phone_setup_volte(self.log, ad):
125            ad.log.error("Failed to setup VoLTE.")
126            return False
127        fail_count = collections.defaultdict(int)
128        test_result = True
129        keyword_time_dict = {}
130
131        text_search_mapping = {
132            'boot_complete': "ModemService: Received: android.intent.action.BOOT_COMPLETED",
133            'Voice_Reg': "< VOICE_REGISTRATION_STATE {.regState = REG_HOME",
134            'Data_Reg': "< DATA_REGISTRATION_STATE {.regState = REG_HOME",
135            'Data_Call_Up': "onSetupConnectionCompleted result=SUCCESS",
136            'VoLTE_Enabled': "isVolteEnabled=true",
137        }
138
139        text_obj_mapping = {
140            "boot_complete": None,
141            "Voice_Reg": None,
142            "Data_Reg": None,
143            "Data_Call_Up": None,
144            "VoLTE_Enabled": None,
145        }
146        blocked_for_calculate = ["boot_complete"]
147        for i in range(1, self.stress_test_number + 1):
148            ad.log.info("Telephony Bootup Time Test %s Iteration: %d / %d",
149                        self.test_name, i, self.stress_test_number)
150            begin_time = get_current_epoch_time()
151            ad.log.debug("Begin Time is %s", begin_time)
152            ad.log.info("reboot!")
153            reboot_device(ad)
154            iteration_result = "pass"
155
156            time.sleep(WAIT_TIME_FOR_BOOT_COMPLETE)
157
158            dict_match = ad.search_logcat(
159                text_search_mapping['boot_complete'], begin_time=begin_time)
160            if len(dict_match) != 0:
161                text_obj_mapping['boot_complete'] = dict_match[0][
162                    'datetime_obj']
163                ad.log.debug("Datetime for boot_complete is %s",
164                             text_obj_mapping['boot_complete'])
165                bootup_time = dict_match[0]['datetime_obj'].strftime('%s')
166                bootup_time = int(bootup_time) * 1000
167                ad.log.info("Bootup Time is %d", bootup_time)
168            else:
169                ad.log.error("TERMINATE- boot_complete not seen in logcat")
170                return False
171
172            for tel_state in text_search_mapping:
173                if tel_state == "boot_complete":
174                    continue
175                dict_match = ad.search_logcat(
176                    text_search_mapping[tel_state], begin_time=bootup_time)
177                if len(dict_match) != 0:
178                    text_obj_mapping[tel_state] = dict_match[0]['datetime_obj']
179                    ad.log.debug("Datetime for %s is %s", tel_state,
180                                 text_obj_mapping[tel_state])
181                else:
182                    ad.log.error("Cannot Find Text %s in logcat",
183                                 text_search_mapping[tel_state])
184                    blocked_for_calculate.append(tel_state)
185                    ad.log.debug("New Blocked %s", blocked_for_calculate)
186
187            ad.log.info("List Blocked %s", blocked_for_calculate)
188            for tel_state in text_search_mapping:
189                if tel_state not in blocked_for_calculate:
190                    time_diff = text_obj_mapping[tel_state] - \
191                                text_obj_mapping['boot_complete']
192                    ad.log.info("Time Diff is %d for %s", time_diff.seconds,
193                                tel_state)
194                    if tel_state in keyword_time_dict:
195                        keyword_time_dict[tel_state].append(time_diff.seconds)
196                    else:
197                        keyword_time_dict[tel_state] = [
198                            time_diff.seconds,
199                        ]
200                    ad.log.debug("Keyword Time Dict %s", keyword_time_dict)
201
202            ad.log.info("Telephony Bootup Time Test %s Iteration: %d / %d %s",
203                        self.test_name, i, self.stress_test_number,
204                        iteration_result)
205        ad.log.info("Final Keyword Time Dict %s", keyword_time_dict)
206        for tel_state in text_search_mapping:
207            if tel_state not in blocked_for_calculate:
208                avg_time = self._get_list_average(keyword_time_dict[tel_state])
209                if avg_time < 12.0:
210                    ad.log.info("Average %s for %d iterations = %.2f seconds",
211                                tel_state, self.stress_test_number, avg_time)
212                else:
213                    ad.log.error("Average %s for %d iterations = %.2f seconds",
214                                 tel_state, self.stress_test_number, avg_time)
215                    fail_count[tel_state] += 1
216
217        ad.log.info("Bootup Time Dict: %s", keyword_time_dict)
218        ad.log.info("fail_count: %s", dict(fail_count))
219        for failure, count in fail_count.items():
220            if count:
221                ad.log.error("%s %s failures in %s iterations", count, failure,
222                             self.stress_test_number)
223                test_result = False
224        return test_result
225
226    def _cbrs_bootup_time_test(self):
227        """CBRS Bootup Perf Test
228
229        Expected Results:
230            Time
231
232        Returns:
233            True is pass, False if fail.
234        """
235        self.number_of_devices = 1
236        ad = self.dut
237        cbrs_subid, default_subid = get_cbrs_and_default_sub_id(ad)
238        toggle_airplane_mode(self.log, ad, False)
239
240        fail_count = collections.defaultdict(int)
241        test_result = True
242        keyword_time_dict = {}
243
244        text_search_mapping = {
245            'boot_complete': "ModemService: Received: android.intent.action.BOOT_COMPLETED",
246            'cbrs_active': "notifyPreferredDataSubIdChanged to %s" % cbrs_subid,
247        }
248
249        text_obj_mapping = {
250            "boot_complete": None,
251            "cbrs_active": None,
252        }
253        blocked_for_calculate = ["boot_complete"]
254        for i in range(1, self.stress_test_number + 1):
255            ad.log.info("CBRS Bootup Time Test %s Iteration: %d / %d",
256                        self.test_name, i, self.stress_test_number)
257            begin_time = get_current_epoch_time()
258            ad.log.debug("Begin Time is %s", begin_time)
259            ad.log.info("reboot!")
260            reboot_device(ad)
261            iteration_result = "pass"
262
263            time.sleep(WAIT_TIME_FOR_BOOT_COMPLETE)
264
265            dict_match = ad.search_logcat(
266                text_search_mapping['boot_complete'], begin_time=begin_time)
267            if len(dict_match) != 0:
268                text_obj_mapping['boot_complete'] = dict_match[0][
269                    'datetime_obj']
270                ad.log.debug("Datetime for boot_complete is %s",
271                             text_obj_mapping['boot_complete'])
272                bootup_time = dict_match[0]['datetime_obj'].strftime('%s')
273                bootup_time = int(bootup_time) * 1000
274                ad.log.info("Bootup Time is %d", bootup_time)
275            else:
276                ad.log.error("TERMINATE- boot_complete not seen in logcat")
277                return False
278
279            for tel_state in text_search_mapping:
280                if tel_state == "boot_complete":
281                    continue
282                dict_match = ad.search_logcat(
283                    text_search_mapping[tel_state], begin_time=bootup_time)
284                if len(dict_match) != 0:
285                    text_obj_mapping[tel_state] = dict_match[0]['datetime_obj']
286                    ad.log.debug("Datetime for %s is %s", tel_state,
287                                 text_obj_mapping[tel_state])
288                else:
289                    ad.log.error("Cannot Find Text %s in logcat",
290                                 text_search_mapping[tel_state])
291                    blocked_for_calculate.append(tel_state)
292                    ad.log.debug("New Blocked %s", blocked_for_calculate)
293
294            ad.log.info("List Blocked %s", blocked_for_calculate)
295            for tel_state in text_search_mapping:
296                if tel_state not in blocked_for_calculate:
297                    time_diff = text_obj_mapping[tel_state] - \
298                                text_obj_mapping['boot_complete']
299                    ad.log.info("Time Diff is %d for %s", time_diff.seconds,
300                                tel_state)
301                    if tel_state in keyword_time_dict:
302                        keyword_time_dict[tel_state].append(time_diff.seconds)
303                    else:
304                        keyword_time_dict[tel_state] = [
305                            time_diff.seconds,
306                        ]
307                    ad.log.debug("Keyword Time Dict %s", keyword_time_dict)
308
309            ad.log.info("CBRS Bootup Time Test %s Iteration: %d / %d %s",
310                        self.test_name, i, self.stress_test_number,
311                        iteration_result)
312        ad.log.info("Final Keyword Time Dict %s", keyword_time_dict)
313        for tel_state in text_search_mapping:
314            if tel_state not in blocked_for_calculate:
315                avg_time = self._get_list_average(keyword_time_dict[tel_state])
316                if avg_time < 12.0:
317                    ad.log.info("Average %s for %d iterations = %.2f seconds",
318                                tel_state, self.stress_test_number, avg_time)
319                else:
320                    ad.log.error("Average %s for %d iterations = %.2f seconds",
321                                 tel_state, self.stress_test_number, avg_time)
322                    fail_count[tel_state] += 1
323
324        ad.log.info("Bootup Time Dict: %s", keyword_time_dict)
325        ad.log.info("fail_count: %s", dict(fail_count))
326        for failure, count in fail_count.items():
327            if count:
328                ad.log.error("%s %s failures in %s iterations", count, failure,
329                             self.stress_test_number)
330                test_result = False
331        return test_result
332
333    """ Tests Begin """
334
335    @test_tracker_info(uuid="109d59ff-a488-4a68-87fd-2d8d0c035326")
336    @TelephonyBaseTest.tel_test_wrap
337    def test_bootup_optimized_stress(self):
338        """Bootup Optimized Reliability Test
339
340        Steps:
341            1. Reboot DUT.
342            2. Parse logcat for time taken by Voice, Data, VoLTE
343            3. Repeat Step 1~2 for N times. (before reboot)
344
345        Expected Results:
346            No crash happens in stress test.
347
348        Returns:
349            True is pass, False if fail.
350        """
351        return self._telephony_bootup_time_test()
352
353    @test_tracker_info(uuid="d29e6e62-3d54-4a58-b67f-2ba0de3d0a19")
354    @TelephonyBaseTest.tel_test_wrap
355    def test_bootup_cbrs_stress(self):
356        """Bootup Optimized Reliability Test
357
358        Steps:
359            1. Reboot DUT.
360            2. Parse logcat for time taken by CBRS data
361            3. Repeat Step 1~2 for N times. (before reboot)
362
363        Expected Results:
364            No crash happens in stress test.
365
366        Returns:
367            True is pass, False if fail.
368        """
369        return self._cbrs_bootup_time_test()
370
371    @test_tracker_info(uuid="67f50d11-a987-4e79-9a20-1569d365511b")
372    @TelephonyBaseTest.tel_test_wrap
373    def test_modem_power_anomaly_file_existence(self):
374        """Verify if the power anomaly file exists
375
376        1. Collect Bugreport
377        2. unzip bugreport
378        3. remane the .bin file to .tar
379        4. unzip dumpstate.tar
380        5. Verify if the file exists
381
382        """
383        ad = self.android_devices[0]
384        cmd = ("am broadcast -a "
385               "com.google.gservices.intent.action.GSERVICES_OVERRIDE "
386               "-e \"ce.cm.power_anomaly_data_enable\" \"true\"")
387        ad.adb.shell(cmd)
388        time.sleep(60)
389        begin_time = get_current_epoch_time()
390        for i in range(3):
391            try:
392                ad.take_bug_report(self.test_name, begin_time)
393                bugreport_path = ad.device_log_path
394                break
395            except Exception as e:
396                ad.log.error("bugreport attempt %s error: %s", i + 1, e)
397        ad.log.info("Bugreport Path is %s" % bugreport_path)
398        try:
399            list_of_files = os.listdir(bugreport_path)
400            ad.log.info(list_of_files)
401            for filename in list_of_files:
402                if ".zip" in filename:
403                    ad.log.info(filename)
404                    file_path = os.path.join(bugreport_path, filename)
405                    ad.log.info(file_path)
406                    unzip_maintain_permissions(file_path, bugreport_path)
407            dumpstate_path = os.path.join(bugreport_path,
408                                          "dumpstate_board.bin")
409            if os.path.isfile(dumpstate_path):
410                os.rename(dumpstate_path,
411                          bugreport_path + "/dumpstate_board.tar")
412                os.chmod(bugreport_path + "/dumpstate_board.tar", 0o777)
413                current_dir = os.getcwd()
414                os.chdir(bugreport_path)
415                exe_cmd("tar -xvf %s" %
416                        (bugreport_path + "/dumpstate_board.tar"))
417                os.chdir(current_dir)
418            else:
419                ad.log.info("The dumpstate_path file %s does not exist" % dumpstate_path)
420            if os.path.isfile(bugreport_path + "/power_anomaly_data.txt"):
421                ad.log.info("Modem Power Anomaly File Exists!!")
422                return True
423            ad.log.info("Modem Power Anomaly File DO NOT Exist!!")
424            return False
425        except Exception as e:
426            ad.log.error(e)
427            return False
428
429    @TelephonyBaseTest.tel_test_wrap
430    def test_lock_lte_band_4(self):
431        """Set LTE band lock 4"""
432        if not self.dut.is_apk_installed("com.google.mdstest"):
433            raise signals.TestSkip("mdstest is not installed")
434        return lock_lte_band_by_mds(self.dut, "4")
435
436    @TelephonyBaseTest.tel_test_wrap
437    def test_lock_lte_band_13(self):
438        """Set LTE band lock 4"""
439        if not self.dut.is_apk_installed("com.google.mdstest"):
440            raise signals.TestSkip("mdstest is not installed")
441        return lock_lte_band_by_mds(self.dut, "13")
442
443    @test_tracker_info(uuid="e2a8cb1e-7998-4912-9e16-d9e5f1daee5d")
444    @TelephonyBaseTest.tel_test_wrap
445    def test_modem_ssr_rampdump_generation(self):
446        """Trigger Modem SSR Crash and Verify if Ramdumps are generated
447
448        1. Empty the rampdump dir
449        2. Trigger ModemSSR
450        3. Verify if rampdumps are getting generated or not
451
452        """
453        if not self.dut.is_apk_installed("com.google.mdstest"):
454            raise signals.TestSkip("mdstest is not installed")
455        try:
456            ad = self.android_devices[0]
457            ad.adb.shell("rm -rf /data/vendor/ssrdump/*", ignore_status=True)
458            if not trigger_modem_crash_by_modem(ad):
459                ad.log.error("Failed to trigger Modem SSR, aborting...")
460                return False
461            out = ad.adb.shell("ls -l /data/vendor/ssrdump/ramdump_modem_*",
462                               ignore_status=True)
463            if "No such file" in out or not out:
464                ad.log.error("Ramdump Modem File not found post SSR\n %s", out)
465                return False
466            ad.log.info("Ramdump Modem File found post SSR\n %s", out)
467            return True
468        except Exception as e:
469            ad.log.error(e)
470            return False
471        finally:
472            ad.adb.shell("rm -rf /data/vendor/ssrdump/*", ignore_status=True)
473
474    @test_tracker_info(uuid="e12b2f00-7e18-47d6-b310-aabb96b165a3")
475    @TelephonyBaseTest.tel_test_wrap
476    def test_factory_modem_offline(self):
477        """Trigger Modem Factory Offline and verify if Modem Offline
478
479        1. Device in Fastboot
480        2. Wipe Userdata and set the fastboot command for factory
481        3. Device will bootup in adb, verify Modem Offline
482        4. Reboot again, and verify camping
483
484        """
485        try:
486            ad = self.android_devices[0]
487            skip_setup_wizard=True
488
489            # Pull sl4a apk from device
490            out = ad.adb.shell("pm path %s" % SL4A_APK_NAME)
491            result = re.search(r"package:(.*)", out)
492            if not result:
493                ad.log.error("Couldn't find sl4a apk")
494                return False
495            else:
496                sl4a_apk = result.group(1)
497                ad.log.info("Get sl4a apk from %s", sl4a_apk)
498                ad.pull_files([sl4a_apk], "/tmp/")
499            ad.stop_services()
500
501            # Fastboot Wipe
502            if ad.serial in list_adb_devices():
503                ad.log.info("Reboot to bootloader")
504                ad.adb.reboot("bootloader", ignore_status=True)
505                time.sleep(10)
506            if ad.serial in list_fastboot_devices():
507                ad.log.info("Wipe in fastboot")
508                ad.fastboot._w(timeout=300, ignore_status=True)
509                time.sleep(30)
510
511                # Factory Silent Mode Test
512                ad.log.info("Factory Offline in fastboot")
513                ad.fastboot.oem("continue-factory")
514                time.sleep(30)
515                ad.wait_for_boot_completion()
516                ad.root_adb()
517                ad.log.info("Re-install sl4a")
518                ad.adb.shell("settings put global verifier_verify_adb_installs"
519                             " 0")
520                ad.adb.install("-r /tmp/base.apk")
521                time.sleep(10)
522                try:
523                    ad.start_adb_logcat()
524                except:
525                    ad.log.error("Failed to start adb logcat!")
526                bring_up_sl4a(ad)
527                radio_state = ad.droid.telephonyIsRadioOn()
528                ad.log.info("Radio State is %s", radio_state)
529                if radio_state:
530                    ad.log.error("Radio state is ON in Factory Mode")
531                    return False
532                toggle_airplane_mode(self.log, ad, True)
533                time.sleep(5)
534                toggle_airplane_mode(self.log, ad, False)
535                radio_state = ad.droid.telephonyIsRadioOn()
536                ad.log.info("Radio State is %s", radio_state)
537                if ad.droid.telephonyIsRadioOn():
538                    ad.log.error("Radio state is ON after Airplane Toggle")
539                    return False
540                ad.log.info("Rebooting and verifying back in service")
541
542                # Bring it back to Online Mode
543                ad.stop_services()
544                ad.adb.reboot()
545                ad.wait_for_boot_completion()
546                ad.root_adb()
547                bring_up_sl4a(ad)
548                radio_state = ad.droid.telephonyIsRadioOn()
549                ad.log.info("Radio State is %s", radio_state)
550                if not radio_state:
551                    ad.log.error("Radio state is OFF in Online Mode")
552                    return False
553                return True
554        except Exception as e:
555            ad.log.error(e)
556            return False
557        finally:
558            ad.exit_setup_wizard()
559            bring_up_sl4a(ad)
560
561    @test_tracker_info(uuid="e681e6e2-a33c-4cbd-9ec4-8faa003cde6b")
562    @TelephonyBaseTest.tel_test_wrap
563    def test_carrier_config_version_after_fdr(self):
564        """Carrier Config Version Test after FDR
565
566        1. Disable Verity, remount, push carriersettings apk
567        2. WiFi is connected
568        3. Perform FDR, and re-connect WiFi
569        4. Wait for 45 mins and keep checking for version match
570
571        """
572        try:
573            cc_version_mapping = {
574                'vzw': VZW_CARRIER_CONFIG_VERSION,
575                'Verizon': VZW_CARRIER_CONFIG_VERSION,
576                'att': ATT_CARRIER_CONFIG_VERSION,
577            }
578            result_flag = False
579            time_var = 1
580            ad = self.android_devices[0]
581            skip_setup_wizard=True
582
583            # CarrierSettingsApk
584            carriersettingsapk = self.user_params["carriersettingsapk"]
585            if isinstance(carriersettingsapk, list):
586                carriersettingsapk = carriersettingsapk[0]
587            ad.log.info("Using file path %s", carriersettingsapk)
588
589            if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid,
590                                         self.wifi_network_pass):
591                ad.log.error("connect WiFi failed")
592                return False
593
594            # Setup Steps
595            adb_disable_verity(ad)
596            install_carriersettings_apk(ad, carriersettingsapk)
597
598            # FDR
599            ad.log.info("Performing FDR")
600            fastboot_wipe(ad)
601            ad.log.info("FDR Complete")
602            if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid,
603                                         self.wifi_network_pass):
604                ad.log.error("Connect WiFi failed")
605
606            # Wait for 45 mins for CC version upgrade
607            while(time_var < WAIT_TIME_FOR_CARRIERCONFIG_CHANGE):
608                current_version = get_carrier_config_version(ad)
609                if current_version == cc_version_mapping[self.dut_operator]:
610                    ad.log.info("Carrier Config Version Match %s in %s mins",
611                                current_version, time_var)
612                    result_flag = True
613                    break
614                else:
615                    ad.log.debug("Carrier Config Version Not Match")
616                time.sleep(60)
617                time_var += 1
618            if not result_flag:
619                ad.log.info("Carrier Config Failed to Update in %s mins",
620                             WAIT_TIME_FOR_CARRIERCONFIG_CHANGE)
621            return result_flag
622        except Exception as e:
623            ad.log.error(e)
624            return False
625
626
627    @test_tracker_info(uuid="41e6f2d3-76c9-4d3d-97b3-7075ad98bd41")
628    @TelephonyBaseTest.tel_test_wrap
629    def test_carrier_id_update_wifi_connected(self):
630        """Carrier Id Version Test after WiFi Connected
631
632        1. WiFi is connected
633        2. Perform setup steps to cleanup shared_prefs
634        3. Send P/H flag update to configUpdater
635        4. Wait for 5 mins and keep checking for version match
636
637        """
638        try:
639            result_flag = False
640            time_var = 1
641            ad = self.android_devices[0]
642            if ad.adb.getprop("ro.build.version.release") in ("9", "P", "10", "Q",
643                                "11", "R", "12", "S"):
644                CARRIER_ID_VERSION = CARRIER_ID_VERSION_P
645                CARRIER_ID_METADATA_URL = CARRIER_ID_METADATA_URL_P
646                CARRIER_ID_CONTENT_URL = CARRIER_ID_CONTENT_URL_P
647
648            ad.log.info("Before - CarrierId is %s", get_carrier_id_version(ad))
649            # Setup Steps
650            if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid,
651                                         self.wifi_network_pass):
652                ad.log.error("connect WiFi failed")
653                return False
654            cleanup_configupdater(ad)
655            time.sleep(5)
656
657            # Trigger Config Update
658            ad.log.info("Triggering Config Update")
659            ad.log.info("%s", CARRIER_ID_METADATA_URL)
660            exe_cmd("adb -s %s shell %s" % (ad.serial,CARRIER_ID_METADATA_URL))
661            ad.log.info("%s", CARRIER_ID_CONTENT_URL)
662            exe_cmd("adb -s %s shell %s" % (ad.serial,CARRIER_ID_CONTENT_URL))
663
664            # Wait for 5 mins for Carrier Id version upgrade
665            while(time_var < WAIT_TIME_FOR_CARRIERID_CHANGE):
666                current_version = get_carrier_id_version(ad)
667                if current_version == CARRIER_ID_VERSION:
668                    ad.log.info("After CarrierId is %s in %s mins",
669                                current_version, time_var)
670                    result_flag = True
671                    break
672                else:
673                    ad.log.debug("Carrier Id Version Not Match")
674                time.sleep(60)
675                time_var += 1
676
677            if not result_flag:
678                ad.log.info("Carrier Id Failed to Update in %s mins",
679                             WAIT_TIME_FOR_CARRIERID_CHANGE)
680
681            # pb file check
682            out = ad.adb.shell("ls -l data/misc/carrierid/carrier_list.pb")
683            if "No such" in out:
684                ad.log.error("carrier_list.pb file is missing")
685                result_flag = False
686            else:
687                ad.log.info("carrier_list.pb file is present")
688            return result_flag
689        except Exception as e:
690            ad.log.error(e)
691            return False
692        finally:
693            carrier_id_path = os.path.join(self.log_path, self.test_name,
694                                           "CarrierId_%s" % ad.serial)
695            pull_carrier_id_files(ad, carrier_id_path)
696
697
698    @test_tracker_info(uuid="836d3963-f56d-438e-a35c-0706ac385153")
699    @TelephonyBaseTest.tel_test_wrap
700    def test_carrier_id_update_wifi_disconnected(self):
701        """Carrier Id Version Test with WiFi disconnected
702
703        1. WiFi is connected
704        2. Perform setup steps to cleanup shared_prefs
705        3. Send P/H flag update to configUpdater
706        4. Wait for 5 mins and keep checking for version match
707
708        """
709        try:
710            result_flag = False
711            time_var = 1
712            ad = self.android_devices[0]
713            ad.log.info("Before - CarrierId is %s", get_carrier_id_version(ad))
714
715            # Wifi Disconnect
716            cleanup_configupdater(ad)
717            wifi_toggle_state(ad.log, ad, False)
718
719            # Trigger Config Update
720            ad.log.info("Triggering Config Update")
721            ad.log.info("%s", CARRIER_ID_METADATA_URL)
722            exe_cmd("adb -s %s shell %s" % (ad.serial,CARRIER_ID_METADATA_URL))
723            ad.log.info("%s", CARRIER_ID_CONTENT_URL)
724            exe_cmd("adb -s %s shell %s" % (ad.serial,CARRIER_ID_CONTENT_URL))
725
726            # Wait for 5 mins for Carrier Id version upgrade
727            while(time_var < WAIT_TIME_FOR_CARRIERID_CHANGE):
728                current_version = get_carrier_id_version(ad)
729                if current_version == CARRIER_ID_VERSION:
730                    ad.log.info("After CarrierId is %s in %s mins",
731                                current_version, time_var)
732                    return False
733                else:
734                    ad.log.debug("Carrier Id Version Not Match")
735                time.sleep(60)
736                time_var += 1
737            time_var = 1
738            ad.log.info("Success - CarrierId not upgraded during WiFi OFF")
739
740            # WiFi Connect
741            if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid,
742                                         self.wifi_network_pass):
743                ad.log.error("connect WiFi failed")
744                return False
745
746            # Wait for 5 mins for Carrier Id version upgrade
747            while(time_var < WAIT_TIME_FOR_CARRIERID_CHANGE):
748                current_version = get_carrier_id_version(ad)
749                if current_version == CARRIER_ID_VERSION:
750                    ad.log.info("After CarrierId is %s in %s mins",
751                                current_version, time_var)
752                    result_flag = True
753                    break
754                else:
755                    ad.log.debug("Carrier Id Version Not Match")
756                time.sleep(60)
757                time_var += 1
758
759            if not result_flag:
760                ad.log.info("Carrier Id Failed to Update in %s mins",
761                             WAIT_TIME_FOR_CARRIERID_CHANGE)
762            # pb file check
763            out = ad.adb.shell("ls -l data/misc/carrierid/carrier_list.pb")
764            if not out or "No such" in out:
765                ad.log.error("carrier_list.pb file is missing")
766                result_flag = False
767            else:
768                ad.log.info("carrier_list.pb file is present")
769            return result_flag
770        except Exception as e:
771            ad.log.error(e)
772            return False
773        finally:
774            carrier_id_path = os.path.join(self.log_path, self.test_name,
775                                           "CarrierId_%s" % ad.serial)
776            pull_carrier_id_files(ad, carrier_id_path)
777
778    @test_tracker_info(uuid="0edcafb1-b76d-45e1-9774-0638820b8b6e")
779    @TelephonyBaseTest.tel_test_wrap
780    def test_emergency_database_update_wifi_connected(self):
781        """Emergency DB Id Version Test after WiFi Connected
782
783        1. WiFi is connected
784        2. Login with whitelisted gmail credentials
785        3. Wait for 5 mins and keep checking for version match
786
787        """
788        try:
789            result_flag = False
790            time_var = 1
791            ad = self.android_devices[0]
792
793            # Get the Emergency database Id
794            ad.log.info("Before - Emergency DB Id is %s",
795                    get_er_db_id_version(ad))
796
797            # Connect to Wifi
798            if not ensure_wifi_connected(self.log, ad, self.wifi_network_ssid,
799                                         self.wifi_network_pass):
800                ad.log.error("connect WiFi failed")
801                return False
802            time.sleep(5)
803
804            #Login with whitelisted google account
805            if not install_googleaccountutil_apk(ad,self.path):
806                ad.log.error("Failed to install Google Util")
807                return False
808            if not add_whitelisted_account(ad,self.user_account,
809            self.user_password):
810                ad.log.error("Failed to Login with Google Account")
811                return False
812
813            # Wait for 5 mins for Emergency database Id version upgrade
814            while(time_var < WAIT_TIME_FOR_ER_DB_CHANGE):
815                current_version = get_er_db_id_version(ad)
816                if current_version == ER_DB_ID_VERSION:
817                    ad.log.info("ER DB version is %s in %s mins",
818                                current_version, time_var)
819                    result_flag = True
820                    break
821                else:
822                    ad.log.debug("ER Database Version Id Not Match")
823                time.sleep(60)
824                time_var += 1
825
826            if not result_flag:
827                ad.log.info("ER Database version Id Failed to Update in %s mins",
828                             WAIT_TIME_FOR_ER_DB_CHANGE)
829                return False
830            # Verify Emerency Database content
831            if not get_database_content(ad):
832                ad.log.error("Emergency Number does not match")
833                result_flag = False
834                return False
835            ad.log.info("Emergency Number is 54321")
836
837        except Exception as e:
838            ad.log.error(e)
839            return False
840        finally:
841            return result_flag
842""" Tests End """
843