• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright (c) 2012 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 dbus
6import logging
7import random
8import time
9
10from autotest_lib.client.bin import test
11from autotest_lib.client.common_lib import error
12from autotest_lib.client.cros.networking import cellular_proxy
13from autotest_lib.client.cros.networking import shill_context
14from autotest_lib.client.cros.networking import shill_proxy
15
16
17class cellular_SafetyDance(test.test):
18    """
19    Stress tests all connection manager 3G operations.
20
21    This test runs a long series of 3G operations in pseudorandom order. All of
22    these 3G operations must return a convincing result (EINPROGRESS or no
23    error).
24
25    """
26    version = 1
27
28    def _filterexns(self, fn):
29        v = None
30        try:
31            v = fn()
32        except dbus.exceptions.DBusException, error:
33            if error.get_dbus_name() in self.okerrors:
34                return v, error.get_dbus_message()
35            else:
36                raise error
37        return v, ''
38
39    def _enable(self):
40        logging.info('Enable')
41        self._filterexns(lambda:
42            self.test_env.shill.manager.EnableTechnology('cellular'))
43
44    def _disable(self):
45        logging.info('Disable')
46        self._filterexns(lambda:
47            self.test_env.shill.manager.DisableTechnology('cellular'))
48
49    def _ignoring(self, reason):
50        if ('AlreadyConnected' in reason or
51            'Not connected' in reason or
52            'Bearer already being connected' in reason or
53            'Bearer already being disconnected' in reason or
54            'InProgress' in reason):
55            return True
56        if 'NotSupported' in reason:
57            # We should only ignore this error if we've previously disabled
58            # cellular technology and the service subsequently disappeared
59            # when we tried to connect again.
60            return not self.test_env.shill.find_cellular_service_object()
61        return False
62
63    def _connect(self):
64        logging.info('Connect')
65        try:
66            service = self.test_env.shill.wait_for_cellular_service_object(
67                    timeout_seconds=5)
68        except shill_proxy.ShillProxyError:
69            return
70
71        success, reason = self._filterexns(lambda:
72                self.test_env.shill.connect_service_synchronous(
73                        service=service,
74                        timeout_seconds=
75                        cellular_proxy.CellularProxy.SERVICE_CONNECT_TIMEOUT))
76        if not success and not self._ignoring(reason):
77            raise error.TestFail('Could not connect: %s' % reason)
78
79    def _disconnect(self):
80        logging.info('Disconnect')
81        try:
82            service = self.test_env.shill.wait_for_cellular_service_object(
83                    timeout_seconds=5)
84        except shill_proxy.ShillProxyError:
85            return
86
87        success, reason = self._filterexns(lambda:
88                self.test_env.shill.disconnect_service_synchronous(
89                        service=service,
90                        timeout_seconds=
91                        cellular_proxy.CellularProxy.
92                        SERVICE_DISCONNECT_TIMEOUT))
93        if not success and not self._ignoring(reason):
94            raise error.TestFail('Could not disconnect: %s' % reason)
95
96    def _op(self):
97        n = random.randint(0, len(self.ops) - 1)
98        self.ops[n]()
99        time.sleep(random.randint(5, 20) / 10.0)
100
101    def _run_once_internal(self, ops=30, seed=None):
102        if not seed:
103            seed = int(time.time())
104        self.okerrors = [
105            'org.chromium.flimflam.Error.InProgress',
106            'org.chromium.flimflam.Error.AlreadyConnected',
107            'org.chromium.flimflam.Error.AlreadyEnabled',
108            'org.chromium.flimflam.Error.AlreadyDisabled'
109        ]
110        self.ops = [ self._enable,
111                     self._disable,
112                     self._connect,
113                     self._disconnect ]
114        self.device = self.test_env.shill.find_cellular_device_object()
115        if not self.device:
116            raise error.TestFail('Could not find cellular device.')
117
118        # Start in a disabled state.
119        self._disable()
120        logging.info('Seed: %d', seed)
121        random.seed(seed)
122        for _ in xrange(ops):
123            self._op()
124
125    def run_once(self, test_env, ops=30, seed=None):
126        self.test_env = test_env
127        with test_env, shill_context.ServiceAutoConnectContext(
128                test_env.shill.wait_for_cellular_service_object, False):
129            self._run_once_internal(ops, seed)
130
131            # Enable device to restore autoconnect settings.
132            self._enable()
133            test_env.shill.wait_for_cellular_service_object()
134