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