# Copyright (c) 2013 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 dbus
import logging
import time

from autotest_lib.client.cros.networking import shill_proxy


class CellularProxy(shill_proxy.ShillProxy):
    """Wrapper around shill dbus interface used by cellular tests."""

    # Properties exposed by shill.
    DEVICE_PROPERTY_DBUS_OBJECT = 'DBus.Object'
    DEVICE_PROPERTY_MODEL_ID = 'Cellular.ModelID'
    DEVICE_PROPERTY_OUT_OF_CREDITS = 'Cellular.OutOfCredits'
    DEVICE_PROPERTY_SIM_LOCK_STATUS = 'Cellular.SIMLockStatus'
    DEVICE_PROPERTY_SIM_PRESENT = 'Cellular.SIMPresent'
    DEVICE_PROPERTY_TECHNOLOGY_FAMILY = 'Cellular.Family'
    DEVICE_PROPERTY_TECHNOLOGY_FAMILY_CDMA = 'CDMA'
    DEVICE_PROPERTY_TECHNOLOGY_FAMILY_GSM = 'GSM'
    SERVICE_PROPERTY_LAST_GOOD_APN = 'Cellular.LastGoodAPN'

    # APN info property names.
    APN_INFO_PROPERTY_APN = 'apn'

    # Keys into the dictionaries exposed as properties.
    PROPERTY_KEY_SIM_LOCK_TYPE = 'LockType'
    PROPERTY_KEY_SIM_LOCK_ENABLED = 'LockEnabled'
    PROPERTY_KEY_SIM_LOCK_RETRIES_LEFT = 'RetriesLeft'

    # Valid values taken by properties exposed by shill.
    VALUE_SIM_LOCK_TYPE_PIN = 'sim-pin'
    VALUE_SIM_LOCK_TYPE_PUK = 'sim-puk'

    # Various timeouts in seconds.
    SERVICE_CONNECT_TIMEOUT = 60
    SERVICE_DISCONNECT_TIMEOUT = 60
    SERVICE_REGISTRATION_TIMEOUT = 60
    SLEEP_INTERVAL = 0.1

    def set_logging_for_cellular_test(self):
        """Set the logging in shill for a test of cellular technology.

        Set the log level to |ShillProxy.LOG_LEVEL_FOR_TEST| and the log scopes
        to the ones defined in |ShillProxy.LOG_SCOPES_FOR_TEST| for
        |ShillProxy.TECHNOLOGY_CELLULAR|.

        """
        self.set_logging_for_test(self.TECHNOLOGY_CELLULAR)


    def find_cellular_service_object(self):
        """Returns the first dbus object found that is a cellular service.

        @return DBus object for the first cellular service found. None if no
                service found.

        """
        return self.find_object('Service', {'Type': self.TECHNOLOGY_CELLULAR})


    def wait_for_cellular_service_object(
            self, timeout_seconds=SERVICE_REGISTRATION_TIMEOUT):
        """Waits for the cellular service object to show up.

        @param timeout_seconds: Amount of time to wait for cellular service.
        @return DBus object for the first cellular service found.
        @raises ShillProxyError if no cellular service is found within the
            registration timeout period.

        """
        CellularProxy._poll_for_condition(
                lambda: self.find_cellular_service_object() is not None,
                'Failed to find cellular service object',
                timeout=timeout_seconds)
        return self.find_cellular_service_object()


    def find_cellular_device_object(self):
        """Returns the first dbus object found that is a cellular device.

        @return DBus object for the first cellular device found. None if no
                device found.

        """
        return self.find_object('Device', {'Type': self.TECHNOLOGY_CELLULAR})


    def reset_modem(self, modem, expect_device=True, expect_powered=True,
                    expect_service=True):
        """Reset |modem|.

        Do, in sequence,
        (1) Ensure that the current device object disappears.
        (2) If |expect_device|, ensure that the device reappears.
        (3) If |expect_powered|, ensure that the device is powered.
        (4) If |expect_service|, ensure that the service reappears.

        This function does not check the service state for the device after
        reset.

        @param modem: DBus object for the modem to reset.
        @param expect_device: If True, ensure that a DBus object reappears for
                the same modem after the reset.
        @param expect_powered: If True, ensure that the modem is powered on
                after the reset.
        @param expect_service: If True, ensure that a service managing the
                reappeared modem also reappears.

        @return (device, service)
                device: DBus object for the reappeared Device after the reset.
                service: DBus object for the reappeared Service after the reset.
                Either of these may be None, if the object is not expected to
                reappear.

        @raises ShillProxyError if any of the conditions (1)-(4) fail.

        """
        logging.info('Resetting modem')
        # Obtain identifying information about the modem.
        properties = modem.GetProperties(utf8_strings=True)
        # NOTE: Using the Model ID means that this will break if we have two
        # identical cellular modems in a DUT. Fortunately, we only support one
        # modem at a time.
        model_id = properties.get(self.DEVICE_PROPERTY_MODEL_ID)
        if not model_id:
            raise shill_proxy.ShillProxyError(
                    'Failed to get identifying information for the modem.')
        old_modem_path = modem.object_path
        old_modem_mm_object = properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT)
        if not old_modem_mm_object:
            raise shill_proxy.ShillProxyError(
                    'Failed to get the mm object path for the modem.')

        modem.Reset()

        # (1) Wait for the old modem to disappear
        CellularProxy._poll_for_condition(
                lambda: self._is_old_modem_gone(old_modem_path,
                                                old_modem_mm_object),
                'Old modem disappeared',
                timeout=60)

        # (2) Wait for the device to reappear
        if not expect_device:
            return None, None
        # The timeout here should be sufficient for our slowest modem to
        # reappear.
        CellularProxy._poll_for_condition(
                lambda: self._get_reappeared_modem(model_id,
                                                   old_modem_mm_object),
                desc='The modem reappeared after reset.',
                timeout=60)
        new_modem = self._get_reappeared_modem(model_id, old_modem_mm_object)

        # (3) Check powered state of the device
        if not expect_powered:
            return new_modem, None
        success, _, _ = self.wait_for_property_in(new_modem,
                                                  self.DEVICE_PROPERTY_POWERED,
                                                  [self.VALUE_POWERED_ON],
                                                  timeout_seconds=10)
        if not success:
            raise shill_proxy.ShillProxyError(
                    'After modem reset, new modem failed to enter powered '
                    'state.')

        # (4) Check that service reappears
        if not expect_service:
            return new_modem, None
        new_service = self.get_service_for_device(new_modem)
        if not new_service:
            raise shill_proxy.ShillProxyError(
                    'Failed to find a shill service managing the reappeared '
                    'device.')
        return new_modem, new_service


    def disable_modem_for_test_setup(self, timeout_seconds=10):
        """
        Disables all cellular modems.

        Use this method only for setting up tests.  Do not use this method to
        test disable functionality because this method repeatedly attempts to
        disable the cellular technology until it succeeds (ignoring all DBus
        errors) since the DisableTechnology() call may fail for various reasons
        (eg. an enable is in progress).

        @param timeout_seconds: Amount of time to wait until the modem is
                disabled.
        @raises ShillProxyError if the modems fail to disable within
                |timeout_seconds|.

        """
        def _disable_cellular_technology(self):
            try:
                self._manager.DisableTechnology(self.TECHNOLOGY_CELLULAR)
                return True
            except dbus.DBusException as e:
                return False

        CellularProxy._poll_for_condition(
                lambda: _disable_cellular_technology(self),
                'Failed to disable cellular technology.',
                timeout=timeout_seconds)
        modem = self.find_cellular_device_object()
        self.wait_for_property_in(modem, self.DEVICE_PROPERTY_POWERED,
                                  [self.VALUE_POWERED_OFF],
                                  timeout_seconds=timeout_seconds)


    # TODO(pprabhu) Use utils.poll_for_condition instead
    @staticmethod
    def _poll_for_condition(condition, desc, timeout=10):
        """Poll till |condition| is satisfied.

        @param condition: A function taking no arguments. The condition is
                met when the return value can be cast to the bool True.
        @param desc: The description given when we timeout waiting for
                |condition|.

        """
        start_time = time.time()
        while True:
            value = condition()
            if value:
                return value
            if(time.time() + CellularProxy.SLEEP_INTERVAL - start_time >
               timeout):
                raise shill_proxy.ShillProxyError(
                        'Timed out waiting for condition %s.' % desc)
            time.sleep(CellularProxy.SLEEP_INTERVAL)


    def _is_old_modem_gone(self, modem_path, modem_mm_object):
        """Tests if the DBus object for modem disappears after Reset.

        @param modem_path: The DBus path for the modem object that must vanish.
        @param modem_mm_object: The modemmanager object path reported by the
            old modem. This is unique everytime a new modem is (re)exposed.

        @return True if the object disappeared, false otherwise.

        """
        device = self.get_dbus_object(self.DBUS_TYPE_DEVICE, modem_path)
        try:
            properties = device.GetProperties()
            # DBus object exists, perhaps a reappeared device?
            return (properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT) !=
                    modem_mm_object)
        except dbus.DBusException as e:
            if e.get_dbus_name() == self.DBUS_ERROR_UNKNOWN_OBJECT:
                return True
            return False


    def _get_reappeared_modem(self, model_id, old_modem_mm_object):
        """Check that a vanished modem reappers.

        @param model_id: The model ID reported by the vanished modem.
        @param old_modem_mm_object: The previously reported modemmanager object
                path for this modem.

        @return The reappeared DBus object, if any. None otherwise.

        """
        # TODO(pprabhu) This will break if we have multiple cellular devices
        # in the system at the same time.
        device = self.find_cellular_device_object()
        if not device:
            return None
        properties = device.GetProperties(utf8_strings=True)
        if (model_id == properties.get(self.DEVICE_PROPERTY_MODEL_ID) and
            (old_modem_mm_object !=
             properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT))):
            return device
        return None