1#!/usr/bin/env python3
2#
3#   Copyright 2021 - The Android Open Source Project
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
18
19import sys
20import collections
21import random
22import time
23import datetime
24import os
25import logging
26import subprocess
27import math
28import re
29
30from acts import asserts
31from acts.test_decorators import test_info
32from acts.test_decorators import test_tracker_info
33from acts.logger import epoch_to_log_line_timestamp
34from acts.utils import get_current_epoch_time
35
36from acts.base_test import BaseTestClass
37from acts_contrib.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
38
39from acts_contrib.test_utils.tel.tel_test_utils import multithread_func
40from acts_contrib.test_utils.tel.tel_test_utils import run_multithread_func
41from acts_contrib.test_utils.tel.tel_test_utils import get_service_state_by_adb
42from acts_contrib.test_utils.tel.tel_test_utils import get_screen_shot_log
43from acts_contrib.test_utils.tel.tel_test_utils import get_screen_shot_logs
44from acts_contrib.test_utils.tel.tel_test_utils import log_screen_shot
45from acts_contrib.test_utils.tel.tel_test_utils import hangup_call
46from acts_contrib.test_utils.tel.tel_test_utils import is_phone_in_call
47
48from acts_contrib.test_utils.tel.gft_inout_utils import mo_voice_call
49from acts_contrib.test_utils.tel.gft_inout_utils import get_voice_call_type
50from acts_contrib.test_utils.tel.gft_inout_utils import verify_data_connection
51from acts_contrib.test_utils.tel.gft_inout_utils import check_network_service
52
53from acts_contrib.test_utils.tel.gft_inout_defines import NO_SERVICE_POWER_LEVEL
54from acts_contrib.test_utils.tel.gft_inout_defines import IN_SERVICE_POWER_LEVEL
55from acts_contrib.test_utils.tel.gft_inout_defines import NO_SERVICE_AREA
56from acts_contrib.test_utils.tel.gft_inout_defines import IN_SERVICE_AREA
57from acts_contrib.test_utils.tel.gft_inout_defines import WIFI_AREA
58from acts_contrib.test_utils.tel.gft_inout_defines import NO_WIFI_AREA
59from acts_contrib.test_utils.tel.gft_inout_defines import NO_SERVICE_TIME
60from acts_contrib.test_utils.tel.gft_inout_defines import WAIT_FOR_SERVICE_TIME
61
62CELLULAR_PORT = 0
63WIFI_PORT = 1
64UNKNOWN = "UNKNOWN"
65
66class GFTInOutBaseTest(TelephonyBaseTest):
67    def __init__(self, controllers):
68        TelephonyBaseTest.__init__(self, controllers)
69        self.my_error_msg = ""
70
71    def setup_test(self):
72        TelephonyBaseTest.setup_test(self)
73        self.my_error_msg = ""
74
75    def teardown_test(self):
76        TelephonyBaseTest.teardown_test(self)
77        begin_time = get_current_epoch_time()
78        self._take_bug_report(self.test_name, begin_time)
79        for ad in self.android_devices:
80            hangup_call(self.log, ad)
81        get_screen_shot_logs(self.android_devices)
82        for ad in self.android_devices:
83            ad.adb.shell("rm -rf /sdcard/Pictures/screenvideo_*", ignore_status=True)
84
85    def check_network(self):
86        """check service state of network
87
88        Returns:
89            return True if android_devices are in service else return false
90        """
91        for ad in self.android_devices:
92            network_type_voice = ad.droid.telephonyGetCurrentVoiceNetworkType()
93            network_type_data = ad.droid.telephonyGetCurrentDataNetworkType()
94            service_state = get_service_state_by_adb(ad.log,ad)
95            sim_state = ad.droid.telephonyGetSimState()
96            wifi_info = ad.droid.wifiGetConnectionInfo()
97            if wifi_info["supplicant_state"] == "completed":
98                ad.log.info("Wifi is connected to %s" %(wifi_info["SSID"]))
99            else:
100                ad.log.info("wifi_info =%s" %(wifi_info))
101            ad.log.info("sim_state=%s" %(sim_state))
102            ad.log.info("networkType_voice=%s" %(network_type_voice))
103            ad.log.info("network_type_data=%s" %(network_type_data))
104            ad.log.info("service_state=%s" %(service_state))
105            if network_type_voice == UNKNOWN:
106                return False
107        return True
108
109
110    def adjust_cellular_signal(self, power):
111        """Sets the attenuation value of cellular port
112
113        Args:
114             power: power level for attenuator to be set
115
116        Returns:
117            return True if ceullular port is set
118        """
119        if self.user_params.get("Attenuator"):
120            self.log.info("adjust cellular signal set mini-circuits to %s" %(power))
121            self.attenuators[CELLULAR_PORT].set_atten(power)
122            return True
123        else:
124            self.log.info("Attenuator is set to False in config file")
125            return False
126
127    def adjust_wifi_signal(self, power):
128        """Sets the attenuation value of wifi port
129
130        Args:
131             power: power level for attenuator to be set
132
133        Returns:
134            return True if wifi port is set
135        """
136        if self.user_params.get("Attenuator"):
137            self.log.info("adjust wifi signal set mini-circuits to %s" %(power))
138            self.attenuators[WIFI_PORT].set_atten(power)
139            self.attenuators[2].set_atten(power)
140            self.attenuators[3].set_atten(power)
141        else:
142            self.log.info("Attenuator is set to False in config file")
143            return False
144        return True
145
146    def adjust_attens(self, power):
147        """Sets the attenuation value of all attenuators in the group
148
149        Args:
150             power: power level for attenuator to be set
151
152        Returns:
153            return True if all ports are set
154        """
155        if self.user_params.get("Attenuator"):
156            self.log.info("set attenuator ports to %s" %(power))
157            self.attenuators[0].set_atten(power)
158            self.attenuators[1].set_atten(power)
159            self.attenuators[2].set_atten(power)
160            self.attenuators[3].set_atten(power)
161        else:
162            self.log.info("Attenuator is set to False in config file")
163            return False
164        return True
165
166    def adjust_atten(self, port , power):
167        """Sets the attenuation value of given port
168
169        Args:
170            port: port of attenuator
171            power: power level for attenuator to be set
172
173        Returns:
174            return True if given port is set
175        """
176        if self.user_params.get("Attenuator"):
177            self.log.info("set attenuator port=%s to %s" %(port, power))
178            self.attenuators[port].set_atten(power)
179        else:
180            self.log.info("Attenuator is set to False in config file")
181            return False
182        return True
183
184    def adjust_atten_slowly(self, adjust_level, move_to, step=9 , step_time=5):
185        """adjust attenuator slowly
186
187        Args:
188            port: port of attenuator
189            power: power level for attenuator to be set
190
191        Returns:
192            return True if given port is set
193        """
194        if move_to == NO_SERVICE_AREA:
195            self.log.info("Move UE from coverage area to no/poor service area")
196            power_level = IN_SERVICE_POWER_LEVEL
197            for x in range (step):
198                power_level += adjust_level
199                self.log.info("adjust power level = %s" %power_level)
200                self.adjust_cellular_signal(power_level)
201                time.sleep(step_time)
202                self.log.info("wait device to be in no service state")
203                self.check_network()
204        elif move_to == IN_SERVICE_AREA:
205            # Move UE to service area
206            self.log.info("Move UE to service area")
207            power_level = NO_SERVICE_POWER_LEVEL
208            for x in range (step):
209                power_level -= adjust_level
210                self.log.info("adjust power level = %s" %power_level)
211                self.adjust_cellular_signal(power_level)
212                time.sleep(step_time)
213                self.log.info("wait device to be in service state")
214                self.check_network()
215        elif move_to == WIFI_AREA:
216            self.log.info("Move UE to good wifi area")
217            power_level = IN_SERVICE_POWER_LEVEL
218            for x in range (step):
219                power_level += adjust_level
220                self.log.info("adjust power level = %s" %power_level)
221                self.adjust_wifi_signal(power_level)
222                time.sleep(step_time)
223                self.log.info("wait device to connected to wifi")
224                self.check_network()
225        elif move_to == NO_WIFI_AREA:
226            self.log.info("Move UE to poor/no wifi area")
227            power_level = NO_SERVICE_POWER_LEVEL
228            for x in range (step):
229                power_level -= adjust_level
230                self.log.info("adjust power level = %s" %power_level)
231                self.adjust_wifi_signal(power_level)
232                time.sleep(step_time)
233                self.log.info("wait device to disconnected from wifi")
234                self.check_network()
235        return True
236
237    def verify_device_status(self, ad, call_type=None, end_call=True, talk_time=30):
238        """verfiy device status includes network service, data connection and voice call
239
240        Args:
241            ad: android device
242            call_type: WFC call, VOLTE call. CSFB call, voice call
243            end_call: hangup call after voice call flag
244            talk_time: in call duration in sec
245
246        Returns:
247            return True if given port is set
248        """
249        test_result = True
250        tasks = [(check_network_service, (ad, )) for ad in self.android_devices]
251        if not multithread_func(self.log, tasks):
252            test_result = False
253        tasks = [(verify_data_connection, (ad, )) for ad in self.android_devices]
254        if not multithread_func(self.log, tasks):
255            test_result = False
256        if call_type != None:
257            tasks = [(mo_voice_call, (self.log, ad, call_type, end_call, talk_time))
258                for ad in self.android_devices]
259        if not multithread_func(self.log, tasks):
260            test_result = False
261        return test_result
262