• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright (c) 2013 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 contextlib
6import dbus
7import logging
8import random
9import time
10
11from autotest_lib.client.bin import test, utils
12from autotest_lib.client.common_lib import error
13from autotest_lib.client.common_lib.cros import chrome
14from autotest_lib.client.cros.cellular import cellular
15from autotest_lib.client.cros.networking import cellular_proxy
16from autotest_lib.client.cros.networking import shill_context
17from autotest_lib.client.cros.networking import shill_proxy
18
19# Number of seconds we wait for the cellular service to perform an action.
20DEVICE_TIMEOUT=45
21SERVICE_TIMEOUT=75
22
23# Number of times and seconds between modem state checks to ensure that the
24# modem is not in a temporary transition state.
25NUM_MODEM_STATE_CHECKS=2
26MODEM_STATE_CHECK_PERIOD_SECONDS=5
27
28# Number of seconds to sleep after a connect request in slow-connect mode.
29SLOW_CONNECT_WAIT_SECONDS=20
30
31
32class TechnologyCommands():
33    """Control the modem mostly using shill Technology interfaces."""
34    def __init__(self, shill, command_delegate):
35        self.shill = shill
36        self.command_delegate = command_delegate
37
38    def Enable(self):
39        self.shill.manager.EnableTechnology(
40                shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)
41
42    def Disable(self):
43        self.shill.manager.DisableTechnology(
44                shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)
45
46    def Connect(self, **kwargs):
47        self.command_delegate.Connect(**kwargs)
48
49    def Disconnect(self):
50        return self.command_delegate.Disconnect()
51
52    def __str__(self):
53        return 'Technology Commands'
54
55
56class ModemCommands():
57    """Control the modem using modem manager DBUS interfaces."""
58    def __init__(self, modem, slow_connect):
59        self.modem = modem
60        self.slow_connect = slow_connect
61
62    def Enable(self):
63        self.modem.Enable(True)
64
65    def Disable(self):
66        self.modem.Enable(False)
67
68    def Connect(self, simple_connect_props):
69        logging.debug('Connecting with properties: %r' % simple_connect_props)
70        self.modem.Connect(simple_connect_props)
71        if self.slow_connect:
72            time.sleep(SLOW_CONNECT_WAIT_SECONDS)
73
74    def Disconnect(self):
75        """
76        Disconnect Modem.
77
78        Returns:
79            True - to indicate that shill may autoconnect again.
80        """
81        try:
82            self.modem.Disconnect()
83        except dbus.DBusException as e:
84            if (e.get_dbus_name() !=
85                    'org.chromium.ModemManager.Error.OperationInitiated'):
86                raise e
87        return True
88
89    def __str__(self):
90        return 'Modem Commands'
91
92
93class DeviceCommands():
94    """Control the modem using shill device interfaces."""
95    def __init__(self, shill, device, slow_connect):
96        self.shill = shill
97        self.device = device
98        self.slow_connect = slow_connect
99        self.service = None
100
101    def GetService(self):
102        service = self.shill.find_cellular_service_object()
103        if not service:
104            raise error.TestFail(
105                'Service failed to appear when using device commands.')
106        return service
107
108    def Enable(self):
109        self.device.Enable(timeout=DEVICE_TIMEOUT)
110
111    def Disable(self):
112        self.service = None
113        self.device.Disable(timeout=DEVICE_TIMEOUT)
114
115    def Connect(self, **kwargs):
116        self.GetService().Connect()
117        if self.slow_connect:
118            time.sleep(SLOW_CONNECT_WAIT_SECONDS)
119
120    def Disconnect(self):
121        """
122        Disconnect Modem.
123
124        Returns:
125            False - to indicate that shill may not autoconnect again.
126        """
127        self.GetService().Disconnect()
128        return False
129
130    def __str__(self):
131        return 'Device Commands'
132
133
134class MixedRandomCommands():
135    """Control the modem using a mixture of commands on device, modems, etc."""
136    def __init__(self, commands_list):
137        self.commands_list = commands_list
138
139    def PickRandomCommands(self):
140        return self.commands_list[random.randrange(len(self.commands_list))]
141
142    def Enable(self):
143        cmds = self.PickRandomCommands()
144        logging.info('Enable with %s' % cmds)
145        cmds.Enable()
146
147    def Disable(self):
148        cmds = self.PickRandomCommands()
149        logging.info('Disable with %s' % cmds)
150        cmds.Disable()
151
152    def Connect(self, **kwargs):
153        cmds = self.PickRandomCommands()
154        logging.info('Connect with %s' % cmds)
155        cmds.Connect(**kwargs)
156
157    def Disconnect(self):
158        cmds = self.PickRandomCommands()
159        logging.info('Disconnect with %s' % cmds)
160        return cmds.Disconnect()
161
162    def __str__(self):
163        return 'Mixed Commands'
164
165
166class cellular_ModemControl(test.test):
167    version = 1
168
169    def CompareModemPowerState(self, modem, expected_state):
170        """Compare modem manager power state of a modem to an expected state."""
171        return modem.IsEnabled() == expected_state
172
173    def CompareDevicePowerState(self, device, expected_state):
174        """Compare the shill device power state to an expected state."""
175        state = self.test_env.shill.get_dbus_property(
176                device, shill_proxy.ShillProxy.DEVICE_PROPERTY_POWERED)
177        logging.info('Device Enabled = %s' % state)
178        return state == expected_state
179
180    def CompareServiceState(self, service, expected_states):
181        """Compare the shill service state to a set of expected states."""
182        if not service:
183            logging.info('Service not found.')
184            return False
185
186        state = self.test_env.shill.get_dbus_property(
187                service, shill_proxy.ShillProxy.SERVICE_PROPERTY_STATE)
188        logging.info('Service State = %s' % state)
189        return state in expected_states
190
191    def EnsureNotConnectingOrDisconnecting(self):
192        """
193        Ensure modem is not connecting or disconnecting.
194
195        Raises:
196            error.TestFail if it timed out waiting for the modem to finish
197            connecting or disconnecting.
198        """
199        # Shill retries a failed connect attempt with a different APN so
200        # check a few times to ensure the modem is not in between connect
201        # attempts.
202        for _ in range(NUM_MODEM_STATE_CHECKS):
203            utils.poll_for_condition(
204                lambda: not self.test_env.modem.IsConnectingOrDisconnecting(),
205                error.TestFail('Timed out waiting for modem to finish ' +
206                               'connecting or disconnecting.'),
207                timeout=SERVICE_TIMEOUT)
208            time.sleep(MODEM_STATE_CHECK_PERIOD_SECONDS)
209
210    def EnsureDisabled(self):
211        """
212        Ensure modem disabled, device powered off, and no service.
213
214        Raises:
215            error.TestFail if the states are not consistent.
216        """
217        utils.poll_for_condition(
218            lambda: self.CompareModemPowerState(self.test_env.modem, False),
219            error.TestFail('Modem failed to enter state Disabled.'))
220        utils.poll_for_condition(
221            lambda: self.CompareDevicePowerState(self.device, False),
222            error.TestFail('Device failed to enter state Powered=False.'))
223        utils.poll_for_condition(
224            lambda: not self.test_env.shill.find_cellular_service_object(),
225            error.TestFail('Service should not be available.'),
226            timeout=SERVICE_TIMEOUT)
227
228    def EnsureEnabled(self, check_idle):
229        """
230        Ensure modem enabled, device powered and service exists.
231
232        Args:
233            check_idle: if True, then ensure that the service is idle
234                        (i.e. not connected) otherwise ignore the
235                        service state
236
237        Raises:
238            error.TestFail if the states are not consistent.
239        """
240        utils.poll_for_condition(
241            lambda: self.CompareModemPowerState(self.test_env.modem, True),
242            error.TestFail('Modem failed to enter state Enabled'))
243        utils.poll_for_condition(
244            lambda: self.CompareDevicePowerState(self.device, True),
245            error.TestFail('Device failed to enter state Powered=True.'),
246            timeout=30)
247
248        service = self.test_env.shill.wait_for_cellular_service_object()
249        if check_idle:
250            utils.poll_for_condition(
251                lambda: self.CompareServiceState(service, ['idle']),
252                error.TestFail('Service failed to enter idle state.'),
253                timeout=SERVICE_TIMEOUT)
254
255    def EnsureConnected(self):
256        """
257        Ensure modem connected, device powered on, service connected.
258
259        Raises:
260            error.TestFail if the states are not consistent.
261        """
262        self.EnsureEnabled(check_idle=False)
263        utils.poll_for_condition(
264            lambda: self.CompareServiceState(
265                    self.test_env.shill.find_cellular_service_object(),
266                    ['ready', 'portal', 'online']),
267            error.TestFail('Service failed to connect.'),
268            timeout=SERVICE_TIMEOUT)
269
270
271    def TestCommands(self, commands):
272        """
273        Manipulate the modem using modem, device or technology commands.
274
275        Changes the state of the modem in various ways including
276        disable while connected and then verifies the state of the
277        modem manager and shill.
278
279        Raises:
280            error.TestFail if the states are not consistent.
281
282        """
283        logging.info('Testing using %s' % commands)
284
285        logging.info('Enabling')
286        commands.Enable()
287        self.EnsureEnabled(check_idle=not self.autoconnect)
288
289        technology_family = self.test_env.modem.GetCurrentTechnologyFamily()
290        if technology_family == cellular.TechnologyFamily.CDMA:
291            simple_connect_props = {'number': r'#777'}
292        else:
293            simple_connect_props = {'number': r'#777', 'apn': self.FindAPN()}
294
295        # Icera modems behave weirdly if we cancel the operation while the
296        # modem is connecting. Work around the issue by waiting until the
297        # connect operation completes.
298        # TODO(benchan): Remove this workaround once the issue is addressed
299        # on the modem side.
300        self.EnsureNotConnectingOrDisconnecting()
301
302        logging.info('Disabling')
303        commands.Disable()
304        self.EnsureDisabled()
305
306        logging.info('Enabling again')
307        commands.Enable()
308        self.EnsureEnabled(check_idle=not self.autoconnect)
309
310        if not self.autoconnect:
311            logging.info('Connecting')
312            commands.Connect(simple_connect_props=simple_connect_props)
313        else:
314            logging.info('Expecting AutoConnect to connect')
315        self.EnsureConnected()
316
317        logging.info('Disconnecting')
318        will_autoreconnect = commands.Disconnect()
319
320        if not (self.autoconnect and will_autoreconnect):
321            # Icera modems behave weirdly if we cancel the operation while the
322            # modem is disconnecting. Work around the issue by waiting until
323            # the disconnect operation completes.
324            # TODO(benchan): Remove this workaround once the issue is addressed
325            # on the modem side.
326            self.EnsureNotConnectingOrDisconnecting()
327
328            self.EnsureEnabled(check_idle=True)
329            logging.info('Connecting manually, since AutoConnect was on')
330            commands.Connect(simple_connect_props=simple_connect_props)
331        self.EnsureConnected()
332
333        logging.info('Disabling')
334        commands.Disable()
335        self.EnsureDisabled()
336
337    def FindAPN(self):
338        default = 'None'
339        service = self.test_env.shill.find_cellular_service_object()
340        last_good_apn = self.test_env.shill.get_dbus_property(
341                service,
342                cellular_proxy.CellularProxy.SERVICE_PROPERTY_LAST_GOOD_APN)
343        if not last_good_apn:
344            return default
345        return last_good_apn.get(
346                cellular_proxy.CellularProxy.APN_INFO_PROPERTY_APN, default)
347
348    def run_once(self, test_env, autoconnect, mixed_iterations=2,
349                 slow_connect=False):
350        self.test_env = test_env
351        self.autoconnect = autoconnect
352
353        with test_env:
354            self.device = self.test_env.shill.find_cellular_device_object()
355
356            modem_commands = ModemCommands(self.test_env.modem,
357                                           slow_connect)
358            technology_commands = TechnologyCommands(self.test_env.shill,
359                                                     modem_commands)
360            device_commands = DeviceCommands(self.test_env.shill,
361                                             self.device,
362                                             slow_connect)
363
364            # shill disables autoconnect on any cellular service before a user
365            # logs in (CL:851267). To test the autoconnect scenario, we need a
366            # user session to run the test.
367            chrome_context = chrome.Chrome()
368
369            # Set up the autoconnect context after starting a user session so
370            # that we ensure the autoconnect property is set on the cellular
371            # service that may be in the user profile.
372            autoconnect_context = shill_context.ServiceAutoConnectContext(
373                    self.test_env.shill.wait_for_cellular_service_object,
374                    self.autoconnect)
375
376            with contextlib.nested(chrome_context, autoconnect_context):
377                # Start with cellular disabled.
378                self.test_env.shill.manager.DisableTechnology(
379                        shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)
380                self.EnsureDisabled()
381
382                # Run the device commands test first to make sure we have
383                # a valid APN needed to connect using the modem commands.
384                self.TestCommands(device_commands)
385                self.TestCommands(technology_commands)
386                self.TestCommands(modem_commands)
387
388                # Run several times using commands mixed from each type
389                mixed = MixedRandomCommands([modem_commands,
390                                             technology_commands,
391                                             device_commands])
392                for _ in range(mixed_iterations):
393                    self.TestCommands(mixed)
394
395                # Ensure cellular is re-enabled in order to restore AutoConnect
396                # settings when ServiceAutoConnectContext exits.
397                # TODO(benchan): Refactor this logic into
398                # ServiceAutoConnectContext and update other users of
399                # ServiceAutoConnectContext.
400                self.test_env.shill.manager.EnableTechnology(
401                        shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR)
402                self.EnsureEnabled(check_idle=False)
403