• 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 logging, os, time
6
7from autotest_lib.client.bin import test, utils
8from autotest_lib.client.common_lib import error
9from autotest_lib.client.cros.cellular import mm
10from autotest_lib.client.cros.networking import shill_proxy
11
12
13class ResetAuthorizedContext(object):
14    def __init__(self, test):
15        self.test = test
16
17    def __enter__(self):
18        pass
19
20    def __exit__(self, exception, value, traceback):
21        if exception:
22            self.test.SetAuthorized(1)
23        return False
24
25
26class network_3GRebootStress(test.test):
27    version = 1
28
29    def IsCromo(self, modem_manager):
30        path = modem_manager.path
31        return path.startswith('/org/chromium')
32
33    def CountModems(self):
34        count = len(mm.EnumerateDevices(''))
35        logging.debug('Modem count is %d' % count)
36        return count
37
38    def EnsureModemAbsent(self):
39        utils.poll_for_condition(
40            lambda: self.CountModems() == 0,
41            error.TestFail('Modem failed to disappear'),
42            timeout=shill_proxy.ShillProxy.DEVICE_ENABLE_DISABLE_TIMEOUT)
43
44    def EnsureModemPresent(self):
45        utils.poll_for_condition(
46            lambda: self.CountModems() == 1,
47            error.TestFail('Modem failed to reappear'),
48            timeout=shill_proxy.ShillProxy.DEVICE_ENABLE_DISABLE_TIMEOUT)
49
50    def FindUsbDevicePath(self, modem_manager, modem_path):
51        logging.info('Modem path: %s' % modem_path)
52
53        modem_obj = modem_manager.GetModem(modem_path)
54        props = modem_obj.GetModemProperties()
55        net_device = props['Device']
56        logging.info('Network device: %s' % net_device)
57        if self.IsCromo(modem_manager):
58            usb_interface_path, _ = os.path.realpath(
59                    '/sys/class/net/%s/device' % net_device).rsplit('/', 1)
60        else:
61            usb_interface_path = os.path.realpath(net_device)
62        self.usb_device_path = usb_interface_path
63        logging.info('USB device path: %s' % self.usb_device_path)
64
65    def SetAuthorized(self, authorized):
66        logging.debug('Setting authorized to %d' % authorized)
67        authorized_path = '%s/authorized' % self.usb_device_path
68        authorized_file = open(authorized_path, 'w')
69        authorized_file.write('%d' % authorized)
70        authorized_file.close()
71
72    def ShouldContinue(self):
73        should_continue = True
74        message = 'Starting loop %d' % (self.loops_done + 1)
75
76        if self.max_loops != None:
77            if self.loops_done >= self.max_loops:
78                return False
79            loops_left = self.max_loops - self.loops_done
80            message += '; %d loops left' % loops_left
81
82        if self.max_seconds != None:
83            seconds_done = time.time() - self.start_time
84            if seconds_done >= self.max_seconds:
85                return False
86            seconds_left = self.max_seconds - seconds_done
87            message += '; %d seconds left' % seconds_left
88
89        message += '.'
90        logging.info(message)
91
92        return True
93
94
95    # It takes a Gobi about 1.5 seconds to cycle away and back, on average.
96    # Assume 2, to have a margin of error.  (If we run out of time, it's
97    # fine, but we'd like to run the same number of loops every time.)
98    def run_once(self, test_env, max_loops=300, max_seconds=600):
99        with test_env:
100            self.FindUsbDevicePath(test_env.modem_manager, test_env.modem_path)
101
102            self.max_loops = max_loops
103            self.max_seconds = max_seconds
104
105            self.loops_done = 0
106            self.start_time = time.time()
107
108            # Use a context to ensure that we don't leave the modem
109            # unauthorized if we die with an exception.
110            with ResetAuthorizedContext(self):
111                while self.ShouldContinue():
112                    self.SetAuthorized(0)
113                    self.EnsureModemAbsent()
114                    self.SetAuthorized(1)
115                    self.EnsureModemPresent()
116                    self.loops_done += 1
117
118            if self.loops_done < self.max_loops:
119                logging.warning('Only got %d loops done in %d seconds. :(',
120                                self.loops_done, max_seconds)
121