acts_tests/acts_contrib/test_utils/cellular/cellular_base_test.py

#!/usr/bin/env python3
#
#   Copyright 2021 - The Android Open Source Project
#
#   Licensed under the Apache License, Version 2.0 (the 'License');
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an 'AS IS' BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
import time
import json

from acts import base_test

import acts.controllers.cellular_simulator as simulator
from acts.controllers.anritsu_lib import md8475_cellular_simulator as anritsu
from acts.controllers.rohdeschwarz_lib import cmw500_cellular_simulator as cmw
from acts.controllers.rohdeschwarz_lib import cmx500_cellular_simulator as cmx
from acts.controllers.cellular_lib import AndroidCellularDut
from acts.controllers.cellular_lib import GsmSimulation
from acts.controllers.cellular_lib import LteSimulation
from acts.controllers.cellular_lib import UmtsSimulation
from acts.controllers.cellular_lib import LteCaSimulation
from acts.controllers.cellular_lib import LteImsSimulation

from acts_contrib.test_utils.tel import tel_test_utils as telutils


class CellularBaseTest(base_test.BaseTestClass):
    """ Base class for modem functional tests. """

    # List of test name keywords that indicate the RAT to be used

    PARAM_SIM_TYPE_LTE = "lte"
    PARAM_SIM_TYPE_LTE_CA = "lteca"
    PARAM_SIM_TYPE_LTE_IMS = "lteims"
    PARAM_SIM_TYPE_UMTS = "umts"
    PARAM_SIM_TYPE_GSM = "gsm"

    # Custom files
    FILENAME_CALIBRATION_TABLE_UNFORMATTED = 'calibration_table_{}.json'

    # Name of the files in the logs directory that will contain test results
    # and other information in csv format.
    RESULTS_SUMMARY_FILENAME = 'cellular_power_results.csv'
    CALIBRATION_TABLE_FILENAME = 'calibration_table.csv'

    def __init__(self, controllers):
        """ Class initialization.

        Sets class attributes to None.
        """

        super().__init__(controllers)

        self.simulation = None
        self.cellular_simulator = None
        self.calibration_table = {}

    def setup_class(self):
        """ Executed before any test case is started.
        Connects to the cellular instrument.

        Returns:
            False if connecting to the callbox fails.
        """

        super().setup_class()

        if not hasattr(self, 'dut'):
            self.dut = self.android_devices[0]

        TEST_PARAMS = self.TAG + '_params'
        self.cellular_test_params = self.user_params.get(TEST_PARAMS, {})

        # Unpack test parameters used in this class
        self.unpack_userparams(['custom_files'],
                               md8475_version=None,
                               md8475a_ip_address=None,
                               cmw500_ip=None,
                               cmw500_port=None,
                               cmx500_ip=None,
                               cmx500_port=None,
                               qxdm_logs=None)

        # Load calibration tables
        filename_calibration_table = (
            self.FILENAME_CALIBRATION_TABLE_UNFORMATTED.format(
                self.testbed_name))

        for file in self.custom_files:
            if filename_calibration_table in file:
                self.calibration_table = self.unpack_custom_file(file, False)
                self.log.info('Loading calibration table from ' + file)
                self.log.debug(self.calibration_table)
                break

        # Ensure the calibration table only contains non-negative values
        self.ensure_valid_calibration_table(self.calibration_table)

        # Turn on airplane mode for all devices, as some might
        # be unused during the test
        for ad in self.android_devices:
            telutils.toggle_airplane_mode(self.log, ad, True)

        # Establish a connection with the cellular simulator equipment
        try:
            self.cellular_simulator = self.initialize_simulator()
        except ValueError:
            self.log.error('No cellular simulator could be selected with the '
                           'current configuration.')
            raise
        except simulator.CellularSimulatorError:
            self.log.error('Could not initialize the cellular simulator.')
            raise

    def initialize_simulator(self):
        """ Connects to Anritsu Callbox and gets handle object.

        Returns:
            False if a connection with the callbox could not be started
        """

        if self.md8475_version:

            self.log.info('Selecting Anrtisu MD8475 callbox.')

            # Verify the callbox IP address has been indicated in the configs
            if not self.md8475a_ip_address:
                raise RuntimeError(
                    'md8475a_ip_address was not included in the test '
                    'configuration.')

            if self.md8475_version == 'A':
                return anritsu.MD8475CellularSimulator(self.md8475a_ip_address)
            elif self.md8475_version == 'B':
                return anritsu.MD8475BCellularSimulator(
                    self.md8475a_ip_address)
            else:
                raise ValueError('Invalid MD8475 version.')

        elif self.cmw500_ip or self.cmw500_port:

            for key in ['cmw500_ip', 'cmw500_port']:
                if not getattr(self, key):
                    raise RuntimeError('The CMW500 cellular simulator '
                                       'requires %s to be set in the '
                                       'config file.' % key)

            return cmw.CMW500CellularSimulator(self.cmw500_ip,
                                               self.cmw500_port)
        elif self.cmx500_ip or self.cmx500_port:
            for key in ['cmx500_ip', 'cmx500_port']:
                if not getattr(self, key):
                    raise RuntimeError('The CMX500 cellular simulator '
                                       'requires %s to be set in the '
                                       'config file.' % key)

            return cmx.CMX500CellularSimulator(self.cmx500_ip,
                                               self.cmx500_port)

        else:
            raise RuntimeError(
                'The simulator could not be initialized because '
                'a callbox was not defined in the configs file.')

    def setup_test(self):
        """ Executed before every test case.

        Parses parameters from the test name and sets a simulation up according
        to those values. Also takes care of attaching the phone to the base
        station. Because starting new simulations and recalibrating takes some
        time, the same simulation object is kept between tests and is only
        destroyed and re instantiated in case the RAT is different from the
        previous tests.

        Children classes need to call the parent method first. This method will
        create the list self.parameters with the keywords separated by
        underscores in the test name and will remove the ones that were consumed
        for the simulation config. The setup_test methods in the children
        classes can then consume the remaining values.
        """

        super().setup_test()

        # Get list of parameters from the test name
        self.parameters = self.current_test_name.split('_')

        # Remove the 'test' keyword
        self.parameters.remove('test')

        # Decide what type of simulation and instantiate it if needed
        if self.consume_parameter(self.PARAM_SIM_TYPE_LTE):
            self.init_simulation(self.PARAM_SIM_TYPE_LTE)
        elif self.consume_parameter(self.PARAM_SIM_TYPE_LTE_CA):
            self.init_simulation(self.PARAM_SIM_TYPE_LTE_CA)
        elif self.consume_parameter(self.PARAM_SIM_TYPE_LTE_IMS):
            self.init_simulation(self.PARAM_SIM_TYPE_LTE_IMS)
        elif self.consume_parameter(self.PARAM_SIM_TYPE_UMTS):
            self.init_simulation(self.PARAM_SIM_TYPE_UMTS)
        elif self.consume_parameter(self.PARAM_SIM_TYPE_GSM):
            self.init_simulation(self.PARAM_SIM_TYPE_GSM)
        else:
            self.log.error(
                "Simulation type needs to be indicated in the test name.")
            return False

        # Changing cell parameters requires the phone to be detached
        self.simulation.detach()

        # Parse simulation parameters.
        # This may throw a ValueError exception if incorrect values are passed
        # or if required arguments are omitted.
        try:
            self.simulation.parse_parameters(self.parameters)
        except ValueError as error:
            self.log.error(str(error))
            return False

        # Wait for new params to settle
        time.sleep(5)

        # Enable QXDM logger if required
        if self.qxdm_logs:
            self.log.info('Enabling the QXDM logger.')
            telutils.set_qxdm_logger_command(self.dut)
            telutils.start_qxdm_logger(self.dut)

        # Start the simulation. This method will raise an exception if
        # the phone is unable to attach.
        self.simulation.start()

        return True

    def teardown_test(self):
        """ Executed after every test case, even if it failed or an exception
        happened.

        Save results to dictionary so they can be displayed after completing
        the test batch.
        """
        super().teardown_test()

        # If QXDM logging was enabled pull the results
        if self.qxdm_logs:
            self.log.info('Stopping the QXDM logger and pulling results.')
            telutils.stop_qxdm_logger(self.dut)
            self.dut.get_qxdm_logs()

    def consume_parameter(self, parameter_name, num_values=0):
        """ Parses a parameter from the test name.

        Allows the test to get parameters from its name. Deletes parameters from
        the list after consuming them to ensure that they are not used twice.

        Args:
            parameter_name: keyword to look up in the test name
            num_values: number of arguments following the parameter name in the
                test name
        Returns:
            A list containing the parameter name and the following num_values
            arguments.
        """

        try:
            i = self.parameters.index(parameter_name)
        except ValueError:
            # parameter_name is not set
            return []

        return_list = []

        try:
            for j in range(num_values + 1):
                return_list.append(self.parameters.pop(i))
        except IndexError:
            self.log.error(
                "Parameter {} has to be followed by {} values.".format(
                    parameter_name, num_values))
            raise ValueError()

        return return_list

    def teardown_class(self):
        """Clean up the test class after tests finish running.

        Stops the simulation and disconnects from the Anritsu Callbox. Then
        displays the test results.
        """
        super().teardown_class()

        try:
            if self.cellular_simulator:
                self.cellular_simulator.destroy()
        except simulator.CellularSimulatorError as e:
            self.log.error('Error while tearing down the callbox controller. '
                           'Error message: ' + str(e))

    def init_simulation(self, sim_type):
        """ Starts a new simulation only if needed.

        Only starts a new simulation if type is different from the one running
        before.

        Args:
            type: defines the type of simulation to be started.
        """

        simulation_dictionary = {
            self.PARAM_SIM_TYPE_LTE: LteSimulation.LteSimulation,
            self.PARAM_SIM_TYPE_UMTS: UmtsSimulation.UmtsSimulation,
            self.PARAM_SIM_TYPE_GSM: GsmSimulation.GsmSimulation,
            self.PARAM_SIM_TYPE_LTE_CA: LteCaSimulation.LteCaSimulation,
            self.PARAM_SIM_TYPE_LTE_IMS: LteImsSimulation.LteImsSimulation
        }

        if not sim_type in simulation_dictionary:
            raise ValueError("The provided simulation type is invalid.")

        simulation_class = simulation_dictionary[sim_type]

        if isinstance(self.simulation, simulation_class):
            # The simulation object we already have is enough.
            return

        if self.simulation:
            # Make sure the simulation is stopped before loading a new one
            self.simulation.stop()

        # If the calibration table doesn't have an entry for this simulation
        # type add an empty one
        if sim_type not in self.calibration_table:
            self.calibration_table[sim_type] = {}

        cellular_dut = AndroidCellularDut.AndroidCellularDut(
            self.dut, self.log)
        # Instantiate a new simulation
        self.simulation = simulation_class(self.cellular_simulator, self.log,
                                           cellular_dut,
                                           self.cellular_test_params,
                                           self.calibration_table[sim_type])

    def ensure_valid_calibration_table(self, calibration_table):
        """ Ensures the calibration table has the correct structure.

        A valid calibration table is a nested dictionary with non-negative
        number values

        """
        if not isinstance(calibration_table, dict):
            raise TypeError('The calibration table must be a dictionary')
        for val in calibration_table.values():
            if isinstance(val, dict):
                self.ensure_valid_calibration_table(val)
            elif not isinstance(val, float) and not isinstance(val, int):
                raise TypeError('Calibration table value must be a number')
            elif val < 0.0:
                raise ValueError('Calibration table contains negative values')

    def unpack_custom_file(self, file, test_specific=True):
        """Loads a json file.

          Args:
              file: the common file containing pass fail threshold.
              test_specific: if True, returns the JSON element within the file
                  that starts with the test class name.
          """
        with open(file, 'r') as f:
            params = json.load(f)
        if test_specific:
            try:
                return params[self.TAG]
            except KeyError:
                pass
        else:
            return params