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