# Copyright (c) 2012 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.

"""Utilities for cellular tests."""
import copy, dbus, os, tempfile

# TODO(thieule): Consider renaming mm.py, mm1.py, modem.py, etc to be more
# descriptive (crosbug.com/37060).
import common
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.cellular import cellular
from autotest_lib.client.cros.cellular import cellular_system_error
from autotest_lib.client.cros.cellular import mm
from autotest_lib.client.cros.cellular import modem

from autotest_lib.client.cros import flimflam_test_path
import flimflam


TIMEOUT = 30
SERVICE_TIMEOUT = 60

import cellular_logging

logger = cellular_logging.SetupCellularLogging('cell_tools')


def ConnectToCellular(flim, timeout=TIMEOUT):
    """Attempts to connect to a cell network using FlimFlam.

    Args:
        flim: A flimflam object
        timeout: Timeout (in seconds) before giving up on connect

    Returns:
        a tuple of the service and the service state

    Raises:
        Error if connection fails or times out
    """

    service = flim.FindCellularService(timeout=timeout)
    if not service:
        raise cellular_system_error.ConnectionFailure(
            'Could not find cell service')
    properties = service.GetProperties(utf8_strings=True)
    logger.error('Properties are: %s', properties)

    logger.info('Connecting to cell service: %s', service)

    states = ['portal', 'online', 'idle']
    state = flim.WaitForServiceState(service=service,
                                     expected_states=states,
                                     timeout=timeout,
                                     ignore_failure=True)[0]
    logger.debug('Cell connection state : %s ' % state)
    connected_states = ['portal', 'online']
    if state in connected_states:
        logger.debug('Looks good, skip ConnectService')
        return service, state
    else:
        logger.debug('Trying to ConnectService')

    success, status = flim.ConnectService(
        service=service,
        assoc_timeout=timeout,
        config_timeout=timeout)

    if not success:
        logger.error('Connect failed: %s' % status)
        # TODO(rochberg):  Turn off autoconnect
        if 'Error.AlreadyConnected' not in status['reason']:
            raise cellular_system_error.ConnectionFailure(
                'Could not connect: %s.' % status)

    state = flim.WaitForServiceState(service=service,
                                     expected_states=connected_states,
                                     timeout=timeout,
                                     ignore_failure=True)[0]
    if not state in connected_states:
        raise cellular_system_error.BadState(
            'Still in state %s, expecting one of: %s ' %
            (state, str(connected_states)))

    return service, state


def FindLastGoodAPN(service, default=None):
    if not service:
        return default
    props = service.GetProperties()
    if 'Cellular.LastGoodAPN' not in props:
        return default
    last_good_apn = props['Cellular.LastGoodAPN']
    return last_good_apn.get('apn', default)


def DisconnectFromCellularService(bs, flim, service):
    """Attempts to disconnect from the supplied cellular service.

    Args:
        bs:  A basestation object.  Pass None to skip basestation-side checks
        flim:  A flimflam object
        service:  A cellular service object
    """

    flim.DisconnectService(service)  # Waits for flimflam state to go to idle

    if bs:
        verifier = bs.GetAirStateVerifier()
        # This is racy: The modem is free to report itself as
        # disconnected before it actually finishes tearing down its RF
        # connection.
        verifier.AssertDataStatusIn([
            cellular.UeGenericDataStatus.DISCONNECTING,
            cellular.UeGenericDataStatus.REGISTERED,
            cellular.UeGenericDataStatus.NONE,])

        def _ModemIsFullyDisconnected():
            return verifier.IsDataStatusIn([
                cellular.UeGenericDataStatus.REGISTERED,
                cellular.UeGenericDataStatus.NONE,])

        utils.poll_for_condition(
            _ModemIsFullyDisconnected,
            timeout=20,
            exception=cellular_system_error.BadState(
                'modem not disconnected from base station'))


def _EnumerateModems(manager):
    """Get a set of modem paths."""
    return set([x[1] for x in mm.EnumerateDevices(manager)])


def _SawNewModem(manager, preexisting_modems, old_modem):
    current_modems = _EnumerateModems(manager)
    if old_modem in current_modems:
        return False
    # NB: This fails if an unrelated modem disappears.  Not fixing
    # until we support > 1 modem
    return preexisting_modems != current_modems


def _WaitForModemToReturn(manager, preexisting_modems_original, modem_path):
    preexisting_modems = copy.copy(preexisting_modems_original)
    preexisting_modems.remove(modem_path)

    utils.poll_for_condition(
        lambda: _SawNewModem(manager, preexisting_modems, modem_path),
        timeout=50,
        exception=cellular_system_error.BadState(
            'Modem did not come back after settings change'))

    current_modems = _EnumerateModems(manager)

    new_modems = [x for x in current_modems - preexisting_modems]
    if len(new_modems) != 1:
        raise cellular_system_error.BadState(
            'Unexpected modem list change: %s vs %s' %
            (current_modems, new_modems))

    logger.info('New modem: %s' % new_modems[0])
    return new_modems[0]


def SetFirmwareForTechnologyFamily(manager, modem_path, family):
    """Set the modem to firmware.  Return potentially-new modem path."""
    # todo(byronk): put this in a modem object?
    if family == cellular.TechnologyFamily.LTE:
        return  # nothing to set up on a Pixel. todo(byronk) how about others?
    logger.debug('SetFirmwareForTechnologyFamily : manager : %s ' % manager)
    logger.debug('SetFirmwareForTechnologyFamily : modem_path : %s ' %
                 modem_path)
    logger.debug('SetFirmwareForTechnologyFamily : family : %s ' % family)
    preexisting_modems = _EnumerateModems(manager)
    # We do not currently support any multi-family modems besides Gobi
    gobi = manager.GetModem(modem_path).GobiModem()
    if not gobi:
        raise cellular_system_error.BadScpiCommand(
            'Modem %s does not support %s, cannot change technologies' %
            modem_path, family)

    logger.info('Changing firmware to technology family %s' % family)

    FamilyToCarrierString = {
            cellular.TechnologyFamily.UMTS: 'Generic UMTS',
            cellular.TechnologyFamily.CDMA: 'Verizon Wireless',}

    gobi.SetCarrier(FamilyToCarrierString[family])
    return _WaitForModemToReturn(manager, preexisting_modems, modem_path)


# A test PRL that has an ID of 3333 and sets the device to aquire the
# default config of an 8960 with system_id 331.  Base64 encoding
# Generated with "base64 < prl"

TEST_PRL_3333 = (
    'ADENBQMAAMAAAYADAgmABgIKDQsEAYAKDUBAAQKWAAICQGAJApYAAgIw8BAAAQDhWA=='.
    decode('base64_codec'))


# A modem with this MDN will always report itself as activated
TESTING_MDN = dbus.String('1115551212', variant_level=1)


def _IsCdmaModemConfiguredCorrectly(manager, modem_path):
    """Returns true iff the CDMA modem at modem_path is configured correctly."""
    # We don't test for systemID because the PRL should take care of
    # that.

    status = manager.GetModem(modem_path).SimpleModem().GetStatus()

    required_settings = {'mdn': TESTING_MDN,
                         'min': TESTING_MDN,
                         'prl_version': 3333}
    configured_correctly = True

    for rk, rv in required_settings.iteritems():
        if rk not in status or rv != status[rk]:
            logger.error('_CheckCdmaModemStatus:  %s: expected %s, got %s' % (
                rk, rv, status.get(rk)))
            configured_correctly = False
    return configured_correctly


def PrepareCdmaModem(manager, modem_path):
    """Configure a CDMA device (including PRL, MIN, and MDN)."""

    if _IsCdmaModemConfiguredCorrectly(manager, modem_path):
        return modem_path

    logger.info('Updating modem settings')
    preexisting_modems = _EnumerateModems(manager)
    cdma = manager.GetModem(modem_path).CdmaModem()

    with tempfile.NamedTemporaryFile() as f:
        os.chmod(f.name, 0744)
        f.write(TEST_PRL_3333)
        f.flush()
        logger.info('Calling ActivateManual to change PRL')

        cdma.ActivateManual({
            'mdn': TESTING_MDN,
            'min': TESTING_MDN,
            'prlfile': dbus.String(f.name, variant_level=1),
            'system_id': dbus.UInt16(331, variant_level=1),  # Default 8960 SID
            'spc': dbus.String('000000'),})
        new_path = _WaitForModemToReturn(
            manager, preexisting_modems, modem_path)

    if not _IsCdmaModemConfiguredCorrectly(manager, new_path):
        raise cellular_system_error.BadState('Modem configuration failed')
    return new_path


def PrepareModemForTechnology(modem_path, target_technology):
    """Prepare modem for the technology: Sets things like firmware, PRL."""

    manager, modem_path = mm.PickOneModem(modem_path)

    logger.info('Found modem %s' % modem_path)


    # todo(byronk) : This returns TechnologyFamily:UMTS on a Pixel. ????
    current_family = manager.GetModem(modem_path).GetCurrentTechnologyFamily()
    target_family = cellular.TechnologyToFamily[target_technology]

    if current_family != target_family:
        logger.debug('Modem Current Family: %s ' % current_family)
        logger.debug('Modem Target Family : %s ' %target_family )
        modem_path = SetFirmwareForTechnologyFamily(
            manager, modem_path, target_family)

    if target_family == cellular.TechnologyFamily.CDMA:
        modem_path = PrepareCdmaModem(manager, modem_path)
        # Force the modem to report that is has been activated since we
        # use a custom PRL and have already manually activated it.
        manager.GetModem(modem_path).GobiModem().ForceModemActivatedStatus()

    # When testing EVDO, we need to force the modem to register with EVDO
    # directly (bypassing CDMA 1x RTT) else the modem will not register
    # properly because it looks for CDMA 1x RTT first but can't find it
    # because the call box can only emulate one technology at a time (EVDO).
    try:
        if target_technology == cellular.Technology.EVDO_1X:
            network_preference = modem.Modem.NETWORK_PREFERENCE_EVDO_1X
        else:
            network_preference = modem.Modem.NETWORK_PREFERENCE_AUTOMATIC
        gobi = manager.GetModem(modem_path).GobiModem()
        gobi.SetNetworkPreference(network_preference)
    except AttributeError:
        # Not a Gobi modem
        pass

    return modem_path


def FactoryResetModem(modem_pattern, spc='000000'):
    """Factory resets modem, returns DBus pathname of modem after reset."""
    manager, modem_path = mm.PickOneModem(modem_pattern)
    preexisting_modems = _EnumerateModems(manager)
    modem = manager.GetModem(modem_path).Modem()
    modem.FactoryReset(spc)
    return _WaitForModemToReturn(manager, preexisting_modems, modem_path)


class OtherDeviceShutdownContext(object):
    """Context manager that shuts down other devices.

    Usage:
        with cell_tools.OtherDeviceShutdownContext('cellular'):
            block

    TODO(rochberg):  Replace flimflam.DeviceManager with this
    """

    def __init__(self, device_type):
        self.device_type = device_type
        self.device_manager = None

    def __enter__(self):
        self.device_manager = flimflam.DeviceManager(flimflam.FlimFlam())
        self.device_manager.ShutdownAllExcept(self.device_type)
        return self

    def __exit__(self, exception, value, traceback):
        if self.device_manager:
            self.device_manager.RestoreDevices()
        return False


class AutoConnectContext(object):
    """Context manager which sets autoconnect to either true or false.

       Enable or Disable autoconnect for the cellular service.
       Restore it when done.

       Usage:
           with cell_tools.DisableAutoConnectContext(device, flim, autoconnect):
               block
    """

    def __init__(self, device, flim, autoconnect):
        self.device = device
        self.flim = flim
        self.autoconnect = autoconnect
        self.autoconnect_changed = False

    def PowerOnDevice(self, device):
        """Power on a flimflam device, ignoring in progress errors."""
        logger.info('powered = %s' % device.GetProperties()['Powered'])
        if device.GetProperties()['Powered']:
            return
        try:
            device.Enable()
        except dbus.exceptions.DBusException, e:
            if e._dbus_error_name != 'org.chromium.flimflam.Error.InProgress':
                raise e

    def __enter__(self):
        """Power up device, get the service and disable autoconnect."""
        changed = False
        self.PowerOnDevice(self.device)

        # Use SERVICE_TIMEOUT*2 here because it may take SERVICE_TIMEOUT
        # seconds for the modem to disconnect when the base emulator is taken
        # offline for reconfiguration and then another SERVICE_TIMEOUT
        # seconds for the modem to reconnect after the base emulator is
        # brought back online.
        #
        # TODO(jglasgow): generalize to use services associated with device
        service = self.flim.FindCellularService(timeout=SERVICE_TIMEOUT*2)
        if not service:
            raise error.TestFail('No cellular service available.')

        # Always set the AutoConnect property even if the requested value
        # is the same so that shill will retain the AutoConnect property, else
        # shill may override it.
        props = service.GetProperties()
        autoconnect = props['AutoConnect']
        logger.info('AutoConnect = %s' % autoconnect)
        logger.info('Setting AutoConnect = %s.', self.autoconnect)
        service.SetProperty('AutoConnect', dbus.Boolean(self.autoconnect))

        if autoconnect != self.autoconnect:
            props = service.GetProperties()
            autoconnect = props['AutoConnect']
            changed = True

        # Make sure the cellular service gets persisted by taking it out of
        # the ephemeral profile.
        if not props['Profile']:
            manager_props = self.flim.manager.GetProperties()
            active_profile = manager_props['ActiveProfile']
            logger.info("Setting cellular service profile to %s",
                        active_profile)
            service.SetProperty('Profile', active_profile)

        if autoconnect != self.autoconnect:
            raise error.TestFail('AutoConnect is %s, but we want it to be %s' %
                                 (autoconnect, self.autoconnect))

        self.autoconnect_changed = changed

        return self

    def __exit__(self, exception, value, traceback):
        """Restore autoconnect state if we changed it."""
        if not self.autoconnect_changed:
            return False

        try:
            self.PowerOnDevice(self.device)
        except Exception as e:
            if exception:
                logger.error(
                    'Exiting AutoConnectContext with one exception, but ' +
                    'PowerOnDevice raised another')
                logger.error(
                    'Swallowing PowerOnDevice exception %s' % e)
                return False
            else:
                raise e

        # TODO(jglasgow): generalize to use services associated with
        # device, and restore state only on changed services
        service = self.flim.FindCellularService()
        if not service:
            logger.error('Cannot find cellular service.  '
                          'Autoconnect state not restored.')
            return False
        service.SetProperty('AutoConnect', dbus.Boolean(not self.autoconnect))

        return False