client/site_tests/network_WiMaxSmoke/network_WiMaxSmoke.py

# 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 logging
import time
import urlparse

from autotest_lib.client.bin import test
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import backchannel, network
from autotest_lib.client.cros.cellular import cell_tools
from autotest_lib.client.cros.networking import shill_context

# Import 'flimflam_test_path' first in order to import 'flimflam'.
# pylint: disable=W0611
from autotest_lib.client.cros import flimflam_test_path
# pylint: enable=W0611
import flimflam


# TODO(benchan): Use the log scopes defined in ShillProxy after
# migrating from FlimFlam to ShillProxy.
SHILL_LOG_SCOPES = 'dbus+device+dhcp+manager+portal+service+wimax'


class network_WiMaxSmoke(test.test):
    """Verifies that a WiMAX module can connect to a WiMAX network.

    The test attempts to connect to a WiMAX network. It assumes that a WiMAX
    module is plugged in to the DUT and a WiMAX network is available. It repeats
    the connect/disconnect sequence several times. Each time after connecting to
    the WiMAX network, it fetches some data from a URL to confirm the WiMAX
    connectivity.

    """
    version = 1

    def _connect_service(self):
        """Connects the WiMAX service under test.

        Raises:
            error.TestFail if it fails to connect the service before timeout.

        """
        logging.info('ConnectService: %s', self._service.object_path)

        service_properties = self._service.GetProperties()
        parameters = {
            'Type': 'wimax',
            'Name': str(service_properties['Name']),
            'NetworkId': str(service_properties['NetworkId']),
            'EAP.Identity': 'test',
            'EAP.Password': 'test',
        }
        logging.info('parameters : %s', parameters)
        self._flim.manager.ConfigureService(parameters)

        success, status = self._flim.ConnectService(
            service=self._service,
            config_timeout=self._connect_timeout)
        if not success:
            raise error.TestFail('Could not connect: %s.' % status)

        logging.info('Waiting for portal or online state.')
        portal_or_online_states = ['portal', 'online']
        state = self._flim.WaitForServiceState(
            service=self._service,
            expected_states=portal_or_online_states,
            timeout=self._connect_timeout,
            ignore_failure=True)[0]
        if not state in portal_or_online_states:
            raise error.TestFail('Still in state %s' % state)


    def _disconnect_service(self):
        """Disconnects the WiMAX service under test.

        Raises:
            error.TestFail if it fails to disconnect the service before
                timeout.

        """
        logging.info('DisonnectService: %s', self._service.object_path)

        success, status = self._flim.DisconnectService(
            service=self._service,
            wait_timeout=self._disconnect_timeout)
        if not success:
            raise error.TestFail('Could not disconnect: %s.' % status)


    def _test_connectivity(self):
        """Tests network connectivity over WiMAX.

        Test network connectivity over WiMAX as follows:
            - Connecting the WiMAX service
            - Fetching data from a URL
            - Disconnecting the WiMAX service

        Raises:
            error.TestFail if any of the steps above fails.

        """
        if self._sleep_kludge:
            logging.info('Sleeping for %.1f seconds', self._sleep_kludge)
            time.sleep(self._sleep_kludge)

        self._connect_service()

        device = self._flim.GetObjectInterface(
            'Device', self._service.GetProperties()['Device'])
        interface = device.GetProperties()['Interface']
        logging.info('Expected interface for %s: %s',
                     self._service.object_path, interface)
        network.CheckInterfaceForDestination(
            urlparse.urlparse(self._fetch_url_pattern).hostname,
            interface)

        fetch_time = network.FetchUrl(self._fetch_url_pattern,
                                      self._bytes_to_fetch,
                                      self._fetch_timeout)
        self.write_perf_keyval({
            'seconds_WiMAX_fetch_time': fetch_time,
            'bytes_WiMAX_bytes_received': self._bytes_to_fetch,
            'bits_second_WiMAX_speed': 8 * self._bytes_to_fetch / fetch_time
        })

        self._disconnect_service()


    def run_once(self, **kwargs):
        # Number of connectivity test runs.
        self._connect_count = kwargs.get('connect_count', 5)

        # Number of seconds to sleep between connect and disconnect operations.
        self._sleep_kludge = kwargs.get('sleep_kludge', 5)

        # URL pattern to fetch data from during each connectivity test run.
        self._fetch_url_pattern = \
            kwargs.get('fetch_url_pattern', network.FETCH_URL_PATTERN_FOR_TEST)

        # Timeout in seconds for connect and disconnect operations, and
        # fetching data from a URL.
        self._connect_timeout = kwargs.get('connect_timeout', 10)
        self._disconnect_timeout = kwargs.get('disconnect_timeout', 10)
        self._fetch_timeout = kwargs.get('fetch_timeout', 120)

        # Number of bytes to fetch during each connectivity test run.
        self._bytes_to_fetch = kwargs.get('bytes_to_fetch', 64 * 1024)

        with backchannel.Backchannel():
            with cell_tools.OtherDeviceShutdownContext('wimax'):
                # TODO(benchan): Replace FlimFlam with ShillProxy.
                self._flim = flimflam.FlimFlam()
                self._flim.SetDebugTags(SHILL_LOG_SCOPES)
                self._service = self._flim.FindWimaxService()
                if not self._service:
                    raise error.TestError('Could not find a WiMAX service.')

                with shill_context.ServiceAutoConnectContext(
                    self._flim.FindWimaxService, False):
                    self._disconnect_service()
                    for _ in xrange(self._connect_count):
                        self._test_connectivity()