AT5-370: Added decorators to check device State before (b08f2369) · Commits · Gianluca Marotta / csp-lmc-subelement-old

.pylintrc

+2 −2

Original line number	Diff line number	Diff line
		@@ -28,7 +28,7 @@ limit-inference-results=100

		# List of plugins (as comma separated values of python modules names) to load,
		# usually to register additional checkers.
		load-plugins=pylint_junit
		load-plugins=

		# Pickle collected data for later comparisons.
		persistent=yes
		@@ -95,7 +95,7 @@ evaluation=10.0 - ((float(5 * error + warning + refactor + convention) / stateme
		# Set the output format. Available formats are text, parseable, colorized, json
		# and msvs (visual studio). You can also give a reporter class, e.g.
		# mypackage.mymodule.MyReporterClass.
		output-format=junit
		output-format=text

		# Tells whether to display a full report or only the messages.
		reports=yes

cspse/lmc/decorators.py

0 → 100644

+108 −0

Original line number	Diff line number	Diff line
		import functools
		import tango

		tasks = {}

		# note: f.__name__ is of type is_XXX_allowed
		# f.__name__[3:-8] select the command name
		# this decorator build a dictionary with the command name as key and
		# the handler as value.
		task = lambda f:tasks.setdefault(f.__name__[3:-8], f)

		@task
		def is_standby_allowed(device_instance):
		"""
		Allowed method for Standby method.
		Command Standby is allowed when the device State is ON or
		STANDBY.

		:return: True if the method is allowed, otherwise False.
		"""
		if device_instance.get_state() in [tango.DevState.ON,
		tango.DevState.STANDBY]:
		return True
		return False

		@task
		def is_on_allowed(device_instance):
		"""
		Allowed method for On method.
		Command On is allowed when the device State is ON or
		STANDBY.

		:return: True if the method is allowed, otherwise False.
		"""
		if device_instance.get_state() in [tango.DevState.ON,
		tango.DevState.STANDBY]:
		return True
		return False

		@task
		def is_off_allowed(device_instance):
		"""
		Allowed method for Off method.
		Command Off is allowed when the device State is OFF or
		STANDBY.

		:return: True if the method is allowed, otherwise False.
		"""
		if device_instance.get_state() in [tango.DevState.OFF,
		tango.DevState.STANDBY]:
		return True
		return False

		def is_command_allowed(device_instance, cmd_name):
		"""
		Call the allowed method for the command name specified
		as input argument
		:param device_istance: the TANGO device instance
		:param cmd_name: the name of command to execute

		:return: True/False
		"""
		tasks[cmd_name](device_instance)

		class IsCommandAllowed(object):
		"""
		Class designed to be a decorator for the Master power methods.
		The decorator function performs a check on the input argument
		to control if the command is issued on the whole sub-element.
		If this is the case, it checks the State of the sub-element Master
		device and rejects the command accordingly to the State
		machine setting.

		:raise: tango.DevFailed exception if the command can't be executed
		"""
		def __init__(self, args, *kwargs):
		# store the decorators parameters:
		# args: the list of sub-element attributes to subscribe, to track the
		# sub-element command progress and detect a command timeout
		# kwargs: a dictionary: key ="cmd_name",
		# value = command to execute ('on', 'off'...)
		self._args = args
		self._kwargs = kwargs

		def __call__(self, f):
		@functools.wraps(f)
		def input_args_check(args, *kwargs):
		# the Master device instance
		dev_instance = args[0]
		# the command name
		cmd_to_exec = f.__name__
		# the command input argument
		input_arg = args[1]
		device_list = input_arg
		# Note: device list is a reference to args[1]: changing
		# device_list content, args[1] changes accordingly!
		num_of_devices = len(input_arg)
		if num_of_devices == 0:
		# check the device State: if it not the proper value the command is
		# not executed
		if not is_command_allowed(dev_instance, cmd_to_exec.lower()):
		msg = "Command {} can't be executed when the device is {}".format(cmd_to_exec,
		dev_instance.get_state())
		tango.Except.throw_exception("Command failure",msg,
		"IsCommandAllowed decorator",
		tango.ErrSeverity.ERR)
		return f(args, *kwargs)
		return input_args_check

cspse/lmc/subelement_master.py

+13 −52

Original line number	Diff line number	Diff line
		@@ -24,9 +24,12 @@ from tango import AttrWriteType, PipeWriteType
		from collections import defaultdict
		# Additional import
		# PROTECTED REGION ID(CspSubElementMaster.additionnal_import) ENABLED START #
		from ska.base.SKAMaster import SKAMaster
		from ska.base import SKAMaster
		from ska.base.control_model import HealthState, AdminMode, LoggingLevel
		from csp_lmc_common.cspcommons.utils import CmdExecState
		from csp_lmc_common.utils.cspcommons import CmdExecState
		from csp_lmc_common.utils.decorators import AdminModeCheck
		from .decorators import IsCommandAllowed
		from . import release
		# PROTECTED REGION END # // CspSubElementMaster.additionnal_import

		__all__ = ["CspSubElementMaster", "main"]
		@@ -198,7 +201,6 @@ class CspSubElementMaster(SKAMaster):
		dtype='DevBoolean',
		label="On execution timeout flag",
		polling_period=2000,
		abs_change=1,
		doc="Signal the occurence of a timeout during the execution of the on command.",
		)

		@@ -206,7 +208,6 @@ class CspSubElementMaster(SKAMaster):
		dtype='DevBoolean',
		label="Off execution timeout flag",
		polling_period=2000,
		abs_change=1,
		doc="Signal the occurence of a timeout during the execution of the Off command.",
		)

		@@ -214,7 +215,6 @@ class CspSubElementMaster(SKAMaster):
		dtype='DevBoolean',
		label="Standby execution timeout flag.",
		polling_period=2000,
		abs_change=1,
		doc="Signal the occurence of a timeout during the execution of the Standby command.",
		)

		@@ -239,6 +239,7 @@ class CspSubElementMaster(SKAMaster):
		def init_device(self):
		"""Initialises the attributes and properties of the CspSubElementMaster."""
		SKAMaster.init_device(self)
		self.set_state(tango.DevState.INIT)
		# PROTECTED REGION ID(CspSubElementMaster.init_device) ENABLED START #
		# PROTECTED REGION ID(CspSubElementMaster.init_device) ENABLED START #
		# _cmd_execution_state: implement the execution state of a long-running
		@@ -441,7 +442,7 @@ class CspSubElementMaster(SKAMaster):
		def read_onCmdDurationMeasured(self):
		# PROTECTED REGION ID(CspSubElementMaster.onCmdDurationMeasured_read) ENABLED START #
		"""Return the onCmdDurationMeasured attribute."""
		return self._cmd_duration_measured['on']return self._cmd_duration_measured['on']
		return self._cmd_duration_measured['on']
		# PROTECTED REGION END # // CspSubElementMaster.onCmdDurationMeasured_read

		def read_offCmdDurationMeasured(self):
		@@ -491,21 +492,6 @@ class CspSubElementMaster(SKAMaster):
		# Commands
		# --------

		@AdminModeCheck('On')
		def is_On_allowed(self):
		"""
		TANGO is_allowed method

		Command On is allowed when the device State is STANDBY.

		:return: True if the method is allowed, otherwise False.
		"""
		# PROTECTED REGION ID(CspMaster.is_On_allowed) ENABLED START #
		# Note: as per SKA Guidelines, the command is allowed when
		if self.get_state() not in [tango.DevState.STANDBY, tango.DevState.ON]:
		return False
		return True

		@command(
		dtype_in='DevVarStringArray',
		doc_in="The list of sub-element components FQDNs to switch-on or an empty list to switch-on the whole "
		@@ -516,6 +502,8 @@ class CspSubElementMaster(SKAMaster):
		"CSP SubElement component to switch ON.",
		)
		@DebugIt()
		@IsCommandAllowed()
		@AdminModeCheck('On')
		def On(self, argin):
		# PROTECTED REGION ID(CspSubElementMaster.On) ENABLED START #
		"""
		@@ -538,21 +526,6 @@ class CspSubElementMaster(SKAMaster):
		pass
		# PROTECTED REGION END # // CspSubElementMaster.On

		@AdminModeCheck('Off')
		def is_Off_allowed(self):
		"""
		TANGO is_allowed method

		Command Off is allowed when the device State is STANDBY.

		:return: True if the method is allowed, otherwise False.
		"""
		# PROTECTED REGION ID(CspMaster.is_On_allowed) ENABLED START #
		# Note: as per SKA Guidelines, the command is allowed when
		if self.get_state() not in [tango.DevState.STANDBY, tango.DevState.OFF]:
		return False
		return True

		@command(
		dtype_in='DevVarStringArray',
		doc_in="If the array length is 0, the command applies to the whole"
		@@ -561,6 +534,8 @@ class CspSubElementMaster(SKAMaster):
		"CSP SubElement component to switch OFF.",
		)
		@DebugIt()
		@IsCommandAllowed()
		@AdminModeCheck('Off')
		def Off(self, argin):
		# PROTECTED REGION ID(CspSubElementMaster.Off) ENABLED START #
		"""
		@@ -579,21 +554,6 @@ class CspSubElementMaster(SKAMaster):
		pass
		# PROTECTED REGION END # // CspSubElementMaster.Off

		@AdminModeCheck('Standby')
		def is_Standby_allowed(self):
		"""
		TANGO is_allowed method

		Command Standby is allowed when the device State is ON.

		:return: True if the method is allowed, otherwise False.
		"""
		# PROTECTED REGION ID(CspMaster.is_On_allowed) ENABLED START #
		# Note: as per SKA Guidelines, the command is allowed when
		if self.get_state() not in [tango.DevState.STANDBY, tango.DevState.ON]:
		return False
		return True

		@command(
		dtype_in='DevVarStringArray',
		doc_in="If the array length is 0, the command applies to the whole"
		@@ -602,6 +562,8 @@ class CspSubElementMaster(SKAMaster):
		"CSP SubElement icomponent to put in STANDBY mode.",
		)
		@DebugIt()
		@IsCommandAllowed()
		@AdminModeCheck('Standby')
		def Standby(self, argin):
		# PROTECTED REGION ID(CspSubElementMaster.Standby) ENABLED START #
		"""
		@@ -635,7 +597,6 @@ class CspSubElementMaster(SKAMaster):
		# Run server
		# ----------


		def main(args=None, **kwargs):
		"""Main function of the CspSubElementMaster module."""
		# PROTECTED REGION ID(CspSubElementMaster.main) ENABLED START #

tests/test_ska_skeleton.py

deleted100644 → 0

+0 −40

Original line number	Diff line number	Diff line
		#!/usr/bin/env python
		# -- coding: utf-8 --

		"""Tests for the ska_python_skeleton module."""
		import pytest

		from ska_python_skeleton import ska_python_skeleton


		# TODO: Replace all the following examples with tests for the ska_python_skeleton package code
		def test_something():
		"""Example: Assert with no defined return value."""
		assert True


		def test_with_error():
		"""Example: Assert raising error."""
		with pytest.raises(ValueError):
		# Do something that raises a ValueError
		raise ValueError


		# Fixture example
		@pytest.fixture
		def an_object():
		"""Example: Define fixture for subsequent test."""
		return {}


		def test_ska_python_skeleton(an_object):
		"""Example: Assert fixture return value."""
		assert an_object == {}


		def test_package():
		"""Example: Assert the ska_python_skeleton package code."""
		assert ska_python_skeleton.function_example() is None
		foo = ska_python_skeleton.SKA()
		assert foo.example_2() == 2
		assert foo.example() is None