• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright (c) 2014 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
6
7import common
8from autotest_lib.client.cros.cellular.pseudomodem import modem_3gpp
9from autotest_lib.client.cros.cellular.pseudomodem import pm_constants
10from autotest_lib.client.cros.cellular.pseudomodem import state_machine
11from autotest_lib.client.cros.cellular.pseudomodem import state_machine_factory
12
13class InteractiveStateMachineFactory(state_machine_factory.StateMachineFactory):
14    """ Run relevant state machines in interactive mode. """
15    def __init__(self):
16        super(InteractiveStateMachineFactory, self).__init__()
17        self.SetInteractive(pm_constants.STATE_MACHINE_ENABLE)
18        self.SetInteractive(pm_constants.STATE_MACHINE_REGISTER)
19
20
21class ScanMachine(state_machine.StateMachine):
22    """
23    Handle shill initiated 3GPP scan request.
24
25    A simple machine that allows the test to hook into the Scan asynchronous
26    call.
27
28    """
29    # State machine states.
30    SCAN_STATE = 'Scan'
31    DONE_STATE = 'Done'
32
33    def __init__(self, modem):
34        super(ScanMachine, self).__init__(modem)
35        self._state = ScanMachine.SCAN_STATE
36
37
38    def _HandleScanState(self):
39        """ The only real state in this machine. """
40        self._modem.DoScan()
41        self._state = ScanMachine.DONE_STATE
42        return True
43
44
45    def _GetCurrentState(self):
46        return self._state
47
48
49    def _GetModemStateFunctionMap(self):
50        return {
51                ScanMachine.SCAN_STATE: ScanMachine._HandleScanState,
52                # ScanMachine.DONE_STATE is the final state. So, no handler.
53        }
54
55
56    def _ShouldStartStateMachine(self):
57        return True
58
59
60class ScanStateMachineFactory(state_machine_factory.StateMachineFactory):
61    """ Extend StateMachineFactory to create an interactive ScanMachine. """
62    def ScanMachine(self, *args, **kwargs):
63        """ Create a ScanMachine when needed in the modem. """
64        machine = ScanMachine(*args, **kwargs)
65        machine.EnterInteractiveMode(self._bus)
66        return machine
67
68
69class AsyncScanModem(modem_3gpp.Modem3gpp):
70    """ 3GPP modem that uses ScanMachine for the Scan call. """
71    def __init__(self):
72        super(AsyncScanModem, self).__init__(
73                state_machine_factory=ScanStateMachineFactory())
74
75
76    def Scan(self, return_cb, raise_cb):
77        """ Overriden from Modem3gpp. """
78        # Stash away the scan_ok callback for when the Scan finishes.
79        logging.debug('Network scan initiated.')
80        self._scan_ok_callback = return_cb
81        self._scan_failed_callback = raise_cb
82        self._scan_machine = self._state_machine_factory.ScanMachine(self)
83        self._scan_machine.Start()
84
85
86    def DoScan(self):
87        """ Defer to Modem3gpp to take the original |SyncScan| action. """
88        # We're done scanning, drop |_scan_machine| reference.
89        self._scan_machine = None
90        try:
91            scan_result = super(AsyncScanModem, self).SyncScan()
92        except dbus.exceptions.DBusException as e:
93            logging.warning('Network scan failed')
94            self._scan_failed_callback(e)
95            return
96
97        logging.debug('Network scan completed.')
98        self._scan_ok_callback(scan_result)
99