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.
16import json
17
18from acts import base_test
19
20import acts.controllers.cellular_simulator as simulator
21from acts.controllers.anritsu_lib import md8475_cellular_simulator as anritsu
22from acts.controllers.rohdeschwarz_lib import cmw500_cellular_simulator as cmw
23from acts.controllers.rohdeschwarz_lib import cmx500_cellular_simulator as cmx
24from acts.controllers.uxm_lib import uxm_cellular_simulator as uxm
25from acts.controllers.cellular_lib import AndroidCellularDut
26from acts.controllers.cellular_lib import BaseSimulation as base_sim
27from acts.controllers.cellular_lib import GsmSimulation as gsm_sim
28from acts.controllers.cellular_lib import LteSimulation as lte_sim
29from acts.controllers.cellular_lib import UmtsSimulation as umts_sim
30from acts.controllers.cellular_lib import LteImsSimulation as lteims_sim
31from acts.controllers.cellular_lib import PresetSimulation
32
33from acts_contrib.test_utils.tel import tel_test_utils as telutils
34
35
36class CellularBaseTest(base_test.BaseTestClass):
37    """ Base class for modem functional tests. """
38
39    # List of test name keywords that indicate the RAT to be used
40
41    PARAM_SIM_TYPE_LTE = 'lte'
42    PARAM_SIM_TYPE_LTE_CA = 'lteca'
43    PARAM_SIM_TYPE_LTE_IMS = 'lteims'
44    PARAM_SIM_TYPE_NR = 'nr'
45    PARAM_SIM_TYPE_UMTS = 'umts'
46    PARAM_SIM_TYPE_GSM = 'gsm'
47    PARAM_SIM_TYPE_PRESET = 'preset'
48
49    # Custom files
50    FILENAME_CALIBRATION_TABLE_UNFORMATTED = 'calibration_table_{}.json'
51    FILENAME_TEST_CONFIGS = 'cellular_test_config.json'
52
53    # Name of the files in the logs directory that will contain test results
54    # and other information in csv format.
55    RESULTS_SUMMARY_FILENAME = 'cellular_power_results.csv'
56    CALIBRATION_TABLE_FILENAME = 'calibration_table.csv'
57
58    def __init__(self, controllers):
59        """Class initialization.
60
61        Sets class attributes to None.
62        """
63
64        super().__init__(controllers)
65
66        self.simulation = None
67        self.cellular_simulator = None
68        self.calibration_table = {}
69        self.test_configs = {}
70
71    def setup_class(self):
72        """Executed before any test case is started.
73
74        Connects to the cellular instrument.
75
76        Returns:
77            False if connecting to the callbox fails.
78        """
79
80        super().setup_class()
81
82        self.cellular_dut = AndroidCellularDut.AndroidCellularDut(
83            self.android_devices[0], self.log)
84
85        TEST_PARAMS = self.TAG + '_params'
86        self.cellular_test_params = self.user_params.get(TEST_PARAMS, {})
87        self.log.info('self.cellular_test_params: ' +
88                      str(self.cellular_test_params))
89
90        # Unpack test parameters used in this class
91        self.unpack_userparams(['custom_files'],
92                               md8475_version=None,
93                               md8475a_ip_address=None,
94                               cmw500_ip=None,
95                               cmw500_port=None,
96                               cmx500_ip=None,
97                               cmx500_port=None,
98                               modem_logging=None,
99                               uxm_ip=None,
100                               disable_data=None)
101
102        # Load calibration tables
103        filename_calibration_table = (
104            self.FILENAME_CALIBRATION_TABLE_UNFORMATTED.format(
105                self.testbed_name))
106
107        for file in self.custom_files:
108            if filename_calibration_table in file:
109                with open(file, 'r') as f:
110                    self.calibration_table = json.load(f)
111                self.log.info('Loading calibration table from ' + file)
112                self.log.debug(self.calibration_table)
113                break
114
115        # Ensure the calibration table only contains non-negative values
116        self.ensure_valid_calibration_table(self.calibration_table)
117
118        # Load test configs from json file
119        for file in self.custom_files:
120            if self.FILENAME_TEST_CONFIGS in file:
121                self.log.info('Loading test configs from ' + file)
122                with open(file, 'r') as f:
123                    config_file = json.load(f)
124                self.test_configs = config_file.get(self.TAG)
125                if not self.test_configs:
126                    self.log.debug(config_file)
127                    raise RuntimeError('Test config file does not include '
128                                       'class %s'.format(self.TAG))
129                self.log.debug(self.test_configs)
130                break
131
132        # Turn on airplane mode for all devices, as some might
133        # be unused during the test
134        for ad in self.android_devices:
135            telutils.toggle_airplane_mode_by_adb(self.log, ad, True)
136
137        # Establish a connection with the cellular simulator equipment
138        try:
139            self.cellular_simulator = self.initialize_simulator()
140        except ValueError:
141            self.log.error('No cellular simulator could be selected with the '
142                           'current configuration.')
143            raise
144        except simulator.CellularSimulatorError:
145            self.log.error('Could not initialize the cellular simulator.')
146            raise
147
148    def initialize_simulator(self):
149        """Connects to Anritsu Callbox and gets handle object.
150
151        Returns:
152            False if a connection with the callbox could not be started
153        """
154
155        if self.md8475_version:
156
157            self.log.info('Selecting Anrtisu MD8475 callbox.')
158
159            # Verify the callbox IP address has been indicated in the configs
160            if not self.md8475a_ip_address:
161                raise RuntimeError(
162                    'md8475a_ip_address was not included in the test '
163                    'configuration.')
164
165            if self.md8475_version == 'A':
166                return anritsu.MD8475CellularSimulator(self.md8475a_ip_address)
167            elif self.md8475_version == 'B':
168                return anritsu.MD8475BCellularSimulator(
169                    self.md8475a_ip_address)
170            else:
171                raise ValueError('Invalid MD8475 version.')
172
173        elif self.cmw500_ip or self.cmw500_port:
174
175            for key in ['cmw500_ip', 'cmw500_port']:
176                if not getattr(self, key):
177                    raise RuntimeError('The CMW500 cellular simulator '
178                                       'requires %s to be set in the '
179                                       'config file.' % key)
180
181            return cmw.CMW500CellularSimulator(self.cmw500_ip,
182                                               self.cmw500_port)
183        elif self.cmx500_ip or self.cmx500_port:
184            for key in ['cmx500_ip', 'cmx500_port']:
185                if not getattr(self, key):
186                    raise RuntimeError('The CMX500 cellular simulator '
187                                       'requires %s to be set in the '
188                                       'config file.' % key)
189
190            return cmx.CMX500CellularSimulator(self.cmx500_ip,
191                                               self.cmx500_port)
192
193        elif self.uxm_ip:
194            # unpack additional uxm info
195            self.unpack_userparams(uxm_user=None,
196                                   ssh_private_key_to_uxm=None,
197                                   ta_exe_path=None,
198                                   ta_exe_name=None)
199            for param in ('uxm_ip', 'uxm_user', 'ssh_private_key_to_uxm',
200                          'ta_exe_path', 'ta_exe_name', 'custom_files'):
201                if getattr(self, param) is None:
202                    raise RuntimeError('The uxm cellular simulator '
203                                       'requires %s to be set in the '
204                                       'config file.' % param)
205            return uxm.UXMCellularSimulator(self.uxm_ip, self.custom_files,
206                                            self.uxm_user,
207                                            self.ssh_private_key_to_uxm,
208                                            self.ta_exe_path, self.ta_exe_name)
209
210        else:
211            raise RuntimeError(
212                'The simulator could not be initialized because '
213                'a callbox was not defined in the configs file.')
214
215    def setup_test(self):
216        """Executed before every test case.
217
218        Parses parameters from the test name and sets a simulation up according
219        to those values. Also takes care of attaching the phone to the base
220        station. Because starting new simulations and recalibrating takes some
221        time, the same simulation object is kept between tests and is only
222        destroyed and re instantiated in case the RAT is different from the
223        previous tests.
224
225        Children classes need to call the parent method first. This method will
226        create the list self.parameters with the keywords separated by
227        underscores in the test name and will remove the ones that were consumed
228        for the simulation config. The setup_test methods in the children
229        classes can then consume the remaining values.
230        """
231
232        super().setup_test()
233
234        # Get list of parameters from the test name
235        self.parameters = self.current_test_name.split('_')
236
237        # Remove the 'test' keyword
238        self.parameters.remove('test')
239
240        # Decide what type of simulation and instantiate it if needed
241        if self.consume_parameter(self.PARAM_SIM_TYPE_LTE):
242            self.init_simulation(self.PARAM_SIM_TYPE_LTE)
243        elif self.consume_parameter(self.PARAM_SIM_TYPE_LTE_CA):
244            self.init_simulation(self.PARAM_SIM_TYPE_LTE_CA)
245        elif self.consume_parameter(self.PARAM_SIM_TYPE_LTE_IMS):
246            self.init_simulation(self.PARAM_SIM_TYPE_LTE_IMS)
247        elif self.consume_parameter(self.PARAM_SIM_TYPE_NR):
248            self.init_simulation(self.PARAM_SIM_TYPE_NR)
249        elif self.consume_parameter(self.PARAM_SIM_TYPE_UMTS):
250            self.init_simulation(self.PARAM_SIM_TYPE_UMTS)
251        elif self.consume_parameter(self.PARAM_SIM_TYPE_GSM):
252            self.init_simulation(self.PARAM_SIM_TYPE_GSM)
253        elif self.consume_parameter(self.PARAM_SIM_TYPE_PRESET):
254            self.init_simulation(self.PARAM_SIM_TYPE_PRESET)
255        else:
256            self.log.error(
257                'Simulation type needs to be indicated in the test name.')
258            return False
259
260        # Changing cell parameters requires the phone to be detached
261        self.simulation.detach()
262
263        # Disable or enable data according to the config. Data is on by default
264        if self.disable_data:
265            self.log.info('disable data for the test')
266        else:
267            self.log.info('enable data for the test')
268        self.cellular_dut.toggle_data(not self.disable_data)
269
270        # Configure simulation with parameters loaded from json file
271        sim_params = self.test_configs.get(self.test_name)
272        if not sim_params:
273            raise KeyError('Test config file does not contain '
274                           'settings for ' + self.test_name)
275
276        # Changes the single band sim_params type to list to make it easier
277        # to apply the class parameters to test parameters for multiple bands
278        if not isinstance(sim_params, list):
279            sim_params = [sim_params]
280        num_band = len(sim_params)
281
282        # Get class parameters and apply if not overwritten by test parameters
283        for key, val in self.test_configs.items():
284            if not key.startswith('test_'):
285                for idx in range(num_band):
286                    if key not in sim_params[idx]:
287                        sim_params[idx][key] = val
288        self.log.info('Simulation parameters: ' + str(sim_params))
289        self.simulation.configure(sim_params)
290
291        if self.modem_logging:
292            try:
293                self.cellular_dut.start_modem_logging()
294            except NotImplementedError:
295                self.log.error('Modem logging couldn\'t be started')
296
297        # Start the simulation. This method will raise an exception if
298        # the phone is unable to attach.
299        self.simulation.start()
300
301        return True
302
303    def teardown_test(self):
304        """Executed after every test case, even if it failed or an exception occur.
305
306        Save results to dictionary so they can be displayed after completing
307        the test batch.
308        """
309        super().teardown_test()
310
311        if self.modem_logging:
312            self.cellular_dut.stop_modem_logging()
313
314    def consume_parameter(self, parameter_name, num_values=0):
315        """ Parses a parameter from the test name.
316
317        Allows the test to get parameters from its name. Deletes parameters from
318        the list after consuming them to ensure that they are not used twice.
319
320        Args:
321            parameter_name: keyword to look up in the test name
322            num_values: number of arguments following the parameter name in the
323              test name
324
325        Returns:
326            A list containing the parameter name and the following num_values
327            arguments.
328        """
329
330        try:
331            i = self.parameters.index(parameter_name)
332        except ValueError:
333            # parameter_name is not set
334            return []
335
336        return_list = []
337
338        try:
339            for j in range(num_values + 1):
340                return_list.append(self.parameters.pop(i))
341        except IndexError:
342            self.log.error(
343                'Parameter {} has to be followed by {} values.'.format(
344                    parameter_name, num_values))
345            raise ValueError()
346
347        return return_list
348
349    def teardown_class(self):
350        """Clean up the test class after tests finish running.
351
352        Stops the simulation and disconnects from the Anritsu Callbox. Then
353        displays the test results.
354        """
355        super().teardown_class()
356
357        try:
358            if self.cellular_simulator:
359                self.cellular_simulator.destroy()
360        except simulator.CellularSimulatorError as e:
361            self.log.error('Error while tearing down the callbox controller. '
362                           'Error message: ' + str(e))
363
364    def init_simulation(self, sim_type):
365        """Starts a new simulation only if needed.
366
367        Only starts a new simulation if type is different from the one running
368        before.
369
370        Args:
371            type: defines the type of simulation to be started.
372        """
373        simulation_dictionary = {
374            self.PARAM_SIM_TYPE_LTE:
375            lte_sim.LteSimulation,
376            self.PARAM_SIM_TYPE_LTE_CA:
377            lte_sim.LteSimulation,
378            # The LteSimulation class is able to handle NR cells as well.
379            # The long-term goal is to consolidate all simulation classes.
380            self.PARAM_SIM_TYPE_NR:
381            lte_sim.LteSimulation,
382            self.PARAM_SIM_TYPE_UMTS:
383            umts_sim.UmtsSimulation,
384            self.PARAM_SIM_TYPE_GSM:
385            gsm_sim.GsmSimulation,
386            self.PARAM_SIM_TYPE_LTE_IMS:
387            lteims_sim.LteImsSimulation,
388            self.PARAM_SIM_TYPE_PRESET:
389            PresetSimulation.PresetSimulation,
390        }
391
392        if not sim_type in simulation_dictionary:
393            raise ValueError('The provided simulation type is invalid.')
394
395        simulation_class = simulation_dictionary[sim_type]
396
397        if isinstance(self.simulation, simulation_class):
398            # The simulation object we already have is enough.
399            return
400
401        if self.simulation:
402            # Make sure the simulation is stopped before loading a new one
403            self.simulation.stop()
404
405        # If the calibration table doesn't have an entry for this simulation
406        # type add an empty one
407        if sim_type not in self.calibration_table:
408            self.calibration_table[sim_type] = {}
409
410        # Instantiate a new simulation
411        if sim_type == self.PARAM_SIM_TYPE_NR:
412            self.simulation = simulation_class(
413                self.cellular_simulator,
414                self.log,
415                self.cellular_dut,
416                self.cellular_test_params,
417                self.calibration_table[sim_type],
418                nr_mode=self.PARAM_SIM_TYPE_NR)
419        elif sim_type == self.PARAM_SIM_TYPE_PRESET:
420            self.simulation = self.simulation = simulation_class(
421                self.cellular_simulator,
422                self.log,
423                self.cellular_dut,
424                self.cellular_test_params,
425                self.calibration_table,
426                nr_mode=self.PARAM_SIM_TYPE_NR)
427        else:
428            self.simulation = simulation_class(
429                self.cellular_simulator, self.log, self.cellular_dut,
430                self.cellular_test_params, self.calibration_table[sim_type])
431
432    def ensure_valid_calibration_table(self, calibration_table):
433        """ Ensures the calibration table has the correct structure.
434
435        A valid calibration table is a nested dictionary with non-negative
436        number values
437
438        """
439        if not isinstance(calibration_table, dict):
440            raise TypeError('The calibration table must be a dictionary')
441        for val in calibration_table.values():
442            if isinstance(val, dict):
443                self.ensure_valid_calibration_table(val)
444            elif not isinstance(val, float) and not isinstance(val, int):
445                raise TypeError('Calibration table value must be a number')
446            elif val < 0.0:
447                raise ValueError('Calibration table contains negative values')
448