• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import logging
6import socket
7import time
8import urlparse
9
10from autotest_lib.client.bin import test
11from autotest_lib.client.common_lib import error
12from autotest_lib.client.cros import network
13from autotest_lib.client.cros.networking import shill_context
14from autotest_lib.client.cros.networking import shill_proxy
15
16
17# Default timeouts in seconds
18CONNECT_TIMEOUT = 120
19DISCONNECT_TIMEOUT = 60
20
21class cellular_Smoke(test.test):
22    """
23    Tests that 3G modem can connect to the network
24
25    The test attempts to connect using the 3G network. The test then
26    disconnects from the network, and verifies that the modem still
27    responds to modem manager DBUS API calls.  It repeats the
28    connect/disconnect sequence several times.
29
30    """
31    version = 1
32
33
34    def run_once_internal(self):
35        """
36        Executes the test.
37
38        """
39        old_modem_info = self.test_env.modem.GetModemProperties()
40
41        for _ in xrange(self.connect_count):
42            device = self.test_env.shill.find_cellular_device_object()
43            if not device:
44                raise error.TestError('No cellular device found.')
45
46            service = self.test_env.shill.wait_for_cellular_service_object()
47            if not service:
48                raise error.TestError('No cellular service found.')
49
50            logging.info('Connecting to service %s', service.object_path)
51            self.test_env.shill.connect_service_synchronous(
52                    service, CONNECT_TIMEOUT)
53
54            state = self.test_env.shill.get_dbus_property(
55                    service, shill_proxy.ShillProxy.SERVICE_PROPERTY_STATE)
56            logging.info('Service state = %s', state)
57
58            if state == 'portal':
59                url_pattern = ('https://quickaccess.verizonwireless.com/'
60                               'images_b2c/shared/nav/'
61                               'vz_logo_quickaccess.jpg?foo=%d')
62                bytes_to_fetch = 4476
63            else:
64                url_pattern = network.FETCH_URL_PATTERN_FOR_TEST
65                bytes_to_fetch = 64 * 1024
66
67            interface = self.test_env.shill.get_dbus_property(
68                    device, shill_proxy.ShillProxy.DEVICE_PROPERTY_INTERFACE)
69            logging.info('Expected interface for %s: %s',
70                         service.object_path, interface)
71            # TODO(b/114292737): Once IPv6 support is enabled on
72            # cellular, we should not need to limit this check to just
73            # AF_INET.
74            network.CheckInterfaceForDestination(
75                urlparse.urlparse(url_pattern).hostname,
76                interface, socket.AF_INET)
77
78            fetch_time = network.FetchUrl(url_pattern, bytes_to_fetch,
79                                          self.fetch_timeout)
80            self.write_perf_keyval({
81                'seconds_3G_fetch_time': fetch_time,
82                'bytes_3G_bytes_received': bytes_to_fetch,
83                'bits_second_3G_speed': 8 * bytes_to_fetch / fetch_time
84            })
85
86            self.test_env.shill.disconnect_service_synchronous(
87                    service, DISCONNECT_TIMEOUT)
88
89            # Verify that we can still get information about the modem
90            logging.info('Old modem info: %s', ', '.join(old_modem_info))
91            new_modem_info = self.test_env.modem.GetModemProperties()
92            if len(new_modem_info) != len(old_modem_info):
93                logging.info('New modem info: %s', ', '.join(new_modem_info))
94                raise error.TestFail('Test shutdown: '
95                                     'failed to leave modem in working state.')
96
97            if self.sleep_kludge:
98                logging.info('Sleeping for %.1f seconds', self.sleep_kludge)
99                time.sleep(self.sleep_kludge)
100
101
102    def run_once(self, test_env, connect_count=5, sleep_kludge=5,
103                 fetch_timeout=120):
104        with test_env, shill_context.ServiceAutoConnectContext(
105                test_env.shill.wait_for_cellular_service_object, False):
106            self.test_env = test_env
107            self.connect_count = connect_count
108            self.sleep_kludge = sleep_kludge
109            self.fetch_timeout = fetch_timeout
110
111            self.run_once_internal()
112