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