client/site_tests/network_Portal/network_Portal.py

# Copyright (c) 2010 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, time

from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import backchannel
from autotest_lib.client.cros.cellular import cell_tools
from autotest_lib.client.cros import network

from autotest_lib.client.cros import flimflam_test_path
import flimflam

class network_Portal(test.test):
    version = 1

    def GetConnectedService(self, service_name):
        service = self.flim.FindElementByNameSubstring('Service',
                                                       service_name)
        if service:
            properties = service.GetProperties(utf8_strings=True)
            state = properties['State']

            if state in ('online', 'portal', 'ready'):
                self.service = service
                return True

        return False

    def TestConnect(self, service_name, expected_state, timeout_seconds=30):
        """Connect to a service and verifies the portal state

        Args:
          service_name: substring to match against services
          expected_state: expected state of service after connecting

        Returns:
          True if the service is in the expected state
          False otherwise

        Raises:
          error.TestFail on non-recoverable failure
        """
        logging.info('TestConnect(%s, %s)' % (service_name, expected_state))

        self.service.Disconnect()
        state = self.flim.WaitForServiceState(
            service=self.service,
            expected_states=['idle', 'failure'],
            timeout=5)[0]

        self.service.Connect()
        state = self.flim.WaitForServiceState(
            service=self.service,
            expected_states=['portal', 'online', 'failure'],
            timeout=timeout_seconds)[0]

        if state != expected_state:
            logging.error('Service state should be %s but is %s' %
                          (expected_state, state))
            return False

        return True


    def run_once(self, force_failure, test_iterations=10, service_name='wifi'):
        """Run a test of the portal code.

        Args:
          force_failure: keyed values for different flavors of tests
              'portal' - blackhole the hosts used for http ensure that
                         the connection manager detects a portal.
              'dns'    - blackhole the DNS servers and ensure that
                         the connection manager detects a portal.
               False   - Do not force any failes and ensure that the
                         connection manager detects the service is online.
              'partial-dns'
                       - blackhole the first DNS server and ensure that the
                         connection manager detects the service is online.
        """
        errors = 0
        if force_failure == 'portal':
            # depends on getting a consistent IP address from DNS
            # depends on portal detection using www.google.com or
            # clients3.google.com
            hosts = [('clients3.google.com', 'OUTPUT'),
                     ('www.google.com', 'OUTPUT')]
            expected_state = 'portal'
        else:
            hosts = []
            expected_state = 'online'

        with backchannel.Backchannel():
            # Immediately after the backchannel is setup there may be no
            # services.  Try for up to 20 seconds to find one.
            # Getting to the ready state on an encrypted network can
            # be slow.
            self.flim = flimflam.FlimFlam()
            self.flim.SetDebugTags('portal+service')
            utils.poll_for_condition(
                lambda: self.GetConnectedService(service_name),
                error.TestFail(
                    'No service named "%s" available' % service_name),
                timeout=20)

            nameservers = []
            if force_failure == 'dns':
                nameservers = network.NameServersForService(self.flim,
                                                            self.service)
                expected_state = 'portal'
            elif force_failure == 'partial-dns':
                nameservers = network.NameServersForService(self.flim,
                                                            self.service)
                nameservers = nameservers[0:1]

            hosts += [(host, 'INPUT') for host in nameservers]

            with cell_tools.BlackholeContext(hosts):
                for _ in range(test_iterations):
                    if not self.TestConnect(service_name, expected_state):
                        errors += 1

                if errors:
                    raise error.TestFail('%d failures to enter state %s ' % (
                        errors, expected_state))

            #
            # Test the portal retry timer which should transition the
            # service to online (at most) 1 minute after the portal
            # restriction is removed.
            #
            if expected_state == 'portal':
                with cell_tools.BlackholeContext(hosts):
                    if not self.TestConnect(service_name, 'portal'):
                        raise error.TestFail('Failed to enter state portal')

                state = self.flim.WaitForServiceState(
                    service=self.service,
                    expected_states=['online'],
                    timeout=60)[0]

                if state != 'online':
                    raise error.TestFail('Failed to enter state online')

                # Check that we recheck and find a portal if the
                # PortalList changes
                with cell_tools.BlackholeContext(hosts):
                    self.flim.SetCheckPortalList('wifi,ethernet,cellular')
                    state = self.flim.WaitForServiceState(
                        service=self.service,
                        expected_states=['portal'],
                        timeout=60)[0]

                    if state != 'portal':
                        raise error.TestFail('Expected to enter state portal')