# 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 contextlib
import dbus
import logging
import random
import time

from autotest_lib.client.bin import test, utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import chrome
from autotest_lib.client.cros.cellular import cellular
from autotest_lib.client.cros.networking import cellular_proxy
from autotest_lib.client.cros.networking import shill_context
from autotest_lib.client.cros.networking import shill_proxy

# Number of seconds we wait for the cellular service to perform an action.
DEVICE_TIMEOUT=45
SERVICE_TIMEOUT=75

# Number of times and seconds between modem state checks to ensure that the
# modem is not in a temporary transition state.
NUM_MODEM_STATE_CHECKS=2
MODEM_STATE_CHECK_PERIOD_SECONDS=5

# Number of seconds to sleep after a connect request in slow-connect mode.
SLOW_CONNECT_WAIT_SECONDS=20


class TechnologyCommands():
    """Control the modem mostly using shill Technology interfaces."""
    def __init__(self, shill, command_delegate):
        self.shill = shill
        self.command_delegate = command_delegate

    def Enable(self):
        self.shill.manager.EnableTechnology(
                shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)

    def Disable(self):
        self.shill.manager.DisableTechnology(
                shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)

    def Connect(self, **kwargs):
        self.command_delegate.Connect(**kwargs)

    def Disconnect(self):
        return self.command_delegate.Disconnect()

    def __str__(self):
        return 'Technology Commands'


class ModemCommands():
    """Control the modem using modem manager DBUS interfaces."""
    def __init__(self, modem, slow_connect):
        self.modem = modem
        self.slow_connect = slow_connect

    def Enable(self):
        self.modem.Enable(True)

    def Disable(self):
        self.modem.Enable(False)

    def Connect(self, simple_connect_props):
        logging.debug('Connecting with properties: %r' % simple_connect_props)
        self.modem.Connect(simple_connect_props)
        if self.slow_connect:
            time.sleep(SLOW_CONNECT_WAIT_SECONDS)

    def Disconnect(self):
        """
        Disconnect Modem.

        Returns:
            True - to indicate that shill may autoconnect again.
        """
        try:
            self.modem.Disconnect()
        except dbus.DBusException as e:
            if (e.get_dbus_name() !=
                    'org.chromium.ModemManager.Error.OperationInitiated'):
                raise e
        return True

    def __str__(self):
        return 'Modem Commands'


class DeviceCommands():
    """Control the modem using shill device interfaces."""
    def __init__(self, shill, device, slow_connect):
        self.shill = shill
        self.device = device
        self.slow_connect = slow_connect
        self.service = None

    def GetService(self):
        service = self.shill.find_cellular_service_object()
        if not service:
            raise error.TestFail(
                'Service failed to appear when using device commands.')
        return service

    def Enable(self):
        self.device.Enable(timeout=DEVICE_TIMEOUT)

    def Disable(self):
        self.service = None
        self.device.Disable(timeout=DEVICE_TIMEOUT)

    def Connect(self, **kwargs):
        self.GetService().Connect()
        if self.slow_connect:
            time.sleep(SLOW_CONNECT_WAIT_SECONDS)

    def Disconnect(self):
        """
        Disconnect Modem.

        Returns:
            False - to indicate that shill may not autoconnect again.
        """
        self.GetService().Disconnect()
        return False

    def __str__(self):
        return 'Device Commands'


class MixedRandomCommands():
    """Control the modem using a mixture of commands on device, modems, etc."""
    def __init__(self, commands_list):
        self.commands_list = commands_list

    def PickRandomCommands(self):
        return self.commands_list[random.randrange(len(self.commands_list))]

    def Enable(self):
        cmds = self.PickRandomCommands()
        logging.info('Enable with %s' % cmds)
        cmds.Enable()

    def Disable(self):
        cmds = self.PickRandomCommands()
        logging.info('Disable with %s' % cmds)
        cmds.Disable()

    def Connect(self, **kwargs):
        cmds = self.PickRandomCommands()
        logging.info('Connect with %s' % cmds)
        cmds.Connect(**kwargs)

    def Disconnect(self):
        cmds = self.PickRandomCommands()
        logging.info('Disconnect with %s' % cmds)
        return cmds.Disconnect()

    def __str__(self):
        return 'Mixed Commands'


class cellular_ModemControl(test.test):
    version = 1

    def CompareModemPowerState(self, modem, expected_state):
        """Compare modem manager power state of a modem to an expected state."""
        return modem.IsEnabled() == expected_state

    def CompareDevicePowerState(self, device, expected_state):
        """Compare the shill device power state to an expected state."""
        state = self.test_env.shill.get_dbus_property(
                device, shill_proxy.ShillProxy.DEVICE_PROPERTY_POWERED)
        logging.info('Device Enabled = %s' % state)
        return state == expected_state

    def CompareServiceState(self, service, expected_states):
        """Compare the shill service state to a set of expected states."""
        if not service:
            logging.info('Service not found.')
            return False

        state = self.test_env.shill.get_dbus_property(
                service, shill_proxy.ShillProxy.SERVICE_PROPERTY_STATE)
        logging.info('Service State = %s' % state)
        return state in expected_states

    def EnsureNotConnectingOrDisconnecting(self):
        """
        Ensure modem is not connecting or disconnecting.

        Raises:
            error.TestFail if it timed out waiting for the modem to finish
            connecting or disconnecting.
        """
        # Shill retries a failed connect attempt with a different APN so
        # check a few times to ensure the modem is not in between connect
        # attempts.
        for _ in range(NUM_MODEM_STATE_CHECKS):
            utils.poll_for_condition(
                lambda: not self.test_env.modem.IsConnectingOrDisconnecting(),
                error.TestFail('Timed out waiting for modem to finish ' +
                               'connecting or disconnecting.'),
                timeout=SERVICE_TIMEOUT)
            time.sleep(MODEM_STATE_CHECK_PERIOD_SECONDS)

    def EnsureDisabled(self):
        """
        Ensure modem disabled, device powered off, and no service.

        Raises:
            error.TestFail if the states are not consistent.
        """
        utils.poll_for_condition(
            lambda: self.CompareModemPowerState(self.test_env.modem, False),
            error.TestFail('Modem failed to enter state Disabled.'))
        utils.poll_for_condition(
            lambda: self.CompareDevicePowerState(self.device, False),
            error.TestFail('Device failed to enter state Powered=False.'))
        utils.poll_for_condition(
            lambda: not self.test_env.shill.find_cellular_service_object(),
            error.TestFail('Service should not be available.'),
            timeout=SERVICE_TIMEOUT)

    def EnsureEnabled(self, check_idle):
        """
        Ensure modem enabled, device powered and service exists.

        Args:
            check_idle: if True, then ensure that the service is idle
                        (i.e. not connected) otherwise ignore the
                        service state

        Raises:
            error.TestFail if the states are not consistent.
        """
        utils.poll_for_condition(
            lambda: self.CompareModemPowerState(self.test_env.modem, True),
            error.TestFail('Modem failed to enter state Enabled'))
        utils.poll_for_condition(
            lambda: self.CompareDevicePowerState(self.device, True),
            error.TestFail('Device failed to enter state Powered=True.'),
            timeout=30)

        service = self.test_env.shill.wait_for_cellular_service_object()
        if check_idle:
            utils.poll_for_condition(
                lambda: self.CompareServiceState(service, ['idle']),
                error.TestFail('Service failed to enter idle state.'),
                timeout=SERVICE_TIMEOUT)

    def EnsureConnected(self):
        """
        Ensure modem connected, device powered on, service connected.

        Raises:
            error.TestFail if the states are not consistent.
        """
        self.EnsureEnabled(check_idle=False)
        utils.poll_for_condition(
            lambda: self.CompareServiceState(
                    self.test_env.shill.find_cellular_service_object(),
                    ['ready', 'portal', 'online']),
            error.TestFail('Service failed to connect.'),
            timeout=SERVICE_TIMEOUT)


    def TestCommands(self, commands):
        """
        Manipulate the modem using modem, device or technology commands.

        Changes the state of the modem in various ways including
        disable while connected and then verifies the state of the
        modem manager and shill.

        Raises:
            error.TestFail if the states are not consistent.

        """
        logging.info('Testing using %s' % commands)

        logging.info('Enabling')
        commands.Enable()
        self.EnsureEnabled(check_idle=not self.autoconnect)

        technology_family = self.test_env.modem.GetCurrentTechnologyFamily()
        if technology_family == cellular.TechnologyFamily.CDMA:
            simple_connect_props = {'number': r'#777'}
        else:
            simple_connect_props = {'number': r'#777', 'apn': self.FindAPN()}

        # Icera modems behave weirdly if we cancel the operation while the
        # modem is connecting. Work around the issue by waiting until the
        # connect operation completes.
        # TODO(benchan): Remove this workaround once the issue is addressed
        # on the modem side.
        self.EnsureNotConnectingOrDisconnecting()

        logging.info('Disabling')
        commands.Disable()
        self.EnsureDisabled()

        logging.info('Enabling again')
        commands.Enable()
        self.EnsureEnabled(check_idle=not self.autoconnect)

        if not self.autoconnect:
            logging.info('Connecting')
            commands.Connect(simple_connect_props=simple_connect_props)
        else:
            logging.info('Expecting AutoConnect to connect')
        self.EnsureConnected()

        logging.info('Disconnecting')
        will_autoreconnect = commands.Disconnect()

        if not (self.autoconnect and will_autoreconnect):
            # Icera modems behave weirdly if we cancel the operation while the
            # modem is disconnecting. Work around the issue by waiting until
            # the disconnect operation completes.
            # TODO(benchan): Remove this workaround once the issue is addressed
            # on the modem side.
            self.EnsureNotConnectingOrDisconnecting()

            self.EnsureEnabled(check_idle=True)
            logging.info('Connecting manually, since AutoConnect was on')
            commands.Connect(simple_connect_props=simple_connect_props)
        self.EnsureConnected()

        logging.info('Disabling')
        commands.Disable()
        self.EnsureDisabled()

    def FindAPN(self):
        default = 'None'
        service = self.test_env.shill.find_cellular_service_object()
        last_good_apn = self.test_env.shill.get_dbus_property(
                service,
                cellular_proxy.CellularProxy.SERVICE_PROPERTY_LAST_GOOD_APN)
        if not last_good_apn:
            return default
        return last_good_apn.get(
                cellular_proxy.CellularProxy.APN_INFO_PROPERTY_APN, default)

    def run_once(self, test_env, autoconnect, mixed_iterations=2,
                 slow_connect=False):
        self.test_env = test_env
        self.autoconnect = autoconnect

        with test_env:
            self.device = self.test_env.shill.find_cellular_device_object()

            modem_commands = ModemCommands(self.test_env.modem,
                                           slow_connect)
            technology_commands = TechnologyCommands(self.test_env.shill,
                                                     modem_commands)
            device_commands = DeviceCommands(self.test_env.shill,
                                             self.device,
                                             slow_connect)

            # shill disables autoconnect on any cellular service before a user
            # logs in (CL:851267). To test the autoconnect scenario, we need a
            # user session to run the test.
            chrome_context = chrome.Chrome()

            # Set up the autoconnect context after starting a user session so
            # that we ensure the autoconnect property is set on the cellular
            # service that may be in the user profile.
            autoconnect_context = shill_context.ServiceAutoConnectContext(
                    self.test_env.shill.wait_for_cellular_service_object,
                    self.autoconnect)

            with contextlib.nested(chrome_context, autoconnect_context):
                # Start with cellular disabled.
                self.test_env.shill.manager.DisableTechnology(
                        shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)
                self.EnsureDisabled()

                # Run the device commands test first to make sure we have
                # a valid APN needed to connect using the modem commands.
                self.TestCommands(device_commands)
                self.TestCommands(technology_commands)
                self.TestCommands(modem_commands)

                # Run several times using commands mixed from each type
                mixed = MixedRandomCommands([modem_commands,
                                             technology_commands,
                                             device_commands])
                for _ in range(mixed_iterations):
                    self.TestCommands(mixed)

                # Ensure cellular is re-enabled in order to restore AutoConnect
                # settings when ServiceAutoConnectContext exits.
                # TODO(benchan): Refactor this logic into
                # ServiceAutoConnectContext and update other users of
                # ServiceAutoConnectContext.
                self.test_env.shill.manager.EnableTechnology(
                        shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)
                self.EnsureEnabled(check_idle=False)