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.cellular import cell_tools, mm
13
14
15# Default timeouts in seconds
16CONNECT_TIMEOUT = 120
17DISCONNECT_TIMEOUT = 60
18
19SHILL_LOG_SCOPES = 'cellular+dbus+device+dhcp+manager+modem+portal+service'
20
21class network_3GSmokeTest(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    # TODO(benchan): Migrate to use ShillProxy when ShillProxy provides a
34    # similar method.
35    def DisconnectFrom3GNetwork(self, disconnect_timeout):
36        """Attempts to disconnect from a 3G network.
37
38        @param disconnect_timeout: Timeout in seconds for disconnecting from
39                                   the network.
40
41        @raises error.TestFail if it fails to disconnect from the network before
42                timeout.
43        @raises error.TestError If no cellular service is available.
44
45        """
46        logging.info('DisconnectFrom3GNetwork')
47
48        service = self.test_env.flim.FindCellularService()
49        if not service:
50            raise error.TestError('Could not find cellular service.')
51
52        success, status = self.test_env.flim.DisconnectService(
53            service=service,
54            wait_timeout=disconnect_timeout)
55        if not success:
56            raise error.TestFail('Could not disconnect: %s.' % status)
57
58
59    def GetModemInfo(self):
60        """Find all modems attached and return an dictionary of information.
61
62        This returns a bunch of information for each modem attached to
63        the system.  In practice collecting all this information
64        sometimes fails if a modem is left in an odd state, so we
65        collect as many things as we can to ensure that the modem is
66        responding correctly.
67
68        @return A dictionary of information for each modem path.
69        """
70        results = {}
71
72        devices = mm.EnumerateDevices()
73        print 'Devices: %s' % ', '.join([p for _, p in devices])
74        for manager, path in devices:
75            modem = manager.GetModem(path)
76            results[path] = modem.GetModemProperties()
77        return results
78
79
80    def run_once_internal(self):
81        """
82        Executes the test.
83
84        """
85        # Get information about all the modems
86        old_modem_info = self.GetModemInfo()
87
88        for _ in xrange(self.connect_count):
89            service, state = cell_tools.ConnectToCellular(self.test_env.flim,
90                                                          CONNECT_TIMEOUT)
91
92            if state == 'portal':
93                url_pattern = ('https://quickaccess.verizonwireless.com/'
94                               'images_b2c/shared/nav/'
95                               'vz_logo_quickaccess.jpg?foo=%d')
96                bytes_to_fetch = 4476
97            else:
98                url_pattern = network.FETCH_URL_PATTERN_FOR_TEST
99                bytes_to_fetch = 64 * 1024
100
101            device = self.test_env.flim.GetObjectInterface(
102                'Device', service.GetProperties()['Device'])
103            interface = device.GetProperties()['Interface']
104            logging.info('Expected interface for %s: %s',
105                         service.object_path, interface)
106            network.CheckInterfaceForDestination(
107                urlparse.urlparse(url_pattern).hostname,
108                interface)
109
110            fetch_time = network.FetchUrl(url_pattern, bytes_to_fetch,
111                                          self.fetch_timeout)
112            self.write_perf_keyval({
113                'seconds_3G_fetch_time': fetch_time,
114                'bytes_3G_bytes_received': bytes_to_fetch,
115                'bits_second_3G_speed': 8 * bytes_to_fetch / fetch_time
116            })
117
118            self.DisconnectFrom3GNetwork(disconnect_timeout=DISCONNECT_TIMEOUT)
119
120            # Verify that we can still get information for all the modems
121            logging.info('Old modem info: %s', ', '.join(old_modem_info))
122            new_modem_info = self.GetModemInfo()
123            if len(new_modem_info) != len(old_modem_info):
124                logging.info('New modem info: %s', ', '.join(new_modem_info))
125                raise error.TestFail('Test shutdown: '
126                                     'failed to leave modem in working state.')
127
128            if self.sleep_kludge:
129                logging.info('Sleeping for %.1f seconds', self.sleep_kludge)
130                time.sleep(self.sleep_kludge)
131
132
133    def run_once(self, test_env, connect_count=5, sleep_kludge=5,
134                 fetch_timeout=120):
135        with test_env:
136            self.test_env = test_env
137            self.connect_count = connect_count
138            self.sleep_kludge = sleep_kludge
139            self.fetch_timeout = fetch_timeout
140
141            self.run_once_internal()
142