diff --git a/.release b/.release
index 02f219ab46a84e7ed5f50ed35b5e6536a431fc21..51faac07be14ca7c1987d83d647ec9ae96fec0fe 100644
--- a/.release
+++ b/.release
@@ -1,2 +1,2 @@
-release=0.6.3
-tag=lmcbaseclasses-0.6.3
+release=0.6.4
+tag=lmcbaseclasses-0.6.4
diff --git a/README.md b/README.md
index 55520e446b963fa62a4fc956ae363662ac7cfb0b..a416cd460894b21d9d1ffa861b858ad64a7916cb 100644
--- a/README.md
+++ b/README.md
@@ -25,6 +25,13 @@ The lmc-base-classe repository contains set of eight classes as mentioned in SKA
## Version History
+#### 0.6.4
+- Refactor state machine to use pytransitions library.
+- Minor behavioural change: Off() command is accepted in every obsState, rather
+than only EMPTY obsState.
+- support `straight_to_state` shortcuts to simplify test setups
+- Refactor of state machine testing to make it more portable
+
#### 0.6.3
- Fix omission of fatal_error transition from base device state machine.
diff --git a/docs/source/SKABaseDevice.rst b/docs/source/SKABaseDevice.rst
index 4e6bcb44ad1ec4d5feb0cb2a8ff2353eb78d963b..9d41032725d5449356f3100f4035213ed5a81c07 100644
--- a/docs/source/SKABaseDevice.rst
+++ b/docs/source/SKABaseDevice.rst
@@ -6,14 +6,6 @@
SKA BaseDevice
============================================
-The SKABaseDevice implements the basic device state machine, as illustrated
-below, but without, at present, a Standby state.
-
-.. image:: images/device_state_diagram.png
- :width: 400
- :alt: Diagram of the device state machine showing states and transitions
-
-
.. toctree::
:maxdepth: 2
diff --git a/docs/source/SKASubarray.rst b/docs/source/SKASubarray.rst
index 93f320456ed91dd4d416971c7368675030a78213..4db79893338aa3fbdf320ff54c852e1896d79b03 100644
--- a/docs/source/SKASubarray.rst
+++ b/docs/source/SKASubarray.rst
@@ -6,13 +6,6 @@
SKA Subarray
============================================
-The SKA Subarray device implements the observation state machine as defined
-in ADR-8:
-
-.. image:: images/ADR-8.png
- :width: 400
- :alt: Diagram of the observation state machine showing states and transitions
-
.. toctree::
:maxdepth: 2
diff --git a/docs/source/State_Machine.rst b/docs/source/State_Machine.rst
new file mode 100644
index 0000000000000000000000000000000000000000..ff7d1687b8cdd628b351bff00989fe4544c8bdc7
--- /dev/null
+++ b/docs/source/State_Machine.rst
@@ -0,0 +1,69 @@
+
+State Machine
+=============
+
+The state machine modules implements SKA's two fundamental state machines: the
+base device state machine, and the observation state machine.
+
+Base device state machine
+-------------------------
+The base device state machine provides basic state needed for all devices,
+covering initialisation, off and on states, and a fault state. This state
+machine is implemented by all SKA Tango devices that inherit from these LMC
+base classes, though some devices with standby power modes may need to
+implement further states.
+
+
+.. figure:: images/device_state_diagram.png
+ :width: 80%
+ :alt: Diagram of the device state machine, taken from SKA design
+ documentation, showing the state machine as designed
+
+ Diagram of the device state machine, taken from SKA design
+ documentation, showing the state machine as designed
+
+
+.. figure:: images/BaseDeviceStateMachine.png
+ :width: 80%
+ :alt: Diagram of the device state machine, automatically generated
+ from the state machine as specified in code.
+
+ Diagram of the device state machine, automatically generated from the
+ state machine as specified in code. The equivalence of this diagram to
+ the diagram previous demonstrates that the machine has been
+ implemented as designed.
+
+
+Observation state machine
+-------------------------
+The observation state machine is implemented by devices that manage
+observations (currently only subarray devices).
+
+.. figure:: images/ADR-8.png
+ :width: 80%
+ :alt: Diagram of the observation state machine, as decided and published in ADR-8.
+
+ Diagram of the observation state machine, as decided and published in ADR-8.
+
+
+.. figure:: images/ObservationStateMachine.png
+ :width: 80%
+ :alt: Diagram of the observation state machine, automatically generated from
+ the state machine as specified in code.
+
+ Diagram of the observation state machine, automatically generated from
+ the state machine as specified in code. The equivalance of this
+ diagram to the diagram previous demonstrates that the machine has been
+ implemented in conformance with ADR-8.
+
+API
+---
+
+.. toctree::
+ :maxdepth: 2
+
+
+.. automodule:: ska.base.state_machine
+ :members:
+ :undoc-members:
+
diff --git a/docs/source/images/BaseDeviceStateMachine.png b/docs/source/images/BaseDeviceStateMachine.png
new file mode 100644
index 0000000000000000000000000000000000000000..ca141bc982bdeededdcb9953e14c20c5b582292b
Binary files /dev/null and b/docs/source/images/BaseDeviceStateMachine.png differ
diff --git a/docs/source/images/ObservationStateMachine.png b/docs/source/images/ObservationStateMachine.png
new file mode 100644
index 0000000000000000000000000000000000000000..112c60101272b7c5e33f7f2071af25340fa6d576
Binary files /dev/null and b/docs/source/images/ObservationStateMachine.png differ
diff --git a/docs/source/index.rst b/docs/source/index.rst
index adec635929c105702b9a76a53ba22effc83ddd0f..9f38ed650a4c8e0c948d6d80cd0a7513f4dcc14c 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -22,6 +22,7 @@ Welcome to LMC Base Classes documentation!
SKA Control Model<Control_Model>
SKA Commands<Commands>
+ SKA State Machine<State_Machine>
Indices and tables
==================
@@ -29,4 +30,3 @@ Indices and tables
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`
-
diff --git a/setup.py b/setup.py
index 6d6fd488c012119aed6f461836304cf05b34efee..d3ca8bf0e8a33d55f0f1f81deb3dd16f8e6fe5d5 100644
--- a/setup.py
+++ b/setup.py
@@ -35,7 +35,7 @@ setuptools.setup(
],
platforms=["OS Independent"],
setup_requires=[] + pytest_runner,
- install_requires=["future", "ska_logging >= 0.3.0"],
+ install_requires=["future", "transitions", "ska_logging >= 0.3.0"],
tests_require=["pytest", "coverage", "pytest-json-report", "pytest-forked"],
entry_points={
"console_scripts": [
diff --git a/src/ska/base/__init__.py b/src/ska/base/__init__.py
index cebd343850fee474796a5d3e1514374c0fa1dc40..049a6be104bcccb29eab16282adf8888bfa8f600 100644
--- a/src/ska/base/__init__.py
+++ b/src/ska/base/__init__.py
@@ -1,12 +1,13 @@
__all__ = (
"commands",
"control_model",
+ "state_machine",
"SKAAlarmHandler",
"SKABaseDevice", "SKABaseDeviceStateModel",
"SKACapability",
"SKALogger",
"SKAMaster",
- "SKAObsDevice", "SKAObsDeviceStateModel",
+ "SKAObsDevice",
"SKASubarray", "SKASubarrayStateModel", "SKASubarrayResourceManager",
"SKATelState",
)
@@ -21,7 +22,7 @@ from .master_device import SKAMaster
from .tel_state_device import SKATelState
# SKAObsDevice, and then classes that inherit from it
-from .obs_device import SKAObsDevice, SKAObsDeviceStateModel
+from .obs_device import SKAObsDevice
from .capability_device import SKACapability
from .subarray_device import (
SKASubarray, SKASubarrayStateModel, SKASubarrayResourceManager
diff --git a/src/ska/base/base_device.py b/src/ska/base/base_device.py
index 30852c8c0c30dff6bf93101c09c0eb58d1dd70a7..656922d2a75f48200be9f8b34e9a9b10f5af1881 100644
--- a/src/ska/base/base_device.py
+++ b/src/ska/base/base_device.py
@@ -19,7 +19,7 @@ import socket
import sys
import threading
import warnings
-
+from transitions import MachineError
from urllib.parse import urlparse
from urllib.request import url2pathname
@@ -35,10 +35,12 @@ from ska.base.commands import (
)
from ska.base.control_model import (
AdminMode, ControlMode, SimulationMode, TestMode, HealthState,
- LoggingLevel, DeviceStateModel
+ LoggingLevel
)
+from ska.base.faults import StateModelError
+from ska.base.state_machine import BaseDeviceStateMachine
-from ska.base.utils import get_groups_from_json
+from ska.base.utils import get_groups_from_json, for_testing_only
from ska.base.faults import (GroupDefinitionsError,
LoggingTargetError,
LoggingLevelError)
@@ -307,211 +309,40 @@ class LoggingUtils:
# PROTECTED REGION END # // SKABaseDevice.additionnal_import
-__all__ = ["SKABaseDevice", "SKABaseDeviceStateModel", "main"]
+__all__ = ["SKABaseDevice", "main"]
-class SKABaseDeviceStateModel(DeviceStateModel):
+class SKABaseDeviceStateModel:
"""
Implements the state model for the SKABaseDevice
"""
- __transitions = {
- ('UNINITIALISED', 'init_started'): (
- "INIT (ENABLED)",
- lambda self: (
- self._set_admin_mode(AdminMode.MAINTENANCE),
- self._set_dev_state(DevState.INIT),
- )
- ),
- ('INIT (ENABLED)', 'init_succeeded'): (
- 'OFF',
- lambda self: self._set_dev_state(DevState.OFF)
- ),
- ('INIT (ENABLED)', 'init_failed'): (
- 'FAULT (ENABLED)',
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('INIT (ENABLED)', 'fatal_error'): (
- "FAULT (ENABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('INIT (ENABLED)', 'to_notfitted'): (
- "INIT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
- ),
- ('INIT (ENABLED)', 'to_offline'): (
- "INIT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.OFFLINE)
- ),
- ('INIT (ENABLED)', 'to_maintenance'): (
- "INIT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
- ),
- ('INIT (ENABLED)', 'to_online'): (
- "INIT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.ONLINE)
- ),
- ('INIT (DISABLED)', 'init_succeeded'): (
- 'DISABLED',
- lambda self: self._set_dev_state(DevState.DISABLE)
- ),
- ('INIT (DISABLED)', 'init_failed'): (
- 'FAULT (DISABLED)',
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('INIT (DISABLED)', 'fatal_error'): (
- "FAULT (DISABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('INIT (DISABLED)', 'to_notfitted'): (
- "INIT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
- ),
- ('INIT (DISABLED)', 'to_offline'): (
- "INIT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.OFFLINE)
- ),
- ('INIT (DISABLED)', 'to_maintenance'): (
- "INIT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
- ),
- ('INIT (DISABLED)', 'to_online'): (
- "INIT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.ONLINE)
- ),
- ('FAULT (DISABLED)', 'reset_succeeded'): (
- "DISABLED",
- lambda self: self._set_dev_state(DevState.DISABLE)
- ),
- ('FAULT (DISABLED)', 'reset_failed'): ("FAULT (DISABLED)", None),
- ('FAULT (DISABLED)', 'fatal_error'): ("FAULT (DISABLED)", None),
- ('FAULT (DISABLED)', 'to_notfitted'): (
- "FAULT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
- ),
- ('FAULT (DISABLED)', 'to_offline'): (
- "FAULT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.OFFLINE)
- ),
- ('FAULT (DISABLED)', 'to_maintenance'): (
- "FAULT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
- ),
- ('FAULT (DISABLED)', 'to_online'): (
- "FAULT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.ONLINE)
- ),
- ('FAULT (ENABLED)', 'reset_succeeded'): (
- "OFF",
- lambda self: self._set_dev_state(DevState.OFF)
- ),
- ('FAULT (ENABLED)', 'reset_failed'): ("FAULT (ENABLED)", None),
- ('FAULT (ENABLED)', 'fatal_error'): ("FAULT (ENABLED)", None),
- ('FAULT (ENABLED)', 'to_notfitted'): (
- "FAULT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)),
- ('FAULT (ENABLED)', 'to_offline'): (
- "FAULT (DISABLED)",
- lambda self: self._set_admin_mode(AdminMode.OFFLINE)),
- ('FAULT (ENABLED)', 'to_maintenance'): (
- "FAULT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
- ),
- ('FAULT (ENABLED)', 'to_online'): (
- "FAULT (ENABLED)",
- lambda self: self._set_admin_mode(AdminMode.ONLINE)
- ),
- ('DISABLED', 'to_offline'): (
- "DISABLED",
- lambda self: self._set_admin_mode(AdminMode.OFFLINE)
- ),
- ('DISABLED', 'to_online'): (
- "OFF",
- lambda self: (
- self._set_admin_mode(AdminMode.ONLINE),
- self._set_dev_state(DevState.OFF)
- )
- ),
- ('DISABLED', 'to_maintenance'): (
- "OFF",
- lambda self: (
- self._set_admin_mode(AdminMode.MAINTENANCE),
- self._set_dev_state(DevState.OFF)
- )
- ),
- ('DISABLED', 'to_notfitted'): (
- "DISABLED",
- lambda self: self._set_admin_mode(AdminMode.NOT_FITTED)
- ),
- ('DISABLED', 'fatal_error'): (
- "FAULT (DISABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('OFF', 'to_notfitted'): (
- "DISABLED",
- lambda self: (
- self._set_admin_mode(AdminMode.NOT_FITTED),
- self._set_dev_state(DevState.DISABLE)
- )
- ),
- ('OFF', 'to_offline'): (
- "DISABLED", lambda self: (
- self._set_admin_mode(AdminMode.OFFLINE),
- self._set_dev_state(DevState.DISABLE)
- )
- ),
- ('OFF', 'to_online'): (
- "OFF",
- lambda self: self._set_admin_mode(AdminMode.ONLINE)
- ),
- ('OFF', 'to_maintenance'): (
- "OFF",
- lambda self: self._set_admin_mode(AdminMode.MAINTENANCE)
- ),
- ('OFF', 'fatal_error'): (
- "FAULT (ENABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('OFF', 'on_succeeded'): (
- "ON",
- lambda self: self._set_dev_state(DevState.ON)
- ),
- ('OFF', 'on_failed'): (
- "FAULT (ENABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('ON', 'off_succeeded'): (
- "OFF",
- lambda self: self._set_dev_state(DevState.OFF)
- ),
- ('ON', 'off_failed'): (
- "FAULT (ENABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('ON', 'fatal_error'): (
- "FAULT (ENABLED)",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
-
- }
-
- def __init__(self, dev_state_callback=None, admin_mode_callback=None):
+ def __init__(self, logger, op_state_callback=None, admin_mode_callback=None):
"""
Initialises the state model.
- :param dev_state_callback: A callback to be called when a
- transition implies a change to device state
- :type dev_state_callback: callable
+ :param logger: the logger to be used by this state model.
+ :type logger: a logger that implements the standard library
+ logger interface
+ :param op_state_callback: A callback to be called when a
+ transition implies a change to op state
+ :type op_state_callback: callable
:param admin_mode_callback: A callback to be called when a
transition causes a change to device admin_mode
:type admin_mode_callback: callable
"""
- super().__init__(self.__transitions, "UNINITIALISED")
+ self.logger = logger
+ self._op_state = None
self._admin_mode = None
+
+ self._op_state_callback = op_state_callback
self._admin_mode_callback = admin_mode_callback
- self._dev_state = None
- self._dev_state_callback = dev_state_callback
+
+ self._state_machine = BaseDeviceStateMachine(
+ op_state_callback=self._update_op_state,
+ admin_mode_callback=self._update_admin_mode
+ )
@property
def admin_mode(self):
@@ -523,40 +354,124 @@ class SKABaseDeviceStateModel(DeviceStateModel):
"""
return self._admin_mode
- def _set_admin_mode(self, admin_mode):
+ def _update_admin_mode(self, admin_mode):
"""
- Helper method: calls the admin_mode callback if one exists
+ Helper method that updates admin_mode, ensuring that the callback is
+ called if one exists.
- :param admin_mode: the new admin_mode value
+ :param admin_mode: the new adminMode attribute value
:type admin_mode: AdminMode
"""
if self._admin_mode != admin_mode:
self._admin_mode = admin_mode
if self._admin_mode_callback is not None:
- self._admin_mode_callback(self._admin_mode)
+ self._admin_mode_callback(admin_mode)
@property
- def dev_state(self):
+ def op_state(self):
"""
- Returns the dev_state
+ Returns the op_state
- :returns: dev_state of this state model
+ :returns: op_state of this state model
:rtype: tango.DevState
"""
- return self._dev_state
+ return self._op_state
+
+ def _update_op_state(self, op_state):
+ """
+ Helper method that updates op_state, ensuring that the callback is
+ called if one exists.
- def _set_dev_state(self, dev_state):
+ :param op_state: the new opState attribute value
+ :type op_state: tango.DevState
"""
- Helper method: sets this state models dev_state, and calls the
- dev_state callback if one exists
+ if self._op_state != op_state:
+ self._op_state = op_state
+ if self._op_state_callback is not None:
+ self._op_state_callback(op_state)
- :param dev_state: the new state value
- :type admin_mode: DevState
+ def is_action_allowed(self, action):
"""
- if self._dev_state != dev_state:
- self._dev_state = dev_state
- if self._dev_state_callback is not None:
- self._dev_state_callback(self._dev_state)
+ Whether a given action is allowed in the current state.
+
+ :param action: an action, as given in the transitions table
+ :type action: ANY
+ """
+ return action in self._state_machine.get_triggers(self._state_machine.state)
+
+ def try_action(self, action):
+ """
+ Checks whether a given action is allowed in the current state,
+ and raises a StateModelError if it is not.
+
+ :param action: an action, as given in the transitions table
+ :type action: ANY
+
+ :raises StateModelError: if the action is not allowed in the
+ current state
+
+ :returns: True if the action is allowed
+ :rtype: boolean
+ """
+ if not self.is_action_allowed(action):
+ raise StateModelError(
+ f"Action '{action}' not allowed in current state ({self._state_machine.state})."
+ )
+ return True
+
+ def perform_action(self, action):
+ """
+ Performs an action on the state model
+
+ :param action: an action, as given in the transitions table
+ :type action: ANY
+ :raises StateModelError: if the action is not allowed in the
+ current state
+
+ """
+ try:
+ self._state_machine.trigger(action)
+ except MachineError as error:
+ raise StateModelError(error)
+
+ @for_testing_only
+ def _straight_to_state(self, state):
+ """
+ Takes the DeviceStateModel straight to the specified state. This method
+ exists to simplify testing; for example, if testing that a command may
+ be run in a given state, one can push the state model straight to that
+ state, rather than having to drive it to that state through a sequence
+ of actions. It is not intended that this method would be called outside
+ of test setups. A warning will be raised if it is.
+
+ Note that states are non-deterministic with respect to adminMode. For
+ example, in state "FAULT-DISABLED", the adminMode could be OFFLINE or
+ NOT_FITTED. When you drive the state machine through its transitions,
+ the adminMode will be set accordingly. When using this method, the
+ adminMode will simply be set to something sensible.
+
+ :param state: the target state
+ :type state: string
+ """
+ if state == "UNINITIALISED":
+ pass
+ elif "DISABLED" in state:
+ if self._admin_mode not in [AdminMode.OFFLINE, AdminMode.NOT_FITTED]:
+ self._state_machine._update_admin_mode(AdminMode.OFFLINE)
+ else:
+ if self._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
+ self._state_machine._update_admin_mode(AdminMode.ONLINE)
+
+ getattr(self._state_machine, f"to_{state}")()
+
+ @property
+ def _state(self):
+ """
+ Returns the state of the underlying state machine. This would normally
+ be a hidden implementation detail, but is exposed here for testing
+ purposes.
+ """
+ return self._state_machine.state
class SKABaseDevice(Device):
@@ -578,8 +493,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -830,11 +744,25 @@ class SKABaseDevice(Device):
self.set_status(f"The device is in {state} state.")
def set_state(self, state):
+ """
+ Helper method for setting device state, ensuring that change
+ events are pushed.
+
+ :param state: the new state
+ :type state: tango.DevState
+ """
super().set_state(state)
self.push_change_event('state')
self.push_archive_event('state')
def set_status(self, status):
+ """
+ Helper method for setting device status, ensuring that change
+ events are pushed.
+
+ :param status: the new status
+ :type status: str
+ """
super().set_status(status)
self.push_change_event('status')
self.push_archive_event('status')
@@ -874,17 +802,43 @@ class SKABaseDevice(Device):
Creates the state model for the device
"""
self.state_model = SKABaseDeviceStateModel(
- dev_state_callback=self._update_state,
+ logger=self.logger,
+ op_state_callback=self._update_state,
admin_mode_callback=self._update_admin_mode
)
def register_command_object(self, command_name, command_object):
+ """
+ Registers a command object as the object to handle invocations
+ of a given command
+
+ :param command_name: name of the command for which the object is
+ being registered
+ :type command_name: str
+ :param command_object: the object that will handle invocations
+ of the given command
+ :type command_object: Command instance
+ """
self._command_objects[command_name] = command_object
def get_command_object(self, command_name):
+ """
+ Returns the command object (handler) for a given command.
+
+ :param command_name: name of the command for which a command
+ object (handler) is sought
+ :type command_name: str
+
+ :return: the registered command object (handler) for the command
+ :rtype: Command instance
+ """
return self._command_objects[command_name]
def init_command_objects(self):
+ """
+ Creates and registers command objects (handlers) for the
+ commands supported by this device.
+ """
device_args = (self, self.state_model, self.logger)
self.register_command_object("On", self.OnCommand(*device_args))
@@ -1164,8 +1118,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -1237,8 +1190,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -1264,6 +1216,7 @@ class SKABaseDevice(Device):
Check if command `On` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1301,8 +1254,7 @@ class SKABaseDevice(Device):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKABaseDeviceStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -1328,6 +1280,7 @@ class SKABaseDevice(Device):
Check if command `Off` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
diff --git a/src/ska/base/commands.py b/src/ska/base/commands.py
index 1d6fc20d3cc00bfb3bf5d4c85fe864708574d773..b8d70baa16f4d10d9fe57dfd7f0fdc9649efe2c7 100644
--- a/src/ska/base/commands.py
+++ b/src/ska/base/commands.py
@@ -153,7 +153,8 @@ class BaseCommand:
return self.state_model.try_action(action)
except StateModelError as exc:
raise CommandError(
- f"Error executing command {self.name}") from exc
+ f"Error executing command {self.name}"
+ ) from exc
def _perform_action(self, action):
"""
@@ -167,6 +168,12 @@ class BaseCommand:
class ResponseCommand(BaseCommand):
+ """
+ Abstract base class for a tango command handler, for commands that
+ execute a procedure/operation and return a (ResultCode, message)
+ tuple.
+ """
+
def __call__(self, argin=None):
"""
What to do when the command is called. This base class simply
diff --git a/src/ska/base/control_model.py b/src/ska/base/control_model.py
index 6f500d55fb9301572a51b026327033b94162c7db..cfc5075708d6d1f7c54c9a1239a57994ecd19905 100644
--- a/src/ska/base/control_model.py
+++ b/src/ska/base/control_model.py
@@ -14,13 +14,11 @@ other useful enumerations.
"""
import enum
-from ska.base.faults import StateModelError
+
# ---------------------------------
# Core SKA Control Model attributes
# ---------------------------------
-
-
class HealthState(enum.IntEnum):
"""Python enumerated type for ``healthState`` attribute."""
@@ -328,94 +326,3 @@ class LoggingLevel(enum.IntEnum):
WARNING = 3
INFO = 4
DEBUG = 5
-
-
-class DeviceStateModel:
- """
- Base class for the state model used by SKA devices.
- """
-
- def __init__(self, transitions, initial_state):
- """
- Create a new device state model.
-
- :param transitions: a dictionary for which each key is a (state,
- event) tuple, and each value is a (state, side-effect)
- tuple. When the device is in state `IN-STATE`, and action
- `ACTION` is attempted, the transitions table will be checked
- for an entry under key `(IN-STATE, EVENT)`. If no such key
- exists, the action will be denied and a model will raise a
- `StateModelError`. If the key does exist, then its value
- `(OUT-STATE, SIDE-EFFECT)` will result in the model
- transitioning to state `OUT-STATE`, and executing
- `SIDE-EFFECT`, which must be a function or lambda that
- takes a single parameter - a model instance.
- :type transitions: dict
- :param initial_state: the starting state of the model
- :type initial_state: a state with an entry in the transitions
- table
- """
- # instances may update transitions dict, so need a copy
- self._transitions = transitions.copy()
- self._state = initial_state
-
- @property
- def state(self):
- """Return current state as a string."""
- return self._state
-
- def update_transitions(self, transitions):
- """
- Update the transitions table with new transitions.
-
- :param transitions: new transitions to be included in the
- transitions table. Transitions with pre-existing keys will
- replace the transitions for that key. Transitions with novel
- keys will be added. There is no provision for removing
- transitions
- :type transitions: dict
- """
- self._transitions.update(transitions)
-
- def is_action_allowed(self, action):
- """
- Whether a given action is allowed in the current state.
-
- :param action: an action, as given in the transitions table
- :type action: ANY
- """
- return (self._state, action) in self._transitions
-
- def try_action(self, action):
- """
- Checks whether a given action is allowed in the current state,
- and raises a StateModelError if it is not.
-
- :param action: an action, as given in the transitions table
- :type action: ANY
- :raises StateModelError: if the action is not allowed in the
- current state
- :returns: True if the action is allowed
- :rtype: boolean
- """
- if not self.is_action_allowed(action):
- raise StateModelError(
- f"Action '{action}' not allowed in current state ({self._state})."
- )
- return True
-
- def perform_action(self, action):
- """
- Performs an action on the state model
-
- :param action: an action, as given in the transitions table
- :type action: ANY
- :raises StateModelError: if the action is not allowed in the
- current state
-
- """
- self.try_action(action)
-
- (self._state, side_effect) = self._transitions[(self._state, action)]
- if side_effect is not None:
- side_effect(self)
diff --git a/src/ska/base/obs_device.py b/src/ska/base/obs_device.py
index 48245f348788cfcc809d6c15a93dd2de1dcd5feb..9119cc0aeb2ebc65546f8aa74325809a31db3795 100644
--- a/src/ska/base/obs_device.py
+++ b/src/ska/base/obs_device.py
@@ -14,78 +14,15 @@ instead of just SKABaseDevice.
# Additional import
# PROTECTED REGION ID(SKAObsDevice.additionnal_import) ENABLED START #
# Tango imports
-from tango import DevState
from tango.server import run, attribute
# SKA specific imports
-from ska.base import SKABaseDevice, SKABaseDeviceStateModel
+from ska.base import SKABaseDevice
from ska.base.commands import ResultCode
-from ska.base.control_model import AdminMode, ObsMode, ObsState
+from ska.base.control_model import ObsMode, ObsState
# PROTECTED REGION END # // SKAObsDevice.additionnal_imports
-__all__ = ["SKAObsDevice", "SKAObsDeviceStateModel", "main"]
-
-
-class SKAObsDeviceStateModel(SKABaseDeviceStateModel):
- """
- Implements the state model for the SKABaseDevice
- """
- def __init__(
- self,
- dev_state_callback=None,
- admin_mode_callback=None,
- obs_state_callback=None
- ):
- """
- Initialises the model. Note that this does not imply moving to
- INIT state. The INIT state is managed by the model itself.
-
- :param dev_state_callback: A callback to be called when a
- transition implies a change to device state
- :type dev_state_callback: callable
- :param admin_mode_callback: A callback to be called when a
- transition causes a change to device admin_mode
- :type admin_mode_callback: callable
- :param obs_state_callback: A callback to be called when a
- transition causes a change to device obs_state
- :type obs_state_callback: callable
- """
- super().__init__(
- dev_state_callback=dev_state_callback,
- admin_mode_callback=admin_mode_callback
- )
- self._obs_state_callback = obs_state_callback
-
- self.update_transitions(
- {
- ('UNINITIALISED', 'init_started'): (
- "INIT (ENABLED)",
- lambda self: (
- self._set_admin_mode(AdminMode.MAINTENANCE),
- self._set_dev_state(DevState.INIT),
- self._set_obs_state(ObsState.EMPTY)
- )
- )
- }
- )
- self._obs_state = None
-
- def _set_obs_state(self, obs_state):
- """
- Helper method: set the value of obs_state value, and calls the
- obs_state_callback if one exists.
-
- :param obs_state: the new obs_state value
- :type obs_state: ObsState
- """
- if self._obs_state != obs_state:
- self._obs_state = obs_state
- if self._obs_state_callback is not None:
- self._obs_state_callback(self._obs_state)
-
- @property
- def obs_state(self):
- return self._obs_state
+__all__ = ["SKAObsDevice", "main"]
class SKAObsDevice(SKABaseDevice):
@@ -111,6 +48,7 @@ class SKAObsDevice(SKABaseDevice):
device.set_change_event("obsState", True, True)
device.set_archive_event("obsState", True, True)
+ device._obs_state = ObsState.EMPTY
device._obs_mode = ObsMode.IDLE
device._config_progress = 0
device._config_delay_expected = 0
@@ -166,26 +104,27 @@ class SKAObsDevice(SKABaseDevice):
:param obs_state: the new obs_state value
:type obs_state: ObsState
"""
+ self._obs_state = obs_state
self.push_change_event("obsState", obs_state)
self.push_archive_event("obsState", obs_state)
- def _init_state_model(self):
- """
- Sets up the state model for the device
- """
- self.state_model = SKAObsDeviceStateModel(
- dev_state_callback=self._update_state,
- admin_mode_callback=self._update_admin_mode,
- obs_state_callback=self._update_obs_state
- )
-
def always_executed_hook(self):
# PROTECTED REGION ID(SKAObsDevice.always_executed_hook) ENABLED START #
+ """
+ Method that is always executed before any device command gets executed.
+
+ :return: None
+ """
pass
# PROTECTED REGION END # // SKAObsDevice.always_executed_hook
def delete_device(self):
# PROTECTED REGION ID(SKAObsDevice.delete_device) ENABLED START #
+ """
+ Method to cleanup when device is stopped.
+
+ :return: None
+ """
pass
# PROTECTED REGION END # // SKAObsDevice.delete_device
@@ -196,7 +135,7 @@ class SKAObsDevice(SKABaseDevice):
def read_obsState(self):
# PROTECTED REGION ID(SKAObsDevice.obsState_read) ENABLED START #
"""Reads Observation State of the device"""
- return self.state_model.obs_state
+ return self._obs_state
# PROTECTED REGION END # // SKAObsDevice.obsState_read
def read_obsMode(self):
diff --git a/src/ska/base/release.py b/src/ska/base/release.py
index 1afbbc52106869070e5e007bea9f57606139e4eb..22e3240fb99e973a8ee33e49ae2fa2b3df5249cb 100644
--- a/src/ska/base/release.py
+++ b/src/ska/base/release.py
@@ -7,7 +7,7 @@
"""Release information for lmc-base-classes Python Package"""
name = """lmcbaseclasses"""
-version = "0.6.3"
+version = "0.6.4"
version_info = version.split(".")
description = """A set of generic base devices for SKA Telescope."""
author = "SKA India and SARAO and CSIRO"
diff --git a/src/ska/base/state_machine.py b/src/ska/base/state_machine.py
new file mode 100644
index 0000000000000000000000000000000000000000..044a97df199f88fd176b370b1e31192801d847d7
--- /dev/null
+++ b/src/ska/base/state_machine.py
@@ -0,0 +1,469 @@
+"""
+This module contains specifications of SKA state machines.
+"""
+from transitions import Machine, State
+from tango import DevState
+
+from ska.base.control_model import AdminMode, ObsState
+
+
+class BaseDeviceStateMachine(Machine):
+ """
+ State machine for an SKA base device. Supports ON and OFF states,
+ states, plus initialisation and fault states, and
+ also the basic admin modes.
+ """
+
+ def __init__(self, op_state_callback=None, admin_mode_callback=None):
+ """
+ Initialises the state model.
+
+ :param op_state_callback: A callback to be called when a
+ transition implies a change to op state
+ :type op_state_callback: callable
+ :param admin_mode_callback: A callback to be called when a
+ transition causes a change to device admin_mode
+ :type admin_mode_callback: callable
+ """
+ self._admin_mode = None
+ self._admin_mode_callback = admin_mode_callback
+ self._op_state = None
+ self._op_state_callback = op_state_callback
+
+ states = [
+ State("UNINITIALISED"),
+ State("INIT_ENABLED", on_enter="_init_entered"),
+ State("INIT_DISABLED", on_enter="_init_entered"),
+ State("FAULT_ENABLED", on_enter="_fault_entered"),
+ State("FAULT_DISABLED", on_enter="_fault_entered"),
+ State("DISABLED", on_enter="_disabled_entered"),
+ State("OFF", on_enter="_off_entered"),
+ State("ON", on_enter="_on_entered"),
+ ]
+
+ transitions = [
+ {
+ "source": "UNINITIALISED",
+ "trigger": "init_started",
+ "dest": "INIT_ENABLED",
+ "after": self._maintenance_callback,
+ },
+ {
+ "source": ["INIT_ENABLED", "OFF", "FAULT_ENABLED", "ON"],
+ "trigger": "fatal_error",
+ "dest": "FAULT_ENABLED",
+ },
+ {
+ "source": ["INIT_DISABLED", "DISABLED", "FAULT_DISABLED"],
+ "trigger": "fatal_error",
+ "dest": "FAULT_DISABLED",
+ },
+ {
+ "source": "INIT_ENABLED",
+ "trigger": "init_succeeded",
+ "dest": "OFF",
+ },
+ {
+ "source": "INIT_DISABLED",
+ "trigger": "init_succeeded",
+ "dest": "DISABLED",
+ },
+ {
+ "source": "INIT_ENABLED",
+ "trigger": "init_failed",
+ "dest": "FAULT_ENABLED",
+ },
+ {
+ "source": "INIT_DISABLED",
+ "trigger": "init_failed",
+ "dest": "FAULT_DISABLED",
+ },
+ {
+ "source": ["INIT_ENABLED", "INIT_DISABLED"],
+ "trigger": "to_notfitted",
+ "dest": "INIT_DISABLED",
+ "after": self._not_fitted_callback
+ },
+ {
+ "source": ["INIT_ENABLED", "INIT_DISABLED"],
+ "trigger": "to_offline",
+ "dest": "INIT_DISABLED",
+ "after": self._offline_callback
+ },
+ {
+ "source": ["INIT_ENABLED", "INIT_DISABLED"],
+ "trigger": "to_maintenance",
+ "dest": "INIT_ENABLED",
+ "after": self._maintenance_callback
+ },
+ {
+ "source": ["INIT_ENABLED", "INIT_DISABLED"],
+ "trigger": "to_online",
+ "dest": "INIT_ENABLED",
+ "after": self._online_callback
+ },
+ {
+ "source": "FAULT_DISABLED",
+ "trigger": "reset_succeeded",
+ "dest": "DISABLED",
+ },
+ {
+ "source": "FAULT_ENABLED",
+ "trigger": "reset_succeeded",
+ "dest": "OFF",
+ },
+ {
+ "source": ["FAULT_DISABLED", "FAULT_ENABLED"],
+ "trigger": "reset_failed",
+ "dest": None,
+ },
+ {
+ "source": ["FAULT_DISABLED", "FAULT_ENABLED"],
+ "trigger": "to_notfitted",
+ "dest": "FAULT_DISABLED",
+ "after": self._not_fitted_callback
+ },
+ {
+ "source": ["FAULT_DISABLED", "FAULT_ENABLED"],
+ "trigger": "to_offline",
+ "dest": "FAULT_DISABLED",
+ "after": self._offline_callback
+ },
+ {
+ "source": ["FAULT_DISABLED", "FAULT_ENABLED"],
+ "trigger": "to_maintenance",
+ "dest": "FAULT_ENABLED",
+ "after": self._maintenance_callback
+ },
+ {
+ "source": ["FAULT_DISABLED", "FAULT_ENABLED"],
+ "trigger": "to_online",
+ "dest": "FAULT_ENABLED",
+ "after": self._online_callback
+ },
+ {
+ "source": "DISABLED",
+ "trigger": "to_notfitted",
+ "dest": "DISABLED",
+ "after": self._not_fitted_callback
+ },
+ {
+ "source": "DISABLED",
+ "trigger": "to_offline",
+ "dest": "DISABLED",
+ "after": self._offline_callback
+ },
+ {
+ "source": "DISABLED",
+ "trigger": "to_maintenance",
+ "dest": "OFF",
+ "after": self._maintenance_callback
+ },
+ {
+ "source": "DISABLED",
+ "trigger": "to_online",
+ "dest": "OFF",
+ "after": self._online_callback
+ },
+ {
+ "source": "OFF",
+ "trigger": "to_notfitted",
+ "dest": "DISABLED",
+ "after": self._not_fitted_callback
+ },
+ {
+ "source": "OFF",
+ "trigger": "to_offline",
+ "dest": "DISABLED",
+ "after": self._offline_callback
+ },
+ {
+ "source": "OFF",
+ "trigger": "to_maintenance",
+ "dest": "OFF",
+ "after": self._maintenance_callback
+ },
+ {
+ "source": "OFF",
+ "trigger": "to_online",
+ "dest": "OFF",
+ "after": self._online_callback
+ },
+ {
+ "source": "OFF",
+ "trigger": "on_succeeded",
+ "dest": "ON",
+ },
+ {
+ "source": "OFF",
+ "trigger": "on_failed",
+ "dest": "FAULT_ENABLED",
+ },
+ {
+ "source": "ON",
+ "trigger": "off_succeeded",
+ "dest": "OFF",
+ },
+ {
+ "source": "ON",
+ "trigger": "off_failed",
+ "dest": "FAULT_ENABLED",
+ },
+ ]
+
+ super().__init__(
+ states=states,
+ initial="UNINITIALISED",
+ transitions=transitions,
+ )
+
+ def _init_entered(self):
+ """
+ called when the state machine enters the "" state.
+ """
+ self._update_op_state(DevState.INIT)
+
+ def _fault_entered(self):
+ """
+ called when the state machine enters the "" state.
+ """
+ self._update_op_state(DevState.FAULT)
+
+ def _disabled_entered(self):
+ """
+ called when the state machine enters the "" state.
+ """
+ self._update_op_state(DevState.DISABLE)
+
+ def _off_entered(self):
+ """
+ called when the state machine enters the "" state.
+ """
+ self._update_op_state(DevState.OFF)
+
+ def _on_entered(self):
+ """
+ called when the state machine enters the "" state.
+ """
+ self._update_op_state(DevState.ON)
+
+ def _not_fitted_callback(self):
+ """
+ callback called when the state machine is set to admin mode
+ NOT FITTED
+ """
+ self._update_admin_mode(AdminMode.NOT_FITTED)
+
+ def _offline_callback(self):
+ """
+ callback called when the state machine is set to admin mode
+ OFFLINE
+ """
+ self._update_admin_mode(AdminMode.OFFLINE)
+
+ def _maintenance_callback(self):
+ """
+ callback called when the state machine is set to admin mode
+ MAINTENANCE
+ """
+ self._update_admin_mode(AdminMode.MAINTENANCE)
+
+ def _online_callback(self):
+ """
+ callback called when the state machine is set to admin mode
+ online
+ """
+ self._update_admin_mode(AdminMode.ONLINE)
+
+ def _update_admin_mode(self, admin_mode):
+ """
+ Helper method: calls the admin_mode callback if one exists
+
+ :param admin_mode: the new admin_mode value
+ :type admin_mode: AdminMode
+ """
+ if self._admin_mode != admin_mode:
+ self._admin_mode = admin_mode
+ if self._admin_mode_callback is not None:
+ self._admin_mode_callback(self._admin_mode)
+
+ def _update_op_state(self, op_state):
+ """
+ Helper method: sets this state models op_state, and calls the
+ op_state callback if one exists
+
+ :param op_state: the new op state value
+ :type op_state: DevState
+ """
+ if self._op_state != op_state:
+ self._op_state = op_state
+ if self._op_state_callback is not None:
+ self._op_state_callback(self._op_state)
+
+
+class ObservationStateMachine(Machine):
+ """
+ The observation state machine used by an observing subarray, per
+ ADR-8.
+ """
+
+ def __init__(self, obs_state_callback=None):
+ """
+ Initialises the model.
+
+ :param obs_state_callback: A callback to be called when a
+ transition causes a change to device obs_state
+ :type obs_state_callback: callable
+ """
+ self._obs_state = ObsState.EMPTY
+ self._obs_state_callback = obs_state_callback
+
+ states = [obs_state.name for obs_state in ObsState]
+ transitions = [
+ {
+ "source": "*",
+ "trigger": "fatal_error",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": [ObsState.EMPTY.name, ObsState.IDLE.name],
+ "trigger": "assign_started",
+ "dest": ObsState.RESOURCING.name,
+ },
+ {
+ "source": ObsState.IDLE.name,
+ "trigger": "release_started",
+ "dest": ObsState.RESOURCING.name,
+ },
+ {
+ "source": ObsState.RESOURCING.name,
+ "trigger": "resourcing_succeeded_some_resources",
+ "dest": ObsState.IDLE.name,
+ },
+ {
+ "source": ObsState.RESOURCING.name,
+ "trigger": "resourcing_succeeded_no_resources",
+ "dest": ObsState.EMPTY.name,
+ },
+ {
+ "source": ObsState.RESOURCING.name,
+ "trigger": "resourcing_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": [ObsState.IDLE.name, ObsState.READY.name],
+ "trigger": "configure_started",
+ "dest": ObsState.CONFIGURING.name,
+ },
+ {
+ "source": ObsState.CONFIGURING.name,
+ "trigger": "configure_succeeded",
+ "dest": ObsState.READY.name,
+ },
+ {
+ "source": ObsState.CONFIGURING.name,
+ "trigger": "configure_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": ObsState.READY.name,
+ "trigger": "end_succeeded",
+ "dest": ObsState.IDLE.name,
+ },
+ {
+ "source": ObsState.READY.name,
+ "trigger": "end_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": ObsState.READY.name,
+ "trigger": "scan_started",
+ "dest": ObsState.SCANNING.name,
+ },
+ {
+ "source": ObsState.SCANNING.name,
+ "trigger": "scan_succeeded",
+ "dest": ObsState.READY.name,
+ },
+ {
+ "source": ObsState.SCANNING.name,
+ "trigger": "scan_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": ObsState.SCANNING.name,
+ "trigger": "end_scan_succeeded",
+ "dest": ObsState.READY.name,
+ },
+ {
+ "source": ObsState.SCANNING.name,
+ "trigger": "end_scan_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": [
+ ObsState.IDLE.name, ObsState.CONFIGURING.name,
+ ObsState.READY.name, ObsState.SCANNING.name,
+ ],
+ "trigger": "abort_started",
+ "dest": ObsState.ABORTING.name,
+ },
+ {
+ "source": ObsState.ABORTING.name,
+ "trigger": "abort_succeeded",
+ "dest": ObsState.ABORTED.name,
+ },
+ {
+ "source": ObsState.ABORTING.name,
+ "trigger": "abort_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": [ObsState.ABORTED.name, ObsState.FAULT.name],
+ "trigger": "obs_reset_started",
+ "dest": ObsState.RESETTING.name,
+ },
+ {
+ "source": ObsState.RESETTING.name,
+ "trigger": "obs_reset_succeeded",
+ "dest": ObsState.IDLE.name,
+ },
+ {
+ "source": ObsState.RESETTING.name,
+ "trigger": "obs_reset_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ {
+ "source": [ObsState.ABORTED.name, ObsState.FAULT.name],
+ "trigger": "restart_started",
+ "dest": ObsState.RESTARTING.name,
+ },
+ {
+ "source": ObsState.RESTARTING.name,
+ "trigger": "restart_succeeded",
+ "dest": ObsState.EMPTY.name,
+ },
+ {
+ "source": ObsState.RESTARTING.name,
+ "trigger": "restart_failed",
+ "dest": ObsState.FAULT.name,
+ },
+ ]
+
+ super().__init__(
+ states=states,
+ initial=ObsState.EMPTY.name,
+ transitions=transitions,
+ after_state_change=self._obs_state_changed
+ )
+
+ def _obs_state_changed(self):
+ """
+ State machine callback that is called every time the obs_state
+ changes. Responsible for ensuring that callbacks are called.
+ """
+ obs_state = ObsState[self.state]
+ if self._obs_state != obs_state:
+ self._obs_state = obs_state
+ if self._obs_state_callback is not None:
+ self._obs_state_callback(self._obs_state)
diff --git a/src/ska/base/subarray_device.py b/src/ska/base/subarray_device.py
index 54407ad74c6059d12a6ae22f4f377d519becbb9f..9a0fe49319e1973b121e67b3f3517b9bd8a557d1 100644
--- a/src/ska/base/subarray_device.py
+++ b/src/ska/base/subarray_device.py
@@ -12,216 +12,34 @@ information like assigned resources, configured capabilities, etc.
"""
# PROTECTED REGION ID(SKASubarray.additionnal_import) ENABLED START #
import json
+import warnings
from tango import DebugIt
-from tango import DevState
from tango.server import run, attribute, command
from tango.server import device_property
# SKA specific imports
-from ska.base import SKAObsDevice, SKAObsDeviceStateModel
+from ska.base import SKAObsDevice, SKABaseDeviceStateModel
from ska.base.commands import ActionCommand, ResultCode
-from ska.base.control_model import ObsState
-from ska.base.faults import CapabilityValidationError
+from ska.base.control_model import AdminMode, ObsState
+from ska.base.faults import CapabilityValidationError, StateModelError
+from ska.base.state_machine import ObservationStateMachine
+from ska.base.utils import for_testing_only
+
# PROTECTED REGION END # // SKASubarray.additionnal_imports
-__all__ = ["SKASubarray", "SKASubarrayResourceManager", "SKASubarrayStateModel", "main"]
+__all__ = ["SKASubarray", "main"]
-class SKASubarrayStateModel(SKAObsDeviceStateModel):
+class SKASubarrayStateModel(SKABaseDeviceStateModel):
"""
Implements the state model for the SKASubarray
"""
- __transitions = {
- ('OFF', 'on_succeeded'): (
- "EMPTY",
- lambda self: self._set_dev_state(DevState.ON)
- ),
- ('OFF', 'on_failed'): (
- "FAULT",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('EMPTY', 'off_succeeded'): (
- "OFF",
- lambda self: self._set_dev_state(DevState.OFF)
- ),
- ('EMPTY', 'off_failed'): (
- "FAULT",
- lambda self: self._set_dev_state(DevState.FAULT)
- ),
- ('EMPTY', 'assign_started'): (
- "RESOURCING",
- lambda self: self._set_obs_state(ObsState.RESOURCING)
- ),
- ('EMPTY', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('RESOURCING', 'resourcing_succeeded_some_resources'): (
- "IDLE",
- lambda self: self._set_obs_state(ObsState.IDLE)
- ),
- ('RESOURCING', 'resourcing_succeeded_no_resources'): (
- "EMPTY",
- lambda self: self._set_obs_state(ObsState.EMPTY)
- ),
- ('RESOURCING', 'resourcing_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('RESOURCING', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('IDLE', 'assign_started'): (
- "RESOURCING",
- lambda self: self._set_obs_state(ObsState.RESOURCING)
- ),
- ('IDLE', 'release_started'): (
- "RESOURCING",
- lambda self: self._set_obs_state(ObsState.RESOURCING)
- ),
- ('IDLE', 'configure_started'): (
- "CONFIGURING",
- lambda self: self._set_obs_state(ObsState.CONFIGURING)
- ),
- ('IDLE', 'abort_started'): (
- "ABORTING",
- lambda self: self._set_obs_state(ObsState.ABORTING)
- ),
- ('IDLE', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('CONFIGURING', 'configure_succeeded'): (
- "READY",
- lambda self: self._set_obs_state(ObsState.READY)
- ),
- ('CONFIGURING', 'configure_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('CONFIGURING', 'abort_started'): (
- "ABORTING",
- lambda self: self._set_obs_state(ObsState.ABORTING)
- ),
- ('CONFIGURING', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('READY', 'end_succeeded'): (
- "IDLE",
- lambda self: self._set_obs_state(ObsState.IDLE)
- ),
- ('READY', 'end_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('READY', 'configure_started'): (
- "CONFIGURING",
- lambda self: self._set_obs_state(ObsState.CONFIGURING)
- ),
- ('READY', 'abort_started'): (
- "ABORTING",
- lambda self: self._set_obs_state(ObsState.ABORTING)
- ),
- ('READY', 'scan_started'): (
- "SCANNING",
- lambda self: self._set_obs_state(ObsState.SCANNING)
- ),
- ('READY', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('SCANNING', 'scan_succeeded'): (
- "READY",
- lambda self: self._set_obs_state(ObsState.READY)
- ),
- ('SCANNING', 'scan_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('SCANNING', 'end_scan_succeeded'): (
- "READY",
- lambda self: self._set_obs_state(ObsState.READY)
- ),
- ('SCANNING', 'end_scan_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('SCANNING', 'abort_started'): (
- "ABORTING",
- lambda self: self._set_obs_state(ObsState.ABORTING)
- ),
- ('SCANNING', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('ABORTING', 'abort_succeeded'): (
- "ABORTED",
- lambda self: self._set_obs_state(ObsState.ABORTED)
- ),
- ('ABORTING', 'abort_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('ABORTING', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('ABORTED', 'obs_reset_started'): (
- "RESETTING",
- lambda self: self._set_obs_state(ObsState.RESETTING)
- ),
- ('ABORTED', 'restart_started'): (
- "RESTARTING",
- lambda self: self._set_obs_state(ObsState.RESTARTING)
- ),
- ('ABORTED', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('OBSFAULT', 'obs_reset_started'): (
- "RESETTING",
- lambda self: self._set_obs_state(ObsState.RESETTING)
- ),
- ('OBSFAULT', 'restart_started'): (
- "RESTARTING",
- lambda self: self._set_obs_state(ObsState.RESTARTING)
- ),
- ('OBSFAULT', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('RESETTING', 'obs_reset_succeeded'): (
- "IDLE",
- lambda self: self._set_obs_state(ObsState.IDLE)
- ),
- ('RESETTING', 'obs_reset_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('RESETTING', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('RESTARTING', 'restart_succeeded'): (
- "EMPTY",
- lambda self: self._set_obs_state(ObsState.EMPTY)
- ),
- ('RESTARTING', 'restart_failed'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- ('RESTARTING', 'fatal_error'): (
- "OBSFAULT",
- lambda self: self._set_obs_state(ObsState.FAULT)
- ),
- }
def __init__(
self,
- dev_state_callback=None,
+ logger,
+ op_state_callback=None,
admin_mode_callback=None,
obs_state_callback=None
):
@@ -229,9 +47,12 @@ class SKASubarrayStateModel(SKAObsDeviceStateModel):
Initialises the model. Note that this does not imply moving to
INIT state. The INIT state is managed by the model itself.
- :param dev_state_callback: A callback to be called when a
- transition implies a change to device state
- :type dev_state_callback: callable
+ :param logger: the logger to be used by this state model.
+ :type logger: a logger that implements the standard library
+ logger interface
+ :param op_state_callback: A callback to be called when a
+ transition implies a change to op state
+ :type op_state_callback: callable
:param admin_mode_callback: A callback to be called when a
transition causes a change to device admin_mode
:type admin_mode_callback: callable
@@ -240,11 +61,147 @@ class SKASubarrayStateModel(SKAObsDeviceStateModel):
:type obs_state_callback: callable
"""
super().__init__(
- dev_state_callback=dev_state_callback,
+ logger,
+ op_state_callback=op_state_callback,
admin_mode_callback=admin_mode_callback,
- obs_state_callback=obs_state_callback
)
- self.update_transitions(self.__transitions)
+
+ self._obs_state = ObsState.EMPTY
+ self._obs_state_callback = obs_state_callback
+
+ self._observation_state_machine = ObservationStateMachine(
+ self._update_obs_state
+ )
+
+ @property
+ def obs_state(self):
+ """
+ Returns the obs_state
+
+ :returns: obs_state of this state model
+ :rtype: ObsState
+ """
+ return self._obs_state
+
+ def _update_obs_state(self, obs_state):
+ """
+ Helper method that updates obs_state, ensuring that the callback
+ is called if one exists.
+
+ :param obs_state: the new obsState attribute value
+ :type obs_state: ObsState
+ """
+ if self._obs_state != obs_state:
+ self._obs_state = obs_state
+ if self._obs_state_callback is not None:
+ self._obs_state_callback(obs_state)
+
+ def is_action_allowed(self, action):
+ """
+ Whether a given action is allowed in the current state.
+
+ :param action: an action, as given in the transitions table
+ :type action: ANY
+ """
+ if self._state_machine.state == "ON":
+ if action in self._observation_state_machine.get_triggers(
+ self._observation_state_machine.state
+ ):
+ return True
+
+ return action in self._state_machine.get_triggers(self._state_machine.state)
+
+ def perform_action(self, action):
+ """
+ Performs an action on the state model
+
+ :param action: an action, as given in the transitions table
+ :type action: ANY
+
+ :raises StateModelError: if the action is not allowed in the
+ current state
+
+ """
+ if self._state_machine.state == "ON":
+ if action in self._observation_state_machine.get_triggers(
+ self._observation_state_machine.state
+ ):
+ self._observation_state_machine.trigger(action)
+ return
+
+ if action in self._state_machine.get_triggers(
+ self._state_machine.state
+ ):
+ if self._observation_state_machine.state != "EMPTY":
+ message = (
+ "Changing device state of a non-EMPTY observing device "
+ "should only be done as an emergency measure and may be "
+ "disallowed in future."
+ )
+ self.logger.warning(message)
+ warnings.warn(message, PendingDeprecationWarning)
+ self._observation_state_machine.to_EMPTY()
+ self._state_machine.trigger(action)
+ return
+
+ raise StateModelError(
+ f"Action {action} is not allowed in device state "
+ "{self._state_machine.state}, observation_state "
+ "{self._observation_state_machine.state}."
+ )
+
+ @for_testing_only
+ def _straight_to_state(self, state):
+ """
+ Takes the DeviceStateModel straight to the specified state. This method
+ exists to simplify testing; for example, if testing that a command may
+ be run in a given state, one can push the state model straight to that
+ state, rather than having to drive it to that state through a sequence
+ of actions. It is not intended that this method would be called outside
+ of test setups. A warning is raised if it is.
+
+ Note that states are non-deterministics with respect to adminMode. For
+ example, in state "FAULT-DISABLED", the adminMode could be OFFLINE or
+ NOT_FITTED. When you drive the state machine through its transitions,
+ the adminMode will be set accordingly. When using this method, the
+ adminMode will simply be set to something sensible.
+
+ :param state: the target state
+ :type state: string
+ """
+ if state == "UNINITIALISED":
+ pass
+ elif "DISABLED" in state:
+ if self._admin_mode not in [AdminMode.OFFLINE, AdminMode.NOT_FITTED]:
+ self._state_machine._update_admin_mode(AdminMode.OFFLINE)
+ else:
+ if self._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
+ self._state_machine._update_admin_mode(AdminMode.ONLINE)
+
+ to_state = getattr(self._observation_state_machine, f"to_{state}", None)
+ if to_state is not None:
+ self._state_machine.to_ON()
+ to_state()
+ return
+
+ to_state = getattr(self._state_machine, f"to_{state}", None)
+ if to_state is not None:
+ self._observation_state_machine.to_EMPTY()
+ to_state()
+ return
+
+ raise StateModelError(f"No such state {state}")
+
+ @property
+ def _state(self):
+ """
+ Returns the state of the underlying state machine. This would normally
+ be a hidden implementation detail, but is exposed here for testing
+ purposes.
+ """
+ if self._state_machine.state == "ON":
+ return self._observation_state_machine.state
+ return self._state_machine.state
class SKASubarrayResourceManager:
@@ -275,6 +232,7 @@ class SKASubarrayResourceManager:
:todo: Currently implemented for testing purposes to take a JSON
string encoding a dictionary with key 'example'. In future this
will take a collection of resources.
+
:param resources: JSON-encoding of a dictionary, with resources to
assign under key 'example'
:type resources: JSON string
@@ -290,6 +248,7 @@ class SKASubarrayResourceManager:
:todo: Currently implemented for testing purposes to take a JSON
string encoding a dictionary with key 'example'. In future this
will take a collection of resources.
+
:param resources: JSON-encoding of a dictionary, with resources to
assign under key 'example'
:type resources: JSON string
@@ -369,8 +328,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel instance
:param action_hook: a hook for the command, used to build
actions that will be sent to the state model; for example,
if the hook is "scan", then success of the command will
@@ -418,8 +376,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel instance
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -433,6 +390,7 @@ class SKASubarray(SKAObsDevice):
:param argin: The resources to be assigned
:type argin: list of str
+
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
@@ -460,8 +418,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel instance
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -475,6 +432,7 @@ class SKASubarray(SKAObsDevice):
:param argin: The resources to be released
:type argin: list of str
+
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
@@ -527,8 +485,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel instance
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -544,6 +501,7 @@ class SKASubarray(SKAObsDevice):
:param argin: The configuration as JSON
:type argin: str
+
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
@@ -582,8 +540,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -599,6 +556,7 @@ class SKASubarray(SKAObsDevice):
:param argin: Scan info
:type argin: str
+
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
@@ -624,8 +582,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -661,8 +618,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -701,8 +657,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -740,8 +695,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -787,8 +741,7 @@ class SKASubarray(SKAObsDevice):
:param state_model: the state model that this command uses
to check that it is allowed to run, and that it drives
with actions.
- :type state_model: SKABaseClassStateModel or a subclass of
- same
+ :type state_model: SKASubarrayStateModel
:param logger: the logger to be used by this Command. If not
provided, then a default module logger will be used.
:type logger: a logger that implements the standard library
@@ -828,7 +781,8 @@ class SKASubarray(SKAObsDevice):
Sets up the state model for the device
"""
self.state_model = SKASubarrayStateModel(
- dev_state_callback=self._update_state,
+ logger=self.logger,
+ op_state_callback=self._update_state,
admin_mode_callback=self._update_admin_mode,
obs_state_callback=self._update_obs_state
)
@@ -885,6 +839,7 @@ class SKASubarray(SKAObsDevice):
:param capability_types: a list strings representing
capability types.
:type capability_types: list
+
:raises ValueError: If any of the capabilities requested are
not valid.
"""
@@ -946,11 +901,21 @@ class SKASubarray(SKAObsDevice):
# ---------------
def always_executed_hook(self):
# PROTECTED REGION ID(SKASubarray.always_executed_hook) ENABLED START #
+ """
+ Method that is always executed before any device command gets executed.
+
+ :return: None
+ """
pass
# PROTECTED REGION END # // SKASubarray.always_executed_hook
def delete_device(self):
# PROTECTED REGION ID(SKASubarray.delete_device) ENABLED START #
+ """
+ Method to cleanup when device is stopped.
+
+ :return: None
+ """
pass
# PROTECTED REGION END # // SKASubarray.delete_device
@@ -1003,6 +968,7 @@ class SKASubarray(SKAObsDevice):
device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1036,6 +1002,7 @@ class SKASubarray(SKAObsDevice):
device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1069,6 +1036,7 @@ class SKASubarray(SKAObsDevice):
device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1100,6 +1068,7 @@ class SKASubarray(SKAObsDevice):
device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1132,6 +1101,7 @@ class SKASubarray(SKAObsDevice):
Check if command `Scan` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1164,6 +1134,7 @@ class SKASubarray(SKAObsDevice):
Check if command `EndScan` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1191,6 +1162,7 @@ class SKASubarray(SKAObsDevice):
Check if command `End` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1219,6 +1191,7 @@ class SKASubarray(SKAObsDevice):
Check if command `Abort` is allowed in the current device state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1248,6 +1221,7 @@ class SKASubarray(SKAObsDevice):
state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
@@ -1276,6 +1250,7 @@ class SKASubarray(SKAObsDevice):
state.
:raises ``tango.DevFailed``: if the command is not allowed
+
:return: ``True`` if the command is allowed
:rtype: boolean
"""
diff --git a/src/ska/base/utils.py b/src/ska/base/utils.py
index a1fe8ca0abce482b70152fd1ff21df306e28c44a..c5fd840835e34ee8dd913398a55b82cae271a242 100644
--- a/src/ska/base/utils.py
+++ b/src/ska/base/utils.py
@@ -1,11 +1,13 @@
"""General utilities that may be useful to SKA devices and clients."""
from builtins import str
import ast
+import functools
import inspect
import json
import pydoc
import traceback
import sys
+import warnings
from datetime import datetime
@@ -462,3 +464,26 @@ def convert_dict_to_list(dictionary):
the_list.append("{}:{}".format(key, value))
return sorted(the_list)
+
+
+def for_testing_only(func, _testing_check=lambda: 'pytest' in sys.modules):
+ """
+ A decorator that marks a function as available for testing purposes only.
+ If the decorated function is called outside of testing, a warning is raised.
+
+ Testing this decorator leads to a Godelian paradox: how to test that a
+ warning is raised when we are not testing. Monkeypatching sys.modules would
+ break everything, so instead, the condition that we evaluate to decide
+ whether we are testing or not is exposed through a `_testing_check`
+ argument, allowing for it to be replaced in testing. (The `_testing_check`
+ argument is inaccessible via the @-syntax, which is a nice bonus.)
+ """
+ @functools.wraps(func)
+ def wrapper(*args, **kwargs):
+ """
+ Function wrapper for `testing_only` decorator.
+ """
+ if not _testing_check():
+ warnings.warn(f"{func.__name__} should only be used for testing purposes.")
+ return func(*args, **kwargs)
+ return wrapper
diff --git a/tests/conftest.py b/tests/conftest.py
index 081bfde7aa24b99c8ba7d5d5827d75552ed536fb..f3299213d1964c75f26dcd1fad3b324c8f712605 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,13 +2,238 @@
A module defining a list of fixtures that are shared across all ska.base tests.
"""
import importlib
+import itertools
+import json
import pytest
from queue import Empty, Queue
+from transitions import MachineError
from tango import EventType
from tango.test_context import DeviceTestContext
+def pytest_configure(config):
+ """
+ pytest hook, used here to register custom "state_machine_tester" marks
+ """
+ config.addinivalue_line(
+ "markers",
+ "state_machine_tester: indicate that this class is state machine "
+ "tester class, and tests should be parameterised by the states and "
+ "actions in the specification provided in its argument."
+ )
+
+
+def pytest_generate_tests(metafunc):
+ """
+ pytest hook that generates tests; this hook ensures that any test
+ class that is marked with the `state_machine_tester` custom marker
+ will have its tests parameterised by the states and actions in the
+ specification provided by that mark
+ """
+ # called once per each test function
+ mark = metafunc.definition.get_closest_marker("state_machine_tester")
+ if mark:
+ spec = mark.args[0]
+ states = set()
+ triggers = set()
+ expected = {}
+
+ for (from_state, trigger, to_state) in spec:
+ states.add(from_state)
+ states.add(to_state)
+ triggers.add(trigger)
+ expected[(from_state, trigger)] = to_state
+
+ states = sorted(states)
+ triggers = sorted(triggers)
+
+ metafunc.parametrize(
+ "state_under_test, action_under_test, expected_state",
+ [
+ (
+ state,
+ trigger,
+ expected[(state, trigger)] if (state, trigger) in expected else None
+ ) for (state, trigger) in itertools.product(states, triggers)
+ ]
+ )
+
+
+class StateMachineTester:
+ """
+ Abstract base class for a class for testing state machines
+ """
+
+ def test_state_machine(
+ self, machine, state_under_test, action_under_test, expected_state,
+ ):
+ """
+ Implements the unit test for a state machine: for a given
+ initial state and an action, does execution of that action, from
+ that state, yield the expected results? If the action was
+ allowed from that state, does the machine transition to the
+ correct state? If the action was not allowed from that state,
+ does the machine reject the action (e.g. raise an exception or
+ return an error code) and remain in the current state?
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param state_under_test: the state from which the
+ `action_under_test` is being tested
+ :type state_under_test: string
+ :param action_under_test: the action being tested from the
+ `state_under_test`
+ :type action_under_test: string
+ :param expected_state: the state to which the machine is
+ expected to transition, as a result of performing the
+ `action_under_test` in the `state_under_test`. If None, then
+ the action should be disallowed and result in no change of
+ state.
+ :type expected_state: string
+
+ """
+ # Put the device into the state under test
+ self.to_state(machine, state_under_test)
+
+ # Check that we are in the state under test
+ self.assert_state(machine, state_under_test)
+
+ # Test that the action under test does what we expect it to
+ if expected_state is None:
+ # Action should fail and the state should not change
+ self.check_action_disallowed(machine, action_under_test)
+ self.assert_state(machine, state_under_test)
+ else:
+ # Action should succeed
+ self.perform_action(machine, action_under_test)
+ self.assert_state(machine, expected_state)
+
+ def assert_state(self, machine, state):
+ """
+ Abstract method for asserting the current state of the state
+ machine under test
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param state: the state that we are asserting to be the current
+ state of the state machine under test
+ :type state: string
+ """
+ raise NotImplementedError()
+
+ def perform_action(self, machine, action):
+ """
+ Abstract method for performing an action on the state machine
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ raise NotImplementedError()
+
+ def check_action_disallowed(self, machine, action):
+ """
+ Abstract method for asserting that an action fails if performed
+ on the state machine under test in its current state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ raise NotImplementedError()
+
+ def to_state(self, machine, target_state):
+ """
+ Abstract method for getting the state machine into a target
+ state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param target_state: the state that we want to get the state
+ machine under test into
+ :type target_state: str
+ """
+ raise NotImplementedError()
+
+
+class TransitionsStateMachineTester(StateMachineTester):
+ """
+ Concrete implementation of a StateMachineTester for a pytransitions
+ state machine (with autotransitions turned on). The states and
+ actions in the state machine specification must correspond exactly
+ with the machine's states and triggers.
+ """
+
+ def assert_state(self, machine, state):
+ """
+ Assert the current state of the state machine under test.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param state: the state that we are asserting to be the current
+ state of the state machine under test
+ :type state: str
+ """
+ assert machine.state == state
+
+ def perform_action(self, machine, action):
+ """
+ Perform a given action on the state machine under test.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ machine.trigger(action)
+
+ def check_action_disallowed(self, machine, action):
+ """
+ Assert that performing a given action on the state maching under
+ test fails in its current state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ with pytest.raises(MachineError):
+ self.perform_action(machine, action)
+
+ def to_state(self, machine, target_state):
+ """
+ Transition the state machine to a target state. This
+ implementation uses autotransitions. If the pytransitions state
+ machine under test has autotransitions turned off, then this
+ method will need to be overridden by some other method of
+ putting the machine into the state under test.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param target_state: the state that we want to get the state
+ machine under test into
+ :type target_state: str
+ """
+ machine.trigger(f"to_{target_state}")
+
+
+def load_data(name):
+ """
+ Loads a dataset by name. This implementation uses the name to find a
+ JSON file containing the data to be loaded.
+
+ :param name: name of the dataset to be loaded; this implementation
+ uses the name to find a JSON file containing the data to be
+ loaded.
+ :type name: string
+ """
+ with open(f"tests/data/{name}.json", "r") as json_file:
+ return json.load(json_file)
+
+
@pytest.fixture(scope="class")
def tango_context(request):
"""Creates and returns a TANGO DeviceTestContext object.
@@ -95,6 +320,11 @@ def tango_change_event_helper(tango_context):
"""
class _Callback:
+ """
+ Private callback handler class, an instance of which is returned
+ by the tango_change_event_helper each time it is used to
+ subscribe to a change event.
+ """
@staticmethod
def subscribe(attribute_name):
"""
diff --git a/tests/data/base_device_state_machine.json b/tests/data/base_device_state_machine.json
new file mode 100644
index 0000000000000000000000000000000000000000..a9891a842b798cb6c0ceff7fbd0c129ba9e42dd5
--- /dev/null
+++ b/tests/data/base_device_state_machine.json
@@ -0,0 +1,222 @@
+[
+ [
+ "UNINITIALISED",
+ "init_started",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_notfitted",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_offline",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_online",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_maintenance",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "init_succeeded",
+ "OFF"
+ ],
+ [
+ "INIT_ENABLED",
+ "init_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_notfitted",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_offline",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_online",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_maintenance",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "init_succeeded",
+ "DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "init_failed",
+ "FAULT_DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "fatal_error",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_notfitted",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_offline",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_online",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_maintenance",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "reset_succeeded",
+ "DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "reset_failed",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "fatal_error",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_notfitted",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_offline",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_online",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_maintenance",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "reset_succeeded",
+ "OFF"
+ ],
+ [
+ "FAULT_ENABLED",
+ "reset_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ],
+ [
+ "DISABLED",
+ "to_notfitted",
+ "DISABLED"
+ ],
+ [
+ "DISABLED",
+ "to_offline",
+ "DISABLED"
+ ],
+ [
+ "DISABLED",
+ "to_online",
+ "OFF"
+ ],
+ [
+ "DISABLED",
+ "to_maintenance",
+ "OFF"
+ ],
+ [
+ "DISABLED",
+ "fatal_error",
+ "FAULT_DISABLED"
+ ],
+ [
+ "OFF",
+ "to_notfitted",
+ "DISABLED"
+ ],
+ [
+ "OFF",
+ "to_offline",
+ "DISABLED"
+ ],
+ [
+ "OFF",
+ "to_online",
+ "OFF"
+ ],
+ [
+ "OFF",
+ "to_maintenance",
+ "OFF"
+ ],
+ [
+ "OFF",
+ "on_succeeded",
+ "ON"
+ ],
+ [
+ "OFF",
+ "on_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "OFF",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ],
+ [
+ "ON",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "ON",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "ON",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ]
+]
\ No newline at end of file
diff --git a/tests/data/observation_state_machine.json b/tests/data/observation_state_machine.json
new file mode 100644
index 0000000000000000000000000000000000000000..4165bae0b69ed7e85936e133b050d3b79f93b8ad
--- /dev/null
+++ b/tests/data/observation_state_machine.json
@@ -0,0 +1,212 @@
+[
+ [
+ "EMPTY",
+ "assign_started",
+ "RESOURCING"
+ ],
+ [
+ "EMPTY",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "RESOURCING",
+ "resourcing_succeeded_some_resources",
+ "IDLE"
+ ],
+ [
+ "RESOURCING",
+ "resourcing_succeeded_no_resources",
+ "EMPTY"
+ ],
+ [
+ "RESOURCING",
+ "resourcing_failed",
+ "FAULT"
+ ],
+ [
+ "RESOURCING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "IDLE",
+ "assign_started",
+ "RESOURCING"
+ ],
+ [
+ "IDLE",
+ "release_started",
+ "RESOURCING"
+ ],
+ [
+ "IDLE",
+ "configure_started",
+ "CONFIGURING"
+ ],
+ [
+ "IDLE",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "IDLE",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "CONFIGURING",
+ "configure_succeeded",
+ "READY"
+ ],
+ [
+ "CONFIGURING",
+ "configure_failed",
+ "FAULT"
+ ],
+ [
+ "CONFIGURING",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "CONFIGURING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "READY",
+ "end_succeeded",
+ "IDLE"
+ ],
+ [
+ "READY",
+ "end_failed",
+ "FAULT"
+ ],
+ [
+ "READY",
+ "configure_started",
+ "CONFIGURING"
+ ],
+ [
+ "READY",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "READY",
+ "scan_started",
+ "SCANNING"
+ ],
+ [
+ "READY",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "SCANNING",
+ "scan_succeeded",
+ "READY"
+ ],
+ [
+ "SCANNING",
+ "scan_failed",
+ "FAULT"
+ ],
+ [
+ "SCANNING",
+ "end_scan_succeeded",
+ "READY"
+ ],
+ [
+ "SCANNING",
+ "end_scan_failed",
+ "FAULT"
+ ],
+ [
+ "SCANNING",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "SCANNING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "ABORTING",
+ "abort_succeeded",
+ "ABORTED"
+ ],
+ [
+ "ABORTING",
+ "abort_failed",
+ "FAULT"
+ ],
+ [
+ "ABORTING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "ABORTED",
+ "obs_reset_started",
+ "RESETTING"
+ ],
+ [
+ "ABORTED",
+ "restart_started",
+ "RESTARTING"
+ ],
+ [
+ "ABORTED",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "FAULT",
+ "obs_reset_started",
+ "RESETTING"
+ ],
+ [
+ "FAULT",
+ "restart_started",
+ "RESTARTING"
+ ],
+ [
+ "FAULT",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "RESETTING",
+ "obs_reset_succeeded",
+ "IDLE"
+ ],
+ [
+ "RESETTING",
+ "obs_reset_failed",
+ "FAULT"
+ ],
+ [
+ "RESETTING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "RESTARTING",
+ "restart_succeeded",
+ "EMPTY"
+ ],
+ [
+ "RESTARTING",
+ "restart_failed",
+ "FAULT"
+ ],
+ [
+ "RESTARTING",
+ "fatal_error",
+ "FAULT"
+ ]
+]
\ No newline at end of file
diff --git a/tests/data/subarray_state_machine.json b/tests/data/subarray_state_machine.json
new file mode 100644
index 0000000000000000000000000000000000000000..2c416dcd8dbb6a5732952f7ba2fc38e9fa51a3fd
--- /dev/null
+++ b/tests/data/subarray_state_machine.json
@@ -0,0 +1,527 @@
+[
+ [
+ "UNINITIALISED",
+ "init_started",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_notfitted",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_offline",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_online",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "to_maintenance",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "init_succeeded",
+ "OFF"
+ ],
+ [
+ "INIT_ENABLED",
+ "init_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "INIT_ENABLED",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_notfitted",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_offline",
+ "INIT_DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_online",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "to_maintenance",
+ "INIT_ENABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "init_succeeded",
+ "DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "init_failed",
+ "FAULT_DISABLED"
+ ],
+ [
+ "INIT_DISABLED",
+ "fatal_error",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_notfitted",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_offline",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_online",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "to_maintenance",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "reset_succeeded",
+ "DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "reset_failed",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_DISABLED",
+ "fatal_error",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_notfitted",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_offline",
+ "FAULT_DISABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_online",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "to_maintenance",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "reset_succeeded",
+ "OFF"
+ ],
+ [
+ "FAULT_ENABLED",
+ "reset_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT_ENABLED",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ],
+ [
+ "DISABLED",
+ "to_notfitted",
+ "DISABLED"
+ ],
+ [
+ "DISABLED",
+ "to_offline",
+ "DISABLED"
+ ],
+ [
+ "DISABLED",
+ "to_online",
+ "OFF"
+ ],
+ [
+ "DISABLED",
+ "to_maintenance",
+ "OFF"
+ ],
+ [
+ "DISABLED",
+ "fatal_error",
+ "FAULT_DISABLED"
+ ],
+ [
+ "OFF",
+ "to_notfitted",
+ "DISABLED"
+ ],
+ [
+ "OFF",
+ "to_offline",
+ "DISABLED"
+ ],
+ [
+ "OFF",
+ "to_online",
+ "OFF"
+ ],
+ [
+ "OFF",
+ "to_maintenance",
+ "OFF"
+ ],
+ [
+ "OFF",
+ "on_succeeded",
+ "EMPTY"
+ ],
+ [
+ "OFF",
+ "on_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "OFF",
+ "fatal_error",
+ "FAULT_ENABLED"
+ ],
+ [
+ "EMPTY",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "EMPTY",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "EMPTY",
+ "assign_started",
+ "RESOURCING"
+ ],
+ [
+ "EMPTY",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "RESOURCING",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "RESOURCING",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "RESOURCING",
+ "resourcing_succeeded_some_resources",
+ "IDLE"
+ ],
+ [
+ "RESOURCING",
+ "resourcing_succeeded_no_resources",
+ "EMPTY"
+ ],
+ [
+ "RESOURCING",
+ "resourcing_failed",
+ "FAULT"
+ ],
+ [
+ "RESOURCING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "IDLE",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "IDLE",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "IDLE",
+ "assign_started",
+ "RESOURCING"
+ ],
+ [
+ "IDLE",
+ "release_started",
+ "RESOURCING"
+ ],
+ [
+ "IDLE",
+ "configure_started",
+ "CONFIGURING"
+ ],
+ [
+ "IDLE",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "IDLE",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "CONFIGURING",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "CONFIGURING",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "CONFIGURING",
+ "configure_succeeded",
+ "READY"
+ ],
+ [
+ "CONFIGURING",
+ "configure_failed",
+ "FAULT"
+ ],
+ [
+ "CONFIGURING",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "CONFIGURING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "READY",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "READY",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "READY",
+ "end_succeeded",
+ "IDLE"
+ ],
+ [
+ "READY",
+ "end_failed",
+ "FAULT"
+ ],
+ [
+ "READY",
+ "configure_started",
+ "CONFIGURING"
+ ],
+ [
+ "READY",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "READY",
+ "scan_started",
+ "SCANNING"
+ ],
+ [
+ "READY",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "SCANNING",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "SCANNING",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "SCANNING",
+ "scan_succeeded",
+ "READY"
+ ],
+ [
+ "SCANNING",
+ "scan_failed",
+ "FAULT"
+ ],
+ [
+ "SCANNING",
+ "end_scan_succeeded",
+ "READY"
+ ],
+ [
+ "SCANNING",
+ "end_scan_failed",
+ "FAULT"
+ ],
+ [
+ "SCANNING",
+ "abort_started",
+ "ABORTING"
+ ],
+ [
+ "SCANNING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "ABORTING",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "ABORTING",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "ABORTING",
+ "abort_succeeded",
+ "ABORTED"
+ ],
+ [
+ "ABORTING",
+ "abort_failed",
+ "FAULT"
+ ],
+ [
+ "ABORTING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "ABORTED",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "ABORTED",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "ABORTED",
+ "obs_reset_started",
+ "RESETTING"
+ ],
+ [
+ "ABORTED",
+ "restart_started",
+ "RESTARTING"
+ ],
+ [
+ "ABORTED",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "FAULT",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "FAULT",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "FAULT",
+ "obs_reset_started",
+ "RESETTING"
+ ],
+ [
+ "FAULT",
+ "restart_started",
+ "RESTARTING"
+ ],
+ [
+ "FAULT",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "RESETTING",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "RESETTING",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "RESETTING",
+ "obs_reset_succeeded",
+ "IDLE"
+ ],
+ [
+ "RESETTING",
+ "obs_reset_failed",
+ "FAULT"
+ ],
+ [
+ "RESETTING",
+ "fatal_error",
+ "FAULT"
+ ],
+ [
+ "RESTARTING",
+ "off_succeeded",
+ "OFF"
+ ],
+ [
+ "RESTARTING",
+ "off_failed",
+ "FAULT_ENABLED"
+ ],
+ [
+ "RESTARTING",
+ "restart_succeeded",
+ "EMPTY"
+ ],
+ [
+ "RESTARTING",
+ "restart_failed",
+ "FAULT"
+ ],
+ [
+ "RESTARTING",
+ "fatal_error",
+ "FAULT"
+ ]
+]
\ No newline at end of file
diff --git a/tests/test_base_device.py b/tests/test_base_device.py
index dfe59795963417f50af18cfea014b32fffdd837d..be517b0e744fcbbd5beb528e24f82587dd533386 100644
--- a/tests/test_base_device.py
+++ b/tests/test_base_device.py
@@ -6,20 +6,19 @@
#
#
#########################################################################################
-"""Contain the tests for the SKABASE."""
-
-import itertools
-import re
-import pytest
+"""
+This module contains the tests for the SKABaseDevice.
+"""
# PROTECTED REGION ID(SKABaseDevice.test_additional_imports) ENABLED START #
import logging
+import re
+import pytest
import socket
import tango
from unittest import mock
from tango import DevFailed, DevState
-from ska.base import SKABaseDeviceStateModel
from ska.base.control_model import (
AdminMode, ControlMode, HealthState, LoggingLevel, SimulationMode, TestMode
)
@@ -28,10 +27,13 @@ from ska.base.base_device import (
_PYTHON_TO_TANGO_LOGGING_LEVEL,
LoggingUtils,
LoggingTargetError,
+ SKABaseDeviceStateModel,
TangoLoggingServiceHandler,
)
from ska.base.faults import StateModelError
+from .conftest import load_data, StateMachineTester
+
# PROTECTED REGION END # // SKABaseDevice.test_additional_imports
# Device test case
# PROTECTED REGION ID(SKABaseDevice.test_SKABaseDevice_decorators) ENABLED START #
@@ -321,7 +323,103 @@ class TestLoggingUtils:
LoggingUtils.create_logging_handler = orig_create_logging_handler
-@pytest.mark.usefixtures("tango_context", "initialize_device")
+@pytest.fixture
+def device_state_model():
+ """
+ Yields a new SKABaseDeviceStateModel for testing
+ """
+ yield SKABaseDeviceStateModel(logging.getLogger())
+
+
+@pytest.mark.state_machine_tester(load_data("base_device_state_machine"))
+class TestSKABaseDeviceStateModel(StateMachineTester):
+ """
+ This class contains the test suite for the ska.base.SKABaseDevice class.
+ """
+ @pytest.fixture
+ def machine(self, device_state_model):
+ """
+ Fixture that returns the state machine under test in this class
+ """
+ yield device_state_model
+
+ state_checks = {
+ "UNINITIALISED":
+ (None, None),
+ "FAULT_ENABLED":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.FAULT),
+ "FAULT_DISABLED":
+ ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.FAULT),
+ "INIT_ENABLED":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.INIT),
+ "INIT_DISABLED":
+ ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.INIT),
+ "DISABLED":
+ ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.DISABLE),
+ "OFF":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.OFF),
+ "ON":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON),
+ }
+
+ def assert_state(self, machine, state):
+ """
+ Assert the current state of this state machine, based on the
+ values of the adminMode, opState and obsState attributes of this
+ model.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param state: the state that we are asserting to be the current
+ state of the state machine under test
+ :type state: str
+ """
+
+ (admin_modes, op_state) = self.state_checks[state]
+ if admin_modes is None:
+ assert machine.admin_mode is None
+ else:
+ assert machine.admin_mode in admin_modes
+ if op_state is None:
+ assert machine.op_state is None
+ else:
+ assert machine.op_state == op_state
+
+ def perform_action(self, machine, action):
+ """
+ Perform a given action on the state machine under test.
+
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ machine.perform_action(action)
+
+ def check_action_disallowed(self, machine, action):
+ """
+ Assert that performing a given action on the state maching under
+ test fails in its current state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ with pytest.raises(StateModelError):
+ self.perform_action(machine, action)
+
+ def to_state(self, machine, target_state):
+ """
+ Transition the state machine to a target state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param target_state: the state that we want to get the state
+ machine under test into
+ :type target_state: str
+ """
+ machine._straight_to_state(target_state)
+
+
# PROTECTED REGION END # // SKABaseDevice.test_SKABaseDevice_decorators
class TestSKABaseDevice(object):
"""
@@ -587,145 +685,3 @@ class TestSKABaseDevice(object):
# PROTECTED REGION ID(SKABaseDevice.test_testMode) ENABLED START #
assert tango_context.device.testMode == TestMode.NONE
# PROTECTED REGION END # // SKABaseDevice.test_testMode
-
-
-@pytest.fixture
-def state_model():
- yield SKABaseDeviceStateModel()
-
-
-class TestSKABaseDeviceStateModel():
- """
- Test cases for SKABaseDeviceStateModel.
- """
-
- @pytest.mark.parametrize(
- 'state_under_test, action_under_test',
- itertools.product(
- ["UNINITIALISED", "INIT_ENABLED", "INIT_DISABLED", "FAULT_ENABLED",
- "FAULT_DISABLED", "DISABLED", "OFF", "ON"],
- ["init_started", "init_succeeded", "init_failed", "fatal_error",
- "reset_succeeded", "reset_failed", "to_notfitted",
- "to_offline", "to_online", "to_maintenance", "on_succeeded",
- "on_failed", "off_succeeded", "off_failed"]
- )
- )
- def test_state_machine(
- self, state_model, state_under_test, action_under_test
- ):
- """
- Test the subarray state machine: for a given initial state and
- an action, does execution of that action, from that initial
- state, yield the expected results? If the action was not allowed
- from that initial state, does the device raise a DevFailed
- exception? If the action was allowed, does it result in the
- correct state transition?
-
- :todo: support starting in different memorised adminModes
- """
- states = {
- "UNINITIALISED":
- (None, None),
- "FAULT_ENABLED":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.FAULT),
- "FAULT_DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.FAULT),
- "INIT_ENABLED":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.INIT),
- "INIT_DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.INIT),
- "DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.DISABLE),
- "OFF":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.OFF),
- "ON":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON),
- }
-
- def assert_state(state):
- (admin_modes, state) = states[state]
- if admin_modes is not None:
- assert state_model.admin_mode in admin_modes
- if state is not None:
- assert state_model.dev_state == state
-
- transitions = {
- ('UNINITIALISED', 'init_started'): "INIT_ENABLED",
- ('INIT_ENABLED', 'to_notfitted'): "INIT_DISABLED",
- ('INIT_ENABLED', 'to_offline'): "INIT_DISABLED",
- ('INIT_ENABLED', 'to_online'): "INIT_ENABLED",
- ('INIT_ENABLED', 'to_maintenance'): "INIT_ENABLED",
- ('INIT_ENABLED', 'init_succeeded'): 'OFF',
- ('INIT_ENABLED', 'init_failed'): 'FAULT_ENABLED',
- ('INIT_ENABLED', 'fatal_error'): "FAULT_ENABLED",
- ('INIT_DISABLED', 'to_notfitted'): "INIT_DISABLED",
- ('INIT_DISABLED', 'to_offline'): "INIT_DISABLED",
- ('INIT_DISABLED', 'to_online'): "INIT_ENABLED",
- ('INIT_DISABLED', 'to_maintenance'): "INIT_ENABLED",
- ('INIT_DISABLED', 'init_succeeded'): 'DISABLED',
- ('INIT_DISABLED', 'init_failed'): 'FAULT_DISABLED',
- ('INIT_DISABLED', 'fatal_error'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'to_notfitted'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'to_offline'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'to_online'): "FAULT_ENABLED",
- ('FAULT_DISABLED', 'to_maintenance'): "FAULT_ENABLED",
- ('FAULT_DISABLED', 'reset_succeeded'): "DISABLED",
- ('FAULT_DISABLED', 'reset_failed'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'fatal_error'): "FAULT_DISABLED",
- ('FAULT_ENABLED', 'to_notfitted'): "FAULT_DISABLED",
- ('FAULT_ENABLED', 'to_offline'): "FAULT_DISABLED",
- ('FAULT_ENABLED', 'to_online'): "FAULT_ENABLED",
- ('FAULT_ENABLED', 'to_maintenance'): "FAULT_ENABLED",
- ('FAULT_ENABLED', 'reset_succeeded'): "OFF",
- ('FAULT_ENABLED', 'reset_failed'): "FAULT_ENABLED",
- ('FAULT_ENABLED', 'fatal_error'): "FAULT_ENABLED",
- ('DISABLED', 'to_notfitted'): "DISABLED",
- ('DISABLED', 'to_offline'): "DISABLED",
- ('DISABLED', 'to_online'): "OFF",
- ('DISABLED', 'to_maintenance'): "OFF",
- ('DISABLED', 'fatal_error'): "FAULT_DISABLED",
- ('OFF', 'to_notfitted'): "DISABLED",
- ('OFF', 'to_offline'): "DISABLED",
- ('OFF', 'to_online'): "OFF",
- ('OFF', 'to_maintenance'): "OFF",
- ('OFF', 'on_succeeded'): "ON",
- ('OFF', 'on_failed'): "FAULT_ENABLED",
- ('OFF', 'fatal_error'): "FAULT_ENABLED",
- ('ON', 'off_succeeded'): "OFF",
- ('ON', 'off_failed'): "FAULT_ENABLED",
- ('ON', 'fatal_error'): "FAULT_ENABLED",
- }
-
- setups = {
- "UNINITIALISED": [],
- "INIT_ENABLED": ['init_started'],
- "INIT_DISABLED": ['init_started', 'to_offline'],
- "FAULT_ENABLED": ['init_started', 'init_failed'],
- "FAULT_DISABLED": ['init_started', 'to_offline', 'init_failed'],
- "OFF": ['init_started', 'init_succeeded'],
- "DISABLED": ['init_started', 'init_succeeded', 'to_offline'],
- "ON": ['init_started', 'init_succeeded', 'on_succeeded'],
- }
-
- # state = "UNINITIALISED" # for test debugging only
- # assert_state(state) # for test debugging only
-
- # Put the device into the state under test
- for action in setups[state_under_test]:
- state_model.perform_action(action)
- # state = transitions[state, action] # for test debugging only
- # assert_state(state) # for test debugging only
-
- # Check that we are in the state under test
- assert_state(state_under_test)
-
- # Test that the action under test does what we expect it to
- if (state_under_test, action_under_test) in transitions:
- # Action should succeed
- state_model.perform_action(action_under_test)
- assert_state(transitions[(state_under_test, action_under_test)])
- else:
- # Action should fail and the state should not change
- with pytest.raises(StateModelError):
- state_model.perform_action(action_under_test)
- assert_state(state_under_test)
diff --git a/tests/test_control_model.py b/tests/test_control_model.py
deleted file mode 100644
index 130077fd7e3ef059e5a210e5ab517ddb81007f63..0000000000000000000000000000000000000000
--- a/tests/test_control_model.py
+++ /dev/null
@@ -1,89 +0,0 @@
-"""Tests for ska.base.control_model."""
-import pytest
-
-from collections import Counter
-
-from ska.base.control_model import DeviceStateModel
-from ska.base.faults import StateModelError
-
-
-class OnOffModel(DeviceStateModel):
- __transitions = {
- ("ON", "off"): (
- "OFF",
- lambda self: self.count_transition("on->off"),
- ),
- ("OFF", "on"): (
- "ON",
- lambda self: self.count_transition("off->on"),
- ),
- }
-
- def __init__(self):
- super().__init__(self.__transitions, "ON")
- self.counter = Counter()
-
- def count_transition(self, name):
- self.counter[name] += 1
-
-
-def test_initial_state():
- on_off = OnOffModel()
- assert on_off.state == "ON"
-
-
-def test_try_valid_action_returns_true_and_does_not_change_state():
- on_off = OnOffModel()
- assert on_off.try_action("off")
- assert on_off.state == "ON"
-
-
-def test_try_invalid_action_raises_and_does_not_change_state():
- on_off = OnOffModel()
- with pytest.raises(StateModelError):
- on_off.try_action("on")
- assert on_off.state == "ON"
-
-
-def test_valid_state_transitions_succeed():
- on_off = OnOffModel()
-
- on_off.perform_action("off")
- assert on_off.state == "OFF"
- on_off.perform_action("on")
- assert on_off.state == "ON"
-
-
-def test_invalid_state_transitions_fail():
- on_off = OnOffModel()
-
- with pytest.raises(StateModelError):
- on_off.perform_action("on")
- on_off.perform_action("off")
- with pytest.raises(StateModelError):
- on_off.perform_action("off")
- with pytest.raises(StateModelError):
- on_off.perform_action("invalid")
-
-
-def test_side_effect_is_called():
- on_off = OnOffModel()
-
- on_off.perform_action("off")
- on_off.perform_action("on")
- on_off.perform_action("off")
-
- assert on_off.counter["on->off"] == 2
- assert on_off.counter["off->on"] == 1
-
-
-def test_update_transitions_only_applies_to_instances():
- on_off = OnOffModel()
-
- on_off_updated = OnOffModel()
- on_off_updated.update_transitions({("ON", "break"): ("BROKEN", None)})
-
- assert on_off.is_action_allowed("off")
- assert not on_off.is_action_allowed("break")
- assert on_off_updated.is_action_allowed("off")
- assert on_off_updated.is_action_allowed("break")
diff --git a/tests/test_state_machines.py b/tests/test_state_machines.py
new file mode 100644
index 0000000000000000000000000000000000000000..d44b641b982f856ed419691b5c16c4ee9ba28a94
--- /dev/null
+++ b/tests/test_state_machines.py
@@ -0,0 +1,33 @@
+"""
+This module contains the tests for the ska.base.state_machine module.
+"""
+import pytest
+
+from ska.base.state_machine import BaseDeviceStateMachine, ObservationStateMachine
+from .conftest import load_data, TransitionsStateMachineTester
+
+
+@pytest.mark.state_machine_tester(load_data("base_device_state_machine"))
+class BaseDeviceStateMachineTester(TransitionsStateMachineTester):
+ """
+ This class contains the test for the BaseDeviceStateMachine class.
+ """
+ @pytest.fixture
+ def machine(self):
+ """
+ Fixture that returns the state machine under test in this class
+ """
+ yield BaseDeviceStateMachine()
+
+
+@pytest.mark.state_machine_tester(load_data("observation_state_machine"))
+class TestObservationStateMachine(TransitionsStateMachineTester):
+ """
+ This class contains the test for the ObservationStateMachine class.
+ """
+ @pytest.fixture
+ def machine(self):
+ """
+ Fixture that returns the state machine under test in this class
+ """
+ yield ObservationStateMachine()
diff --git a/tests/test_subarray_device.py b/tests/test_subarray_device.py
index ae4a8cade5bc4f254d7ef459aa40ab95d93c1e4b..7f91bb27e9bd4db38f136af1ef1d3aa19cfdfa1d 100644
--- a/tests/test_subarray_device.py
+++ b/tests/test_subarray_device.py
@@ -8,7 +8,7 @@
#########################################################################################
"""Contain the tests for the SKASubarray."""
-import itertools
+import logging
import re
import pytest
@@ -20,13 +20,142 @@ from ska.base.commands import ResultCode
from ska.base.control_model import (
AdminMode, ControlMode, HealthState, ObsMode, ObsState, SimulationMode, TestMode
)
-from ska.base.faults import CommandError
+from ska.base.faults import CommandError, StateModelError
+
+from .conftest import load_data, StateMachineTester
# PROTECTED REGION END # // SKASubarray.test_additional_imports
-@pytest.mark.usefixtures("tango_context", "initialize_device")
-class TestSKASubarray(object):
- """Test case for packet generation."""
+@pytest.fixture
+def subarray_state_model():
+ """
+ Yields a new SKASubarrayStateModel for testing
+ """
+ yield SKASubarrayStateModel(logging.getLogger())
+
+
+@pytest.mark.state_machine_tester(load_data("subarray_state_machine"))
+class TestSKASubarrayStateModel(StateMachineTester):
+ """
+ This class contains the test for the SKASubarrayStateModel class.
+ """
+ @pytest.fixture
+ def machine(self, subarray_state_model):
+ """
+ Fixture that returns the state machine under test in this class
+ """
+ yield subarray_state_model
+
+ state_checks = {
+ "UNINITIALISED":
+ (None, None, ObsState.EMPTY),
+ "FAULT_ENABLED":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.FAULT, ObsState.EMPTY),
+ "FAULT_DISABLED":
+ ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.FAULT, ObsState.EMPTY),
+ "INIT_ENABLED":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.INIT, ObsState.EMPTY),
+ "INIT_DISABLED":
+ ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.INIT, ObsState.EMPTY),
+ "DISABLED":
+ ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.DISABLE, ObsState.EMPTY),
+ "OFF":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.OFF, ObsState.EMPTY),
+ "EMPTY":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.EMPTY),
+ "RESOURCING":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.RESOURCING),
+ "IDLE":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.IDLE),
+ "CONFIGURING":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.CONFIGURING),
+ "READY":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.READY),
+ "SCANNING":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.SCANNING),
+ "ABORTING":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.ABORTING),
+ "ABORTED":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.ABORTED),
+ "FAULT":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.FAULT),
+ "RESETTING":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.RESETTING),
+ "RESTARTING":
+ ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.RESTARTING),
+
+ }
+
+ def assert_state(self, machine, state):
+ """
+ Assert the current state of this state machine, based on the
+ values of the adminMode, opState and obsState attributes of this
+ model.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param state: the state that we are asserting to be the current
+ state of the state machine under test
+ :type state: str
+ """
+ # Debugging only -- machine is already tested
+ # assert self.model._state == state
+ # print(f"State is {state}")
+ (admin_modes, op_state, obs_state) = self.state_checks[state]
+ if admin_modes is None:
+ assert machine.admin_mode is None
+ else:
+ assert machine.admin_mode in admin_modes
+ if op_state is None:
+ assert machine.op_state is None
+ else:
+ assert machine.op_state == op_state
+ if obs_state is None:
+ assert machine.obs_state is None
+ else:
+ assert machine.obs_state == obs_state
+
+ def perform_action(self, machine, action):
+ """
+ Perform a given action on the state machine under test.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ machine.perform_action(action)
+
+ def check_action_disallowed(self, machine, action):
+ """
+ Assert that performing a given action on the state maching under
+ test fails in its current state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param action: action to be performed on the state machine
+ :type action: str
+ """
+ with pytest.raises(StateModelError):
+ self.perform_action(machine, action)
+
+ def to_state(self, machine, target_state):
+ """
+ Transition the state machine to a target state.
+
+ :param machine: the state machine under test
+ :type machine: state machine object instance
+ :param target_state: the state that we want to get the state
+ machine under test into
+ :type target_state: str
+ """
+ machine._straight_to_state(target_state)
+
+
+class TestSKASubarray:
+ """
+ Test cases for SKASubarray device.
+ """
properties = {
'CapabilityTypes': '',
@@ -44,11 +173,12 @@ class TestSKASubarray(object):
# PROTECTED REGION ID(SKASubarray.test_mocking) ENABLED START #
# PROTECTED REGION END # // SKASubarray.test_mocking
+ @pytest.mark.skip(reason="Not implemented")
def test_properties(self, tango_context):
# Test the properties
# PROTECTED REGION ID(SKASubarray.test_properties) ENABLED START #
# PROTECTED REGION END # // SKASubarray.test_properties
- pass
+ """Test the Tango device properties of this subarray device"""
# PROTECTED REGION ID(SKASubarray.test_Abort_decorators) ENABLED START #
# PROTECTED REGION END # // SKASubarray.test_Abort_decorators
@@ -403,189 +533,23 @@ class TestSKASubarray(object):
assert tango_context.device.configuredCapabilities == ("BAND1:0", "BAND2:0")
# PROTECTED REGION END # // SKASubarray.test_configuredCapabilities
- @pytest.mark.parametrize(
- 'state_under_test, action_under_test',
- itertools.product(
- [
- # not testing FAULT or OBSFAULT states because in the current
- # implementation the interface cannot be used to get the device
- # into these states
- "DISABLED", "OFF", "EMPTY", "IDLE", "READY", "SCANNING",
- "ABORTED",
- ],
- [
- # not testing 'reset' action because in the current
- # implementation the interface cannot be used to get the device
- # into a state from which 'reset' is a valid action
- "notfitted", "offline", "online", "maintenance", "on", "off",
- "assign", "release", "release (all)", "releaseall",
- "configure", "scan", "endscan", "end", "abort", "obsreset",
- "restart"]
- )
- )
- def test_state_machine(self, tango_context, state_under_test, action_under_test):
- """
- Test the subarray state machine: for a given initial state and
- an action, does execution of that action, from that initial
- state, yield the expected results? If the action was not allowed
- from that initial state, does the device raise a DevFailed
- exception? If the action was allowed, does it result in the
- correct state transition?
- """
- states = {
- "DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.DISABLE, ObsState.EMPTY),
- "FAULT": # not tested
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE, AdminMode.ONLINE, AdminMode.MAINTENANCE],
- DevState.FAULT, ObsState.EMPTY),
- "OFF":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.OFF, ObsState.EMPTY),
- "EMPTY":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.EMPTY),
- "IDLE":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.IDLE),
- "READY":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.READY),
- "SCANNING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.SCANNING),
- "ABORTED":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.ABORTED),
- "OBSFAULT": # not tested
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.FAULT),
- }
-
- def assert_state(state):
- """
- Check that the device is in the state we think it should be in
- """
- (admin_modes, dev_state, obs_state) = states[state]
- assert tango_context.device.adminMode in admin_modes
- assert tango_context.device.state() == dev_state
- assert tango_context.device.obsState == obs_state
-
- actions = {
- "notfitted":
- lambda d: d.write_attribute("adminMode", AdminMode.NOT_FITTED),
- "offline":
- lambda d: d.write_attribute("adminMode", AdminMode.OFFLINE),
- "online":
- lambda d: d.write_attribute("adminMode", AdminMode.ONLINE),
- "maintenance":
- lambda d: d.write_attribute("adminMode", AdminMode.MAINTENANCE),
- "on":
- lambda d: d.On(),
- "off":
- lambda d: d.Off(),
- "reset":
- lambda d: d.Reset(), # not tested
- "assign":
- lambda d: d.AssignResources('{"example": ["BAND1", "BAND2"]}'),
- "release":
- lambda d: d.ReleaseResources('{"example": ["BAND1"]}'),
- "release (all)":
- lambda d: d.ReleaseResources('{"example": ["BAND1", "BAND2"]}'),
- "releaseall":
- lambda d: d.ReleaseAllResources(),
- "configure":
- lambda d: d.Configure('{"BAND1": 2, "BAND2": 2}'),
- "scan":
- lambda d: d.Scan('{"id": 123}'),
- "endscan":
- lambda d: d.EndScan(),
- "end":
- lambda d: d.End(),
- "abort":
- lambda d: d.Abort(),
- "obsreset":
- lambda d: d.ObsReset(),
- "restart":
- lambda d: d.Restart(),
- }
-
- def perform_action(action):
- actions[action](tango_context.device)
-
- transitions = {
- ("DISABLED", "notfitted"): "DISABLED",
- ("DISABLED", "offline"): "DISABLED",
- ("DISABLED", "online"): "OFF",
- ("DISABLED", "maintenance"): "OFF",
- ("OFF", "notfitted"): "DISABLED",
- ("OFF", "offline"): "DISABLED",
- ("OFF", "online"): "OFF",
- ("OFF", "maintenance"): "OFF",
- ("OFF", "on"): "EMPTY",
- ("EMPTY", "off"): "OFF",
- ("EMPTY", "assign"): "IDLE",
- ("IDLE", "assign"): "IDLE",
- ("IDLE", "release"): "IDLE",
- ("IDLE", "release (all)"): "EMPTY",
- ("IDLE", "releaseall"): "EMPTY",
- ("IDLE", "configure"): "READY",
- ("IDLE", "abort"): "ABORTED",
- ("READY", "configure"): "READY",
- ("READY", "end"): "IDLE",
- ("READY", "abort"): "ABORTED",
- ("READY", "scan"): "SCANNING",
- ("SCANNING", "endscan"): "READY",
- ("SCANNING", "abort"): "ABORTED",
- ("ABORTED", "obsreset"): "IDLE",
- ("ABORTED", "restart"): "EMPTY",
- }
-
- setups = {
- "DISABLED":
- ['offline'],
- "OFF":
- [],
- "EMPTY":
- ['on'],
- "IDLE":
- ['on', 'assign'],
- "READY":
- ['on', 'assign', 'configure'],
- "SCANNING":
- ['on', 'assign', 'configure', 'scan'],
- "ABORTED":
- ['on', 'assign', 'abort'],
- }
-
- # state = "OFF" # debugging only
- # assert_state(state) # debugging only
-
- # Put the device into the state under test
- for action in setups[state_under_test]:
- perform_action(action)
- # state = transitions[state, action] # debugging only
- # assert_state(state) # debugging only
-
- # Check that we are in the state under test
- assert_state(state_under_test)
-
- # Test that the action under test does what we expect it to
- if (state_under_test, action_under_test) in transitions:
- # Action should succeed
- perform_action(action_under_test)
- assert_state(transitions[(state_under_test, action_under_test)])
- else:
- # Action should fail and the state should not change
- with pytest.raises(DevFailed):
- perform_action(action_under_test)
- assert_state(state_under_test)
-
@pytest.fixture
def resource_manager():
+ """
+ Fixture that yields an SKASubarrayResourceManager
+ """
yield SKASubarrayResourceManager()
-@pytest.fixture
-def state_model():
- yield SKASubarrayStateModel()
-
-
class TestSKASubarrayResourceManager:
+ """
+ Test suite for SKASubarrayResourceManager class
+ """
def test_ResourceManager_assign(self, resource_manager):
+ """
+ Test that the ResourceManager assigns resource correctly.
+ """
# create a resource manager and check that it is empty
assert not len(resource_manager)
assert resource_manager.get() == set()
@@ -615,6 +579,9 @@ class TestSKASubarrayResourceManager:
assert resource_manager.get() == set(["A", "B", "C", "D"])
def test_ResourceManager_release(self, resource_manager):
+ """
+ Test that the ResourceManager releases resource correctly.
+ """
resource_manager = SKASubarrayResourceManager()
resource_manager.assign('{"example": ["A", "B", "C", "D"]}')
@@ -649,34 +616,49 @@ class TestSKASubarray_commands:
This class contains tests of SKASubarray commands
"""
- def test_AssignCommand(self, resource_manager, state_model):
+ def test_AssignCommand(self, resource_manager, subarray_state_model):
"""
Test for SKASubarray.AssignResourcesCommand
"""
assign_resources = SKASubarray.AssignResourcesCommand(
resource_manager,
- state_model
+ subarray_state_model
)
- # until the state_model is in the right state for it, the
- # command's is_allowed() method will return False, and an
- # attempt to call the command will raise a CommandError, and
- # there will be no side-effect on the resource manager
- for action in ["init_started", "init_succeeded", "on_succeeded"]:
+ all_states = {
+ "UNINITIALISED", "FAULT_ENABLED", "FAULT_DISABLED", "INIT_ENABLED",
+ "INIT_DISABLED", "DISABLED", "OFF", "EMPTY", "RESOURCING", "IDLE",
+ "CONFIGURING", "READY", "SCANNING", "ABORTING", "ABORTED", "FAULT",
+ "RESETTING", "RESTARTING",
+ }
+
+ # in all states except EMPTY and IDLE, the assign resources command is
+ # not permitted, should not be allowed, should fail, should have no
+ # side-effect
+ for state in all_states - {"EMPTY", "IDLE"}:
+ subarray_state_model._straight_to_state(state)
assert not assign_resources.is_allowed()
with pytest.raises(CommandError):
assign_resources('{"example": ["foo"]}')
assert not len(resource_manager)
assert resource_manager.get() == set()
+ assert subarray_state_model._state == state
- state_model.perform_action(action)
-
- # now that the state_model is in the right state, is_allowed()
- # should return True, and the command should succeed, and we
- # should see the result in the resource manager
+ # now push to empty, a state in which is IS allowed
+ subarray_state_model._straight_to_state("EMPTY")
assert assign_resources.is_allowed()
assert assign_resources('{"example": ["foo"]}') == (
ResultCode.OK, "AssignResources command completed OK"
)
assert len(resource_manager) == 1
assert resource_manager.get() == set(["foo"])
+
+ assert subarray_state_model._state == "IDLE"
+
+ # AssignResources is still allowed in ON_IDLE
+ assert assign_resources.is_allowed()
+ assert assign_resources('{"example": ["bar"]}') == (
+ ResultCode.OK, "AssignResources command completed OK"
+ )
+ assert len(resource_manager) == 2
+ assert resource_manager.get() == set(["foo", "bar"])
diff --git a/tests/test_subarray_state_model.py b/tests/test_subarray_state_model.py
deleted file mode 100644
index 804b8cf98f39e2252297856b24ed29f645395248..0000000000000000000000000000000000000000
--- a/tests/test_subarray_state_model.py
+++ /dev/null
@@ -1,280 +0,0 @@
-#########################################################################################
-# -*- coding: utf-8 -*-
-#
-# This file is part of the SKASubarray project
-#
-#
-#
-#########################################################################################
-"""Contain the tests for the SKASubarray."""
-
-import itertools
-import pytest
-
-from tango import DevState
-
-from ska.base import SKASubarrayStateModel
-from ska.base.control_model import AdminMode, ObsState
-from ska.base.faults import StateModelError
-
-
-@pytest.fixture
-def state_model():
- yield SKASubarrayStateModel()
-
-
-class TestSKASubarrayStateModel:
- """
- Test cases for SKASubarrayStateModel.
- """
-
- @pytest.mark.parametrize(
- 'state_under_test, action_under_test',
- itertools.product(
- ["UNINITIALISED", "INIT_ENABLED", "INIT_DISABLED", "FAULT_ENABLED",
- "FAULT_DISABLED", "DISABLED", "OFF", "EMPTY",
- "RESOURCING", "IDLE", "CONFIGURING", "READY", "SCANNING",
- "ABORTING", "ABORTED", "OBSFAULT"],
- ["init_started", "init_succeeded", "init_failed", "fatal_error",
- "reset_succeeded", "reset_failed", "to_notfitted",
- "to_offline", "to_online", "to_maintenance", "on_succeeded",
- "on_failed", "off_succeeded", "off_failed", "assign_started",
- "resourcing_succeeded_no_resources", "resourcing_succeeded_some_resources",
- "resourcing_failed", "release_started", "configure_started",
- "configure_succeeded", "configure_failed", "scan_started",
- "scan_succeeded", "scan_failed", "end_scan_succeeded",
- "end_scan_failed", "abort_started", "abort_succeeded",
- "abort_failed", "obs_reset_started", "obs_reset_succeeded",
- "obs_reset_failed", "restart_started", "restart_succeeded",
- "restart_failed"]
- )
- )
- def test_state_machine(
- self, state_model, state_under_test, action_under_test
- ):
- """
- Test the subarray state machine: for a given initial state and
- an action, does execution of that action, from that initial
- state, yield the expected results? If the action was not allowed
- from that initial state, does the device raise a DevFailed
- exception? If the action was allowed, does it result in the
- correct state transition?
-
- :todo: support starting in different memorised adminModes
- """
-
- states = {
- "UNINITIALISED":
- (None, None, None),
- "FAULT_ENABLED":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.FAULT, None),
- "FAULT_DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.FAULT, None),
- "INIT_ENABLED":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.INIT, ObsState.EMPTY),
- "INIT_DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.INIT, ObsState.EMPTY),
- "DISABLED":
- ([AdminMode.NOT_FITTED, AdminMode.OFFLINE], DevState.DISABLE, ObsState.EMPTY),
- "OFF":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.OFF, ObsState.EMPTY),
- "EMPTY":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.EMPTY),
- "RESOURCING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.RESOURCING),
- "IDLE":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.IDLE),
- "CONFIGURING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.CONFIGURING),
- "READY":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.READY),
- "SCANNING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.SCANNING),
- "ABORTING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.ABORTING),
- "ABORTED":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.ABORTED),
- "RESETTING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.RESETTING),
- "RESTARTING":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.RESTARTING),
- "OBSFAULT":
- ([AdminMode.ONLINE, AdminMode.MAINTENANCE], DevState.ON, ObsState.FAULT),
- }
-
- def assert_state(state):
- (admin_modes, state, obs_state) = states[state]
- if admin_modes is not None:
- assert state_model.admin_mode in admin_modes
- if state is not None:
- assert state_model.dev_state == state
- if obs_state is not None:
- assert state_model.obs_state == obs_state
-
- transitions = {
- ('UNINITIALISED', 'init_started'): "INIT_ENABLED",
- ('INIT_ENABLED', 'to_notfitted'): "INIT_DISABLED",
- ('INIT_ENABLED', 'to_offline'): "INIT_DISABLED",
- ('INIT_ENABLED', 'to_online'): "INIT_ENABLED",
- ('INIT_ENABLED', 'to_maintenance'): "INIT_ENABLED",
- ('INIT_ENABLED', 'init_succeeded'): 'OFF',
- ('INIT_ENABLED', 'init_failed'): 'FAULT_ENABLED',
- ('INIT_ENABLED', 'fatal_error'): "FAULT_ENABLED",
- ('INIT_DISABLED', 'to_notfitted'): "INIT_DISABLED",
- ('INIT_DISABLED', 'to_offline'): "INIT_DISABLED",
- ('INIT_DISABLED', 'to_online'): "INIT_ENABLED",
- ('INIT_DISABLED', 'to_maintenance'): "INIT_ENABLED",
- ('INIT_DISABLED', 'init_succeeded'): 'DISABLED',
- ('INIT_DISABLED', 'init_failed'): 'FAULT_DISABLED',
- ('INIT_DISABLED', 'fatal_error'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'to_notfitted'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'to_offline'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'to_online'): "FAULT_ENABLED",
- ('FAULT_DISABLED', 'to_maintenance'): "FAULT_ENABLED",
- ('FAULT_DISABLED', 'reset_succeeded'): "DISABLED",
- ('FAULT_DISABLED', 'reset_failed'): "FAULT_DISABLED",
- ('FAULT_DISABLED', 'fatal_error'): "FAULT_DISABLED",
- ('FAULT_ENABLED', 'to_notfitted'): "FAULT_DISABLED",
- ('FAULT_ENABLED', 'to_offline'): "FAULT_DISABLED",
- ('FAULT_ENABLED', 'to_online'): "FAULT_ENABLED",
- ('FAULT_ENABLED', 'to_maintenance'): "FAULT_ENABLED",
- ('FAULT_ENABLED', 'reset_succeeded'): "OFF",
- ('FAULT_ENABLED', 'reset_failed'): "FAULT_ENABLED",
- ('FAULT_ENABLED', 'fatal_error'): "FAULT_ENABLED",
- ('DISABLED', 'to_notfitted'): "DISABLED",
- ('DISABLED', 'to_offline'): "DISABLED",
- ('DISABLED', 'to_online'): "OFF",
- ('DISABLED', 'to_maintenance'): "OFF",
- ('DISABLED', 'fatal_error'): "FAULT_DISABLED",
- ('OFF', 'to_notfitted'): "DISABLED",
- ('OFF', 'to_offline'): "DISABLED",
- ('OFF', 'to_online'): "OFF",
- ('OFF', 'to_maintenance'): "OFF",
- ('OFF', 'on_succeeded'): "EMPTY",
- ('OFF', 'on_failed'): "FAULT_ENABLED",
- ('OFF', 'fatal_error'): "FAULT_ENABLED",
- ('EMPTY', 'off_succeeded'): "OFF",
- ('EMPTY', 'off_failed'): "FAULT_ENABLED",
- ('EMPTY', 'assign_started'): "RESOURCING",
- ('EMPTY', 'fatal_error'): "OBSFAULT",
- ('RESOURCING', 'resourcing_succeeded_some_resources'): "IDLE",
- ('RESOURCING', 'resourcing_succeeded_no_resources'): "EMPTY",
- ('RESOURCING', 'resourcing_failed'): "OBSFAULT",
- ('RESOURCING', 'fatal_error'): "OBSFAULT",
- ('IDLE', 'assign_started'): "RESOURCING",
- ('IDLE', 'release_started'): "RESOURCING",
- ('IDLE', 'configure_started'): "CONFIGURING",
- ('IDLE', 'abort_started'): "ABORTING",
- ('IDLE', 'fatal_error'): "OBSFAULT",
- ('CONFIGURING', 'configure_succeeded'): "READY",
- ('CONFIGURING', 'configure_failed'): "OBSFAULT",
- ('CONFIGURING', 'abort_started'): "ABORTING",
- ('CONFIGURING', 'fatal_error'): "OBSFAULT",
- ('READY', 'end_succeeded'): "IDLE",
- ('READY', 'end_failed'): "OBSFAULT",
- ('READY', 'configure_started'): "CONFIGURING",
- ('READY', 'abort_started'): "ABORTING",
- ('READY', 'scan_started'): "SCANNING",
- ('READY', 'fatal_error'): "OBSFAULT",
- ('SCANNING', 'scan_succeeded'): "READY",
- ('SCANNING', 'scan_failed'): "OBSFAULT",
- ('SCANNING', 'end_scan_succeeded'): "READY",
- ('SCANNING', 'end_scan_failed'): "OBSFAULT",
- ('SCANNING', 'abort_started'): "ABORTING",
- ('SCANNING', 'fatal_error'): "OBSFAULT",
- ('ABORTING', 'abort_succeeded'): "ABORTED",
- ('ABORTING', 'abort_failed'): "OBSFAULT",
- ('ABORTING', 'fatal_error'): "OBSFAULT",
- ('ABORTED', 'obs_reset_started'): "RESETTING",
- ('ABORTED', 'restart_started'): "RESTARTING",
- ('ABORTED', 'fatal_error'): "OBSFAULT",
- ('OBSFAULT', 'obs_reset_started'): "RESETTING",
- ('OBSFAULT', 'restart_started'): "RESTARTING",
- ('OBSFAULT', 'fatal_error'): "OBSFAULT",
- ('RESETTING', 'obs_reset_succeeded'): "IDLE",
- ('RESETTING', 'obs_reset_failed'): "OBSFAULT",
- ('RESETTING', 'fatal_error'): "OBSFAULT",
- ('RESTARTING', 'restart_succeeded'): "EMPTY",
- ('RESTARTING', 'restart_failed'): "OBSFAULT",
- ('RESTARTING', 'fatal_error'): "OBSFAULT",
- }
-
- setups = {
- "UNINITIALISED": [],
- "INIT_ENABLED": ['init_started'],
- "INIT_DISABLED": ['init_started', 'to_offline'],
- "FAULT_ENABLED": ['init_started', 'init_failed'],
- "FAULT_DISABLED": ['init_started', 'to_offline', 'init_failed'],
- "OFF": ['init_started', 'init_succeeded'],
- "DISABLED": ['init_started', 'init_succeeded', 'to_offline'],
- "EMPTY": ['init_started', 'init_succeeded', 'on_succeeded'],
- "RESOURCING": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started'
- ],
- "IDLE": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources'],
- "CONFIGURING": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'configure_started'
- ],
- "READY": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'configure_started', 'configure_succeeded'
- ],
- "SCANNING": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'configure_started', 'configure_succeeded', 'scan_started'
- ],
- "ABORTING": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'abort_started'
- ],
- "ABORTED": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'abort_started', 'abort_succeeded'
- ],
- "OBSFAULT": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'fatal_error'
- ],
- "RESETTING": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'abort_started', 'abort_succeeded', 'obs_reset_started'
- ],
- "RESTARTING": [
- 'init_started', 'init_succeeded', 'on_succeeded',
- 'assign_started', 'resourcing_succeeded_some_resources',
- 'abort_started', 'abort_succeeded', 'restart_started'
- ],
- }
-
- # state = "UNINITIALISED" # for test debugging only
- # assert_state(state) # for test debugging only
-
- # Put the device into the state under test
- for action in setups[state_under_test]:
- state_model.perform_action(action)
- # state = transitions[state, action] # for test debugging only
- # assert_state(state) # for test debugging only
-
- # Check that we are in the state under test
- assert_state(state_under_test)
-
- # Test that the action under test does what we expect it to
- if (state_under_test, action_under_test) in transitions:
- # Action should succeed
- state_model.perform_action(action_under_test)
- assert_state(transitions[(state_under_test, action_under_test)])
- else:
- # Action should fail and the state should not change
- with pytest.raises(StateModelError):
- state_model.perform_action(action_under_test)
- assert_state(state_under_test)
diff --git a/tests/test_utils.py b/tests/test_utils.py
index fb749b7f630cec318d74020ff6e5e5d236e87ccb..f40481273b365fe0e1a5eb176387dc985c649e7f 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,4 +1,5 @@
"""Tests for skabase.utils."""
+from contextlib import nullcontext
import json
import pytest
@@ -6,6 +7,7 @@ from ska.base.utils import (
get_groups_from_json,
get_tango_device_type_id,
GroupDefinitionsError,
+ for_testing_only
)
TEST_GROUPS = {
@@ -209,3 +211,46 @@ def test_get_tango_device_type_id():
device_name = "domain/family/member"
result = get_tango_device_type_id(device_name)
assert result == ["family", "member"]
+
+
+@pytest.mark.parametrize(
+ "in_test, context",
+ [
+ (
+ False,
+ pytest.warns(
+ UserWarning,
+ match='foo should only be used for testing purposes'
+ )
+ ),
+ (True, nullcontext()),
+ ]
+)
+def test_for_testing_only(in_test, context):
+ """
+ Test the @for_testing_only decorator, to ensure that a warning is raised if and only
+ if we are NOT testing. This is achieved by patching the test, which cannot be done
+ using the @decorator syntax.
+ """
+ def foo():
+ """Dummy function for wrapping by decorator under test."""
+ return "foo"
+
+ foo = for_testing_only(foo, _testing_check=lambda: in_test)
+
+ with context:
+ assert foo() == "foo"
+
+
+def test_for_testing_only_decorator():
+ """
+ Test the unpatched for_testing_only decorator using the usual @decorator syntax
+ """
+ @for_testing_only
+ def bah():
+ """Dummy function for wrapping by decorator under test."""
+ return "bah"
+
+ with pytest.warns(None) as warning_record:
+ assert bah() == "bah"
+ assert len(warning_record) == 0 # no warning was raised because we are testing