# 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')