| # Copyright (c) 2012 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import dbus |
| import dbus.service |
| import gobject |
| import logging |
| |
| import pm_errors |
| import pm_constants |
| import utils |
| |
| from autotest_lib.client.cros.cellular import mm1_constants |
| |
| class StateMachine(dbus.service.Object): |
| """ |
| StateMachine is the abstract base class for the complex state machines |
| that are involved in the pseudo modem manager. |
| |
| Every state transition is managed by a function that has been mapped to a |
| specific modem state. For example, the method that handles the case where |
| the modem is in the ENABLED state would look like: |
| |
| def _HandleEnabledState(self): |
| # Do stuff. |
| |
| The correct method will be dynamically located and executed by the step |
| function according to the dictionary returned by the subclass' |
| implementation of StateMachine._GetModemStateFunctionMap. |
| |
| Using the StateMachine in |interactive| mode: |
| In interactive mode, the state machine object exposes a dbus object under |
| the object path |pm_constants.TESTING_PATH|/|self._GetIsmObjectName()|, |
| where |self._GetIsmObjectName()| returns the dbus object name to be used. |
| |
| In this mode, the state machine waits for a dbus method call |
| |pm_constants.I_TESTING_ISM|.|Advance| when a state transition is possible |
| before actually executing the transition. |
| |
| """ |
| def __init__(self, modem): |
| super(StateMachine, self).__init__(None, None) |
| self._modem = modem |
| self._started = False |
| self._done = False |
| self._interactive = False |
| self._trans_func_map = self._GetModemStateFunctionMap() |
| |
| |
| def __exit__(self): |
| self.remove_from_connection() |
| |
| |
| @property |
| def cancelled(self): |
| """ |
| @returns: True, if the state machine has been cancelled or has |
| transitioned to a terminal state. False, otherwise. |
| |
| """ |
| return self._done |
| |
| |
| def Cancel(self): |
| """ |
| Tells the state machine to stop transitioning to further states. |
| |
| """ |
| self._done = True |
| |
| |
| def EnterInteractiveMode(self, bus): |
| """ |
| Run this machine in interactive mode. |
| |
| This function must be called before |Start|. In this mode, the machine |
| waits for an |Advance| call before each step. |
| |
| @param bus: The bus on which the testing interface must be exported. |
| |
| """ |
| if not bus: |
| self.warning('Cannot enter interactive mode without a |bus|.') |
| return |
| |
| self._interactive = True |
| self._ism_object_path = '/'.join([pm_constants.TESTING_PATH, |
| self._GetIsmObjectName()]) |
| self.add_to_connection(bus, self._ism_object_path) |
| self._interactive = True |
| self._waiting_for_advance = False |
| logging.info('Running state machine in interactive mode') |
| logging.info('Exported test object at %s', self._ism_object_path) |
| |
| |
| def Start(self): |
| """ Start the state machine. """ |
| self.Step() |
| |
| |
| @utils.log_dbus_method() |
| @dbus.service.method(pm_constants.I_TESTING_ISM, out_signature='b') |
| def Advance(self): |
| """ |
| Advance a step on a state machine running in interactive mode. |
| |
| @returns: True if the state machine was advanced. False otherwise. |
| @raises: TestError if called on a non-interactive state machine. |
| |
| """ |
| if not self._interactive: |
| raise pm_errors.TestError( |
| 'Can not advance a non-interactive state machine') |
| |
| if not self._waiting_for_advance: |
| logging.warning('%s received an unexpected advance request', |
| self._GetIsmObjectName()) |
| return False |
| logging.info('%s state machine advancing', self._GetIsmObjectName()) |
| self._waiting_for_advance = False |
| if not self._next_transition(self): |
| self._done = True |
| self._ScheduleNextStep() |
| return True |
| |
| |
| @dbus.service.signal(pm_constants.I_TESTING_ISM) |
| def Waiting(self): |
| """ |
| Signal sent out by an interactive machine when it is waiting for remote |
| dbus call on the |Advance| function. |
| |
| """ |
| logging.info('%s state machine waiting', self._GetIsmObjectName()) |
| |
| |
| @utils.log_dbus_method() |
| @dbus.service.method(pm_constants.I_TESTING_ISM, out_signature='b') |
| def IsWaiting(self): |
| """ |
| Determine whether the state machine is waiting for user action. |
| |
| @returns: True if machine is waiting for |Advance| call. |
| |
| """ |
| return self._waiting_for_advance |
| |
| |
| def Step(self): |
| """ |
| Executes the next corresponding state transition based on the modem |
| state. |
| |
| """ |
| logging.info('StateMachine: Step') |
| if self._done: |
| logging.info('StateMachine: Terminating.') |
| return |
| |
| if not self._started: |
| if not self._ShouldStartStateMachine(): |
| logging.info('StateMachine cannot start.') |
| return |
| self._started = True |
| |
| state = self._GetCurrentState() |
| func = self._trans_func_map.get(state, self._GetDefaultHandler()) |
| if not self._interactive: |
| if func and func(self): |
| self._ScheduleNextStep() |
| else: |
| self._done = True |
| return |
| |
| assert not self._waiting_for_advance |
| if func: |
| self._next_transition = func |
| self._waiting_for_advance = True |
| self.Waiting() # Wait for user to |Advance| the machine. |
| else: |
| self._done = True |
| |
| |
| def _ScheduleNextStep(self): |
| """ |
| Schedules the next state transition to execute on the idle loop. |
| subclasses can override this method to implement custom logic, such as |
| delays. |
| |
| """ |
| gobject.idle_add(StateMachine.Step, self) |
| |
| |
| def _GetIsmObjectName(self): |
| """ |
| The name of the dbus object exposed by this object with |I_TESTING_ISM| |
| interface. |
| |
| By default, this is the name of the most concrete class of the object. |
| |
| """ |
| return self.__class__.__name__ |
| |
| |
| def _GetDefaultHandler(self): |
| """ |
| Returns the function to handle a modem state, for which the value |
| returned by StateMachine._GetModemStateFunctionMap is None. The |
| returned function's signature must match: |
| |
| StateMachine -> Boolean |
| |
| This function by default returns None. If no function exists to handle |
| a modem state, the default behavior is to terminate the state machine. |
| |
| """ |
| return None |
| |
| |
| def _GetModemStateFunctionMap(self): |
| """ |
| Returns a mapping from modem states to corresponding transition |
| functions to execute. The returned function's signature must match: |
| |
| StateMachine -> Boolean |
| |
| The first argument to the function is a state machine, which will |
| typically be passed a value of |self|. The return value, if True, |
| indicates that the state machine should keep executing further state |
| transitions. A return value of False indicates that the state machine |
| will transition to a terminal state. |
| |
| This method must be implemented by a subclass. Subclasses can further |
| override this method to provide custom functionality. |
| |
| """ |
| raise NotImplementedError() |
| |
| |
| def _ShouldStartStateMachine(self): |
| """ |
| This method will be called when the state machine is in a starting |
| state. This method should return True, if the state machine can |
| successfully begin its state transitions, False if it should not |
| proceed. This method can also raise an exception in the failure case. |
| |
| In the success case, this method should also execute any necessary |
| initialization steps. |
| |
| This method must be implemented by a subclass. Subclasses can |
| further override this method to provide custom functionality. |
| |
| """ |
| raise NotImplementedError() |
| |
| |
| def _GetCurrentState(self): |
| """ |
| Get the current state of the state machine. |
| |
| This method is called to get the current state of the machine when |
| deciding what the next transition should be. |
| By default, the state machines are tied to the modem state, and this |
| function simply returns the modem state. |
| |
| Subclasses can override this function to use custom states in the state |
| machine. |
| |
| @returns: The modem state. |
| |
| """ |
| return self._modem.Get(mm1_constants.I_MODEM, 'State') |
