xref: /aosp_15_r20/external/autotest/client/cros/cellular/test_environment.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import contextlib
import dbus
import logging
import sys
import time
import traceback

import common
from autotest_lib.client.bin import local_host
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import crash_detector
from autotest_lib.client.cros import upstart
from autotest_lib.client.cros.cellular import mm
from autotest_lib.client.cros.cellular import mm1_constants
from autotest_lib.client.cros.networking import cellular_proxy
from autotest_lib.client.cros.networking import mm1_proxy
from autotest_lib.client.cros.networking import shill_context
from autotest_lib.client.cros.networking import shill_proxy


class CellularTestEnvironment(object):
    """Setup and verify cellular test environment.

    This context manager configures the following:
        - Shuts down other devices except cellular.
        - Shill and MM logging is enabled appropriately for cellular.
        - Initializes members that tests should use to access test environment
          (eg. |shill|, |modem_manager|, |modem|).
        - modemfwd is stopped to prevent the modem from rebooting underneath
          us.

    Then it verifies the following is valid:
        - The SIM is inserted and valid.
        - There is one and only one modem in the device.
        - The modem is registered to the network.
        - There is a cellular service in shill and it's not connected.

    Don't use this base class directly, use the appropriate subclass.

    Setup for over-the-air tests:
        with CellularOTATestEnvironment() as test_env:
            # Test body

    Setup for pseudomodem tests:
        with CellularPseudoMMTestEnvironment(
                pseudomm_args=({'family': '3GPP'})) as test_env:
            # Test body

    """

    def __init__(self,
                 shutdown_other_devices=True,
                 modem_pattern='',
                 skip_modem_reset=False,
                 is_esim_test=False,
                 enable_temp_containments=True):
        """
        @param shutdown_other_devices: If True, shutdown all devices except
                cellular.
        @param modem_pattern: Search string used when looking for the modem.
        @param enable_temp_containments: Enable temporary containments to avoid
                failures on tests with known problems.

        """
        # Tests should use this main loop instead of creating their own.
        self.mainloop = dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
        self.bus = dbus.SystemBus(mainloop=self.mainloop)

        self.shill = None
        self.modem_manager = None
        self.modem = None
        self.modem_path = None

        self._modem_pattern = modem_pattern
        self._skip_modem_reset = skip_modem_reset
        self._is_esim_test = is_esim_test
        self._enable_temp_containments = enable_temp_containments
        self._system_service_order = ''
        self._test_service_order = 'cellular,ethernet'

        self._nested = None
        self._context_managers = []
        self.detect_crash = crash_detector.CrashDetector(
                local_host.LocalHost())
        self.detect_crash.remove_crash_files()
        if shutdown_other_devices:
            self._context_managers.append(
                    shill_context.AllowedTechnologiesContext([
                            shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR,
                            shill_proxy.ShillProxy.TECHNOLOGY_ETHERNET
                    ]))

    @contextlib.contextmanager
    def _disable_shill_autoconnect(self):
        self._enable_shill_cellular_autoconnect(False)
        yield
        self._enable_shill_cellular_autoconnect(True)

    def __enter__(self):
        try:
            # Wait for system daemons to stabilize before beginning the test.
            # Modemfwd, Chrome, Shill and Hermes might be active before the test
            # begins, and interrupting them abruptly during test setup might
            # lead to flaky tests. The modem might also appear/disappear
            # multiple times during this period. Ideally, we would wait for a
            # green signal from these daemons before performing test setup.
            with open('/proc/uptime') as uptime_file:
                uptime = float(uptime_file.readline().split()[0])
            if uptime < 60:
                logging.info(
                        "Waiting %.1f seconds to reach uptime of 1 minute before "
                        "starting test", 60 - uptime)
                time.sleep(60 - uptime)

            if upstart.has_service('modemfwd') and upstart.is_running('modemfwd'):
                # Due to b/179796133, stopping modemfwd right after it was
                # started by a previous test, can wedge the modem. In many
                # devices, a ~1 second delay solves the problem.
                time.sleep(4)
                upstart.stop_job('modemfwd')
            # Temporarily disable shill autoconnect to cellular service while
            # the test environment is setup to prevent a race condition
            # between disconnecting the modem in _verify_cellular_service()
            # and shill autoconnect.
            with self._disable_shill_autoconnect():
                try:
                    from contextlib import nested # Python 2
                except ImportError:
                    from contextlib import ExitStack, contextmanager

                    @contextmanager
                    def nested(*contexts):
                        """ Implementation of nested for python3"""
                        with ExitStack() as stack:
                            for ctx in contexts:
                                stack.enter_context(ctx)
                            yield contexts

                self._nested = nested(*self._context_managers)

                self._nested.__enter__()

                self._initialize_shill()

                # Perform SIM verification now to ensure that we can enable the
                # modem in _initialize_modem_components(). ModemManager does not
                # allow enabling a modem without a SIM.
                self._verify_sim()
                self._initialize_modem_components()

                self._setup_logging()

                if not self._is_esim_test:
                    self._wait_for_modem_registration()
                self._verify_cellular_service()

                return self
        except (error.TestError, dbus.DBusException,
                shill_proxy.ShillProxyError) as e:
            except_type, except_value, except_traceback = sys.exc_info()
            lines = traceback.format_exception(except_type, except_value,
                                               except_traceback)
            logging.error('Error during test initialization:\n%s',
                          ''.join(lines))
            self.__exit__(*sys.exc_info())
            raise error.TestError('INIT_ERROR: %s' % str(e))
        except:
            self.__exit__(*sys.exc_info())
            raise

    def __exit__(self, exception, value, traceback):
        exception_on_restore_state = None
        try:
            self._restore_state()
        except Exception as ex:
            # Exceptions thrown by _restore_state() should be ignored if a
            # previous exception exist, otherwise the root cause of the test
            # failure will be overwritten by the clean up error in
            # _restore_state, and that is not useful.
            if exception is None:
                exception_on_restore_state = ex

        # If a test fails and a crash is detected, the crash error takes
        # priority over the previous failure.
        crash_files = self.detect_crash.get_new_crash_files()
        if any(cf for cf in crash_files if any(pr in cf for pr in [
                'ModemManager', 'shill', 'qmi', 'mbim', 'hermes', 'modemfwd'
        ])):
            logging.info(
                    'A crash was encountered. '
                    'Overriding the previous error: %s', value)
            raise error.TestError(
                    'One or more daemon crashes were detected. '
                    'See crash dumps: {}'.format(crash_files))

        if exception_on_restore_state is not None:
            raise exception_on_restore_state

        if self._nested:
            return self._nested.__exit__(exception, value, traceback)
        self.shill = None
        self.modem_manager = None
        self.modem = None
        self.modem_path = None

    def _restore_state(self):
        """Try to restore the test environment to a good state.
        """
        if upstart.has_service('modemfwd'):
            upstart.restart_job('modemfwd')
        if self.shill:
            self._set_service_order(self._system_service_order)

    def _get_shill_cellular_device_object(self):
        return utils.poll_for_condition(
            lambda: self.shill.find_cellular_device_object(),
            exception=error.TestError('Cannot find cellular device in shill. '
                                      'Is the modem plugged in?'),
            timeout=shill_proxy.ShillProxy.DEVICE_ENUMERATION_TIMEOUT)

    def _get_service_order(self):
        """Get the shill service order.

        @return string service order on success, None otherwise.

        """
        return str(self.shill.manager.GetServiceOrder())

    def _set_service_order(self, order):
        """Set the shill service order.

        @param order string comma-delimited service order
        (eg. 'cellular,ethernet')
        @return bool True on success, False otherwise.

        """
        self.shill.manager.SetServiceOrder(dbus.String(order))
        return True

    def _enable_modem(self):
        modem_device = self._get_shill_cellular_device_object()
        try:
            modem_device.Enable()
        except dbus.DBusException as e:
            if (e.get_dbus_name() !=
                    shill_proxy.ShillProxy.ERROR_IN_PROGRESS):
                raise

        utils.poll_for_condition(
            lambda: modem_device.GetProperties()['Powered'],
            exception=error.TestError(
                'Failed to enable modem.'),
            timeout=shill_proxy.ShillProxy.DEVICE_ENABLE_DISABLE_TIMEOUT)

    def _enable_shill_cellular_autoconnect(self, enable):
        shill = cellular_proxy.CellularProxy.get_proxy(self.bus)
        shill.manager.SetProperty(
            shill_proxy.ShillProxy.
            MANAGER_PROPERTY_NO_AUTOCONNECT_TECHNOLOGIES,
            '' if enable else 'cellular')

    def _is_unsupported_error(self, e):
        return (e.get_dbus_name() ==
                shill_proxy.ShillProxy.ERROR_NOT_SUPPORTED or
                (e.get_dbus_name() ==
                 shill_proxy.ShillProxy.ERROR_FAILURE and
                 'operation not supported' in e.get_dbus_message()))

    def _reset_modem(self):
        modem_device = self._get_shill_cellular_device_object()
        try:
            # MBIM modems do not support being reset.
            self.shill.reset_modem(modem_device, expect_service=False)
        except dbus.DBusException as e:
            if not self._is_unsupported_error(e):
                raise

    def _initialize_shill(self):
        """Get access to shill."""
        # CellularProxy.get_proxy() checks to see if shill is running and
        # responding to DBus requests. It returns None if that's not the case.
        self.shill = cellular_proxy.CellularProxy.get_proxy(self.bus)
        if self.shill is None:
            raise error.TestError('Cannot connect to shill, is shill running?')

        self._system_service_order = self._get_service_order()
        self._set_service_order(self._test_service_order)

    def _initialize_modem_components(self):
        """Reset the modem and get access to modem components."""
        # Enable modem first so shill initializes the modemmanager proxies so
        # we can call reset on it.
        self._enable_modem()
        if not self._skip_modem_reset:
            self._reset_modem()

        # PickOneModem() makes sure there's a modem manager and that there is
        # one and only one modem.
        self.modem_manager, self.modem_path = \
            mm.PickOneModem(self._modem_pattern)
        self.modem = self.modem_manager.GetModem(self.modem_path)
        if self.modem is None:
            raise error.TestError('Cannot get modem object at %s.' %
                                  self.modem_path)

    def _setup_logging(self):
        self.shill.set_logging_for_cellular_test()
        self.modem_manager.SetDebugLogging()

    def _verify_sim(self):
        """Verify SIM is valid.

        Make sure a SIM in inserted and that it is not locked.

        @raise error.TestError if SIM does not exist or is locked.

        """
        # check modem SIM slot and properties and switch slot as needed
        modem_proxy = self._check_for_modem_with_sim()
        if modem_proxy is None:
            raise error.TestError('There is no Modem with non empty SIM path.')

        modem_device = self._get_shill_cellular_device_object()
        props = modem_device.GetProperties()

        # No SIM in CDMA modems.
        family = props[
            cellular_proxy.CellularProxy.DEVICE_PROPERTY_TECHNOLOGY_FAMILY]
        if (family ==
                cellular_proxy.CellularProxy.
                DEVICE_PROPERTY_TECHNOLOGY_FAMILY_CDMA):
            return

        # Make sure there is a SIM.
        if not props[cellular_proxy.CellularProxy.DEVICE_PROPERTY_SIM_PRESENT]:
            raise error.TestError('There is no SIM in the modem.')

        # Make sure SIM is not locked.
        lock_status = props.get(
            cellular_proxy.CellularProxy.DEVICE_PROPERTY_SIM_LOCK_STATUS,
            None)
        if lock_status is None:
            raise error.TestError('Failed to read SIM lock status.')
        locked = lock_status.get(
            cellular_proxy.CellularProxy.PROPERTY_KEY_SIM_LOCK_ENABLED,
            None)
        if locked is None:
            raise error.TestError('Failed to read SIM LockEnabled status.')
        elif locked:
            raise error.TestError(
                'SIM is locked, test requires an unlocked SIM.')

    def _check_for_modem_with_sim(self):
        """
        Make sure modem got active SIM and path is not empty

        switch slot to get non empty sim path and active sim slot for modem

        @return active modem object or None

        """
        mm_proxy = mm1_proxy.ModemManager1Proxy.get_proxy()
        if mm_proxy is None:
            raise error.TestError('Modem manager is not initialized')

        modem_proxy = mm_proxy.wait_for_modem(mm1_constants.MM_MODEM_POLL_TIME)
        if modem_proxy is None:
            raise error.TestError('Modem not initialized')

        primary_slot = modem_proxy.get_primary_sim_slot()
        # Get SIM path from modem SIM properties
        modem_props = modem_proxy.properties(mm1_constants.I_MODEM)
        sim_path = modem_props['Sim']

        logging.info('Device SIM values=> path:%s '
                'primary slot:%d', sim_path, primary_slot)

        def is_usable_sim(path):
            """Check if sim at path can be used to establish a connection"""
            if path == mm1_constants.MM_EMPTY_SLOT_PATH:
                return False
            sim_proxy = modem_proxy.get_sim_at_path(path)
            sim_props = sim_proxy.properties()
            return sim_props[
                    'EsimStatus'] != mm1_constants.MM_SIM_ESIM_STATUS_NO_PROFILES

        # Check current SIM path value and status
        if is_usable_sim(sim_path):
            return modem_proxy

        slots = modem_props['SimSlots']
        logging.info('Dut not in expected state, '
                    'current sim path:%s slots:%s', sim_path, slots)

        for idx, path in enumerate(slots):
            if not is_usable_sim(path):
                continue
            logging.info('Primary slot does not have a SIM, '
                        'switching slot to %d', idx+1)

            if (primary_slot != idx + 1):
                logging.info('setting slot:%d path:%s', idx+1, path)
                modem_proxy.set_primary_slot(idx+1)
                modem_proxy = \
                    mm_proxy.wait_for_modem(mm1_constants.MM_MODEM_POLL_TIME)
                return modem_proxy
        return None

    def _wait_for_modem_registration(self):
        """Wait for the modem to register with the network.

        @raise error.TestError if modem is not registered.

        """
        utils.poll_for_condition(
            self.modem.ModemIsRegistered,
            exception=error.TestError(
                'Modem failed to register with the network.'),
            timeout=cellular_proxy.CellularProxy.SERVICE_REGISTRATION_TIMEOUT)

    def _verify_cellular_service(self):
        """Make sure a cellular service exists.

        The cellular service should not be connected to the network.

        @raise error.TestError if cellular service does not exist or if
                there are multiple cellular services.

        """
        service = self.shill.wait_for_cellular_service_object()

        try:
            service.Disconnect()
        except dbus.DBusException as e:
            if (e.get_dbus_name() !=
                    cellular_proxy.CellularProxy.ERROR_NOT_CONNECTED):
                raise
        success, state, _ = self.shill.wait_for_property_in(
            service,
            cellular_proxy.CellularProxy.SERVICE_PROPERTY_STATE,
            ('idle',),
            cellular_proxy.CellularProxy.SERVICE_DISCONNECT_TIMEOUT)
        if not success:
            raise error.TestError(
                'Cellular service needs to start in the "idle" state. '
                'Current state is "%s". '
                'Modem disconnect may have failed.' %
                state)


class CellularOTATestEnvironment(CellularTestEnvironment):
    """Setup and verify cellular over-the-air (OTA) test environment. """

    def __init__(self, **kwargs):
        super(CellularOTATestEnvironment, self).__init__(**kwargs)

# pseudomodem tests disabled with b/180627893, cleaningup all pseudomodem
# related files and imports through: b/205769777
'''
class CellularPseudoMMTestEnvironment(CellularTestEnvironment):
    """Setup and verify cellular pseudomodem test environment. """

    def __init__(self, pseudomm_args=None, **kwargs):
        """
        @param pseudomm_args: Tuple of arguments passed to the pseudomodem, see
                pseudomodem_context.py for description of each argument in the
                tuple: (flags_map, block_output, bus)

        """
        kwargs["skip_modem_reset"] = True
        super(CellularPseudoMMTestEnvironment, self).__init__(**kwargs)
        self._context_managers.append(
            pseudomodem_context.PseudoModemManagerContext(
                True, bus=self.bus, *pseudomm_args))
'''

class CellularESIMTestEnvironment(CellularTestEnvironment):
    """Setup cellular eSIM test environment. """

    def __init__(self, esim_arguments=None, **kwargs):
        kwargs["skip_modem_reset"] = True
        kwargs["is_esim_test"] = True
        super(CellularESIMTestEnvironment, self).__init__(**kwargs)