Commit 00bf1222 authored by Elisabetta Giani's avatar Elisabetta Giani
Browse files

AT5-370: Added tests for sub-element master device.

Fixed some bugs in device code.
parent b08f2369
Loading
Loading
Loading
Loading
Loading
+7 −0
Original line number Original line Diff line number Diff line
__all__ = (
    "CspSubElementMaster",
    "CspSubElementSubarray"
)

from .subelement_master import CspSubElementMaster
+8 −8
Original line number Original line Diff line number Diff line
@@ -296,10 +296,8 @@ class CspSubElementMaster(SKAMaster):
        
        
       # _list_of_components: report the list of subordinate
       # _list_of_components: report the list of subordinate
        # sub-element components.
        # sub-element components.
        # Implemented as a dictionary
        # Implemented as a list of FQDNs
        # keys: the command name ('on', 'off',...)
        self._list_of_components = []
        # values: the list of components
        self._list_of_components = defaultdict(lambda: [])
        
        
        # _num_dev_completed_task: for each long-running command report the number
        # _num_dev_completed_task: for each long-running command report the number
        #  of subordinate components that completed the task
        #  of subordinate components that completed the task
@@ -309,7 +307,7 @@ class CspSubElementMaster(SKAMaster):
        self._num_dev_completed_task = defaultdict(lambda:0)
        self._num_dev_completed_task = defaultdict(lambda:0)
        
        
        # the last executed command 
        # the last executed command 
        self._last_executed_command = ""
        self._last_executed_command = "none"


        # PROTECTED REGION END #    //  CspSubElementMaster.init_device
        # PROTECTED REGION END #    //  CspSubElementMaster.init_device


@@ -334,7 +332,7 @@ class CspSubElementMaster(SKAMaster):
    def read_numOfDevCompletedTask(self):
    def read_numOfDevCompletedTask(self):
        # PROTECTED REGION ID(CspSubElementMaster.numOfDevCompletedTask_read) ENABLED START #
        # PROTECTED REGION ID(CspSubElementMaster.numOfDevCompletedTask_read) ENABLED START #
        """Return the numOfDevCompletedTask attribute."""
        """Return the numOfDevCompletedTask attribute."""
        return self._num_dev_completed_task
        return self._num_dev_completed_task[self._last_executed_command]
        # PROTECTED REGION END #    //  CspSubElementMaster.numOfDevCompletedTask_read
        # PROTECTED REGION END #    //  CspSubElementMaster.numOfDevCompletedTask_read


    def read_onCmdFailure(self):
    def read_onCmdFailure(self):
@@ -364,7 +362,7 @@ class CspSubElementMaster(SKAMaster):
    def read_standbyCmdFailure(self):
    def read_standbyCmdFailure(self):
        # PROTECTED REGION ID(CspSubElementMaster.standbyCmdFailure_read) ENABLED START #
        # PROTECTED REGION ID(CspSubElementMaster.standbyCmdFailure_read) ENABLED START #
        """Return the standbyCmdFailure attribute."""
        """Return the standbyCmdFailure attribute."""
        return self._failure_message['standby']
        return self._failure_raised['standby']
        # PROTECTED REGION END #    //  CspSubElementMaster.standbyCmdFailure_read
        # PROTECTED REGION END #    //  CspSubElementMaster.standbyCmdFailure_read


    def read_standbyFailureMessage(self):
    def read_standbyFailureMessage(self):
@@ -383,6 +381,8 @@ class CspSubElementMaster(SKAMaster):
        # PROTECTED REGION ID(CspSubElementMaster.adminMode_write) ENABLED START #
        # PROTECTED REGION ID(CspSubElementMaster.adminMode_write) ENABLED START #
        """Set the adminMode attribute."""
        """Set the adminMode attribute."""
        self._admin_mode = value
        self._admin_mode = value
        if self._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
            self.set_state(tango.DevState.DISABLE)
        # PROTECTED REGION END #    //  CspSubElementMaster.adminMode_write
        # PROTECTED REGION END #    //  CspSubElementMaster.adminMode_write


    def read_onCommandProgress(self):
    def read_onCommandProgress(self):
@@ -479,7 +479,7 @@ class CspSubElementMaster(SKAMaster):
    def read_listOfDevCompletedTask(self):
    def read_listOfDevCompletedTask(self):
        # PROTECTED REGION ID(CspSubElementMaster.listOfDevCompletedTask_read) ENABLED START #
        # PROTECTED REGION ID(CspSubElementMaster.listOfDevCompletedTask_read) ENABLED START #
        """Return the listOfDevCompletedTask attribute."""
        """Return the listOfDevCompletedTask attribute."""
        return self._list_dev_completed_task
        return self._list_dev_completed_task[self._last_executed_command]
        # PROTECTED REGION END #    //  CspSubElementMaster.listOfDevCompletedTask_read
        # PROTECTED REGION END #    //  CspSubElementMaster.listOfDevCompletedTask_read


    def read_listOfComponents(self):
    def read_listOfComponents(self):

tests/conftest.py

0 → 100644
+65 −0
Original line number Original line Diff line number Diff line
"""
A module defining a list of fixtures that are shared across tests.
"""
import importlib
import pytest

from tango.test_context import DeviceTestContext


@pytest.fixture(scope="class")
def tango_context(request):
    """Creates and returns a TANGO DeviceTestContext object.

    Parameters
    ----------
    request: _pytest.fixtures.SubRequest
        A request object gives access to the requesting test context.
    """
    test_properties = {
        'CspSubElementMaster': {
            'SkaLevel': '4',
            'LoggingTargetsDefault': '',
            'GroupDefinitions': '',
            'NrSubarrays': '16',
            'CapabilityTypes': '',
            'MaxCapabilities': ['VCC:197', 'FSP:27']
            },

        'CspSubElementSubarray': {
            'CapabilityTypes': ['VCC','FSP'],
            'LoggingTargetsDefault': '',
            'GroupDefinitions': '',
            'SkaLevel': '4',
            'SubID': '1',
        },
    }

    # This fixture is used to decorate classes like "TestSKABaseDevice" or
    # "TestSKALogger". We drop the first "Test" from the string to get the
    # class name of the device under test.
    # Similarly, the test module is like "test_base_device.py".  We drop the
    # first "test_" to get the module name
    test_class_name = request.cls.__name__
    class_name = test_class_name.split('Test', 1)[-1]
    print("class_name:", class_name)
    module = importlib.import_module("cspse.lmc", class_name)
    class_type = getattr(module, class_name)

    tango_context = DeviceTestContext(class_type, properties=test_properties.get(class_name))
    tango_context.start()
    yield tango_context
    tango_context.stop()


@pytest.fixture(scope="function")
def initialize_device(tango_context):
    """Re-initializes the device.

    Parameters
    ----------
    tango_context: tango.test_context.DeviceTestContext
        Context to run a device without a database.
    """
    print("Sono qui")
    yield tango_context.device.Init()
+275 −0
Original line number Original line Diff line number Diff line
#
# -*- coding: utf-8 -*-
#
# This file is part of the CspSubElementMaster project
#
"""Contain the tests for the Master."""

# Standard imports
import sys
import os
import time

# Imports
import re
import pytest
from tango import DevState
from tango import DevFailed

# PROTECTED REGION ID(CspSubElementMaster.test_additional_imports) ENABLED START #
from ska.base.control_model import AdminMode, ControlMode, HealthState, SimulationMode, TestMode
# PROTECTED REGION END #    //  CspSubElementMaster.test_additional_imports
# Device test case
# PROTECTED REGION ID(CspSubElementMaster.test_CspSubElementMaster_decorators) ENABLED START #
@pytest.mark.usefixtures("tango_context")
# PROTECTED REGION END #    //  CspSubElementMaster.test_CspSubElementMaster_decorators
class TestCspSubElementMaster(object):
    """Test case for packet generation."""

    capabilities = ['FSP:27', 'VCC:197']
    properties = {
        'SkaLevel': '4',
        'LoggingTargetsDefault': '',
        'GroupDefinitions': '',
        'NrSubarrays': '16',
        'CapabilityTypes': '',
        'MaxCapabilities': ['FSP:27', 'VCC:197']
        }

    @classmethod
    def mocking(cls):
        """Mock external libraries."""
        # Example : Mock numpy
        # cls.numpy = CspSubElementMaster.numpy = MagicMock()
        # PROTECTED REGION ID(CspSubElementMaster.test_mocking) ENABLED START #
        # PROTECTED REGION END #    //  CspSubElementMaster.test_mocking

    def test_properties(self, tango_context):
        # Test the properties
        # PROTECTED REGION ID(CspSubElementMaster.test_properties) ENABLED START #
        # PROTECTED REGION END #    //  CspSubElementMaster.test_properties
        pass

    # PROTECTED REGION ID(CspSubElementMaster.test_State_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_State_decorators
    def test_State(self, tango_context):
        """Test for State"""
        # PROTECTED REGION ID(CspSubElementMaster.test_State) ENABLED START #
        assert tango_context.device.State() == DevState.INIT
        # PROTECTED REGION END #    //  CspSubElementMaster.test_State

    # PROTECTED REGION ID(CspSubElementMaster.test_Status_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_Status_decorators
    def test_Status(self, tango_context):
        """Test for Status"""
        # PROTECTED REGION ID(CspSubElementMaster.test_Status) ENABLED START #
        assert tango_context.device.Status() == "The device is in INIT state."
        # PROTECTED REGION END #    //  CspSubElementMaster.test_Status

    # PROTECTED REGION ID(CspSubElementMaster.test_GetVersionInfo_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_GetVersionInfo_decorators
    def test_GetVersionInfo(self, tango_context):
        """Test for GetVersionInfo"""
        # PROTECTED REGION ID(CspSubElementMaster.test_GetVersionInfo) ENABLED START #
        versionPattern = re.compile(
            r'CspSubElementMaster, lmcbaseclasses, [0-9].[0-9].[0-9], '
            r'A set of generic base devices for SKA Telescope.')
        versionInfo = tango_context.device.GetVersionInfo()
        assert (re.match(versionPattern, versionInfo[0])) != None
        # PROTECTED REGION END #    //  CspSubElementMaster.test_GetVersionInfo


    # PROTECTED REGION ID(CspSubElementMaster.test_buildState_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_buildState_decorators
    def test_buildState(self, tango_context):
        """Test for buildState"""
        # PROTECTED REGION ID(CspSubElementMaster.test_buildState) ENABLED START #
        buildPattern = re.compile(
            r'lmcbaseclasses, [0-9].[0-9].[0-9], '
            r'A set of generic base devices for SKA Telescope')
        assert (re.match(buildPattern, tango_context.device.buildState)) != None
        # PROTECTED REGION END #    //  CspSubElementMaster.test_buildState

    # PROTECTED REGION ID(CspSubElementMaster.test_versionId_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_versionId_decorators
    def test_versionId(self, tango_context):
        """Test for versionId"""
        # PROTECTED REGION ID(CspSubElementMaster.test_versionId) ENABLED START #
        versionIdPattern = re.compile(r'[0-9].[0-9].[0-9]')
        assert (re.match(versionIdPattern, tango_context.device.versionId)) != None
        # PROTECTED REGION END #    //  CspSubElementMaster.test_versionId

    # PROTECTED REGION ID(CspSubElementMaster.test_healthState_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_healthState_decorators
    def test_healthState(self, tango_context):
        """Test for healthState"""
        # PROTECTED REGION ID(CspSubElementMaster.test_healthState) ENABLED START #
        assert tango_context.device.healthState == HealthState.OK
        # PROTECTED REGION END #    //  CspSubElementMaster.test_healthState

    # PROTECTED REGION ID(CspSubElementMaster.test_adminMode_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_adminMode_decorators
    def test_adminMode(self, tango_context):
        """Test for adminMode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_adminMode) ENABLED START #
        assert tango_context.device.adminMode == AdminMode.ONLINE
        # PROTECTED REGION END #    //  CspSubElementMaster.test_adminMode

    # PROTECTED REGION ID(CspSubElementMaster.test_write_adminMode_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_adminMode_decorators
    def test_write_adminMode(self, tango_context):
        """Test for adminMode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_adminMode) ENABLED START #
        tango_context.device.adminMode = AdminMode.OFFLINE
        time.sleep(2)
        assert tango_context.device.adminMode == AdminMode.OFFLINE
        assert tango_context.device.State() == DevState.DISABLE
        # PROTECTED REGION END #    //  CspSubElementMaster.test_adminMode

    # PROTECTED REGION ID(CspSubElementMaster.test_On_invalid_adminModedecorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_On_invalid_adminMode decorators
    def test_On_invalid_adminMode(self, tango_context):
        """Test On command with offline adminmode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_On_invalid_adminMode) ENABLED START #
        with pytest.raises(DevFailed) as df:
            argin  = [""]
            tango_context.device.On(argin)
        assert "On command can't" in str(df.value.args[0].desc)
        # PROTECTED REGION END #    //  CspSubElementMaster.test_On_invalid_adminMode

    # PROTECTED REGION ID(CspSubElementMaster.test_Off_invalid_adminModedecorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_Off_invalid_adminMode decorators
    def test_Off_invalid_adminMode(self, tango_context):
        """Test Off command with offline adminmode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_Off_invalid_adminMode) ENABLED START #
        with pytest.raises(DevFailed) as df:
            argin  = [""]
            tango_context.device.Off(argin)
        assert "Off command can't" in str(df.value.args[0].desc)
        # PROTECTED REGION END #    //  CspSubElementMaster.test_Off_invalid_adminMode

    # PROTECTED REGION ID(CspSubElementMaster.test_Standby_invalid_adminModedecorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_Standby_invalid_adminMode decorators
    def test_Standby_invalid_adminMode(self, tango_context):
        """Test Standby command with offline adminmode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_Standby_invalid_adminMode) ENABLED START #
        with pytest.raises(DevFailed) as df:
            argin  = [""]
            tango_context.device.Standby(argin)
        assert "Standby command can't" in str(df.value.args[0].desc)
        # PROTECTED REGION END #    //  CspSubElementMaster.test_Standby_invalid_adminMode

    # PROTECTED REGION ID(CspSubElementMaster.test_On_invalid_Statedecorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_On_invalid_State decorators
    def test_On_invalid_State(self, tango_context):
        """Test On command with offline adminmode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_On_invalid_State) ENABLED START #
        # set the adminMode to ONLINE
        # reinitialize the device to return to
        # adminMode =ONLINE and State=INIT
        tango_context.device.Init()
        time.sleep(2)
        assert tango_context.device.adminMode == AdminMode.ONLINE
        assert tango_context.device.State() == DevState.INIT
        with pytest.raises(DevFailed) as df:
            argin = []
            tango_context.device.On(argin)
        assert "Command On can't" in str(df.value.args[0].desc)
        # PROTECTED REGION END #    //  CspSubElementMaster.test_On_invalid_State

    # PROTECTED REGION ID(CspSubElementMaster.test_Off_invalid_Statedecorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_Off_invalid_State decorators
    def test_Off_invalid_State(self, tango_context):
        """Test Off command with offline adminmode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_Off_invalid_State) ENABLED START #
        assert tango_context.device.adminMode == AdminMode.ONLINE
        with pytest.raises(DevFailed) as df:
            argin = []
            tango_context.device.Off(argin)
        assert "Command Off can't" in str(df.value.args[0].desc)
        # PROTECTED REGION END #    //  CspSubElementMaster.test_Off_invalid_State

    # PROTECTED REGION ID(CspSubElementMaster.test_Standby_invalid_Statedecorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_Standby_invalid_State decorators
    def test_Standby_invalid_State(self, tango_context):
        """Test Standby command with offline adminmode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_Standby_invalid_State) ENABLED START #
        assert tango_context.device.adminMode == AdminMode.ONLINE
        with pytest.raises(DevFailed) as df:
            argin = []
            tango_context.device.Standby(argin)
        assert "Command Standby can't" in str(df.value.args[0].desc)
        # PROTECTED REGION END #    //  CspSubElementMaster.test_Standby_invalid_State

    # PROTECTED REGION ID(CspSubElementMaster.test_controlMode_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_controlMode_decorators
    def test_controlMode(self, tango_context):
        """Test for controlMode"""
        # PROTECTED REGION ID(CspSubElementMaster.test_controlMode) ENABLED START #
        assert tango_context.device.controlMode == ControlMode.REMOTE
        # PROTECTED REGION END #    //  CspSubElementMaster.test_controlMode

    # PROTECTED REGION ID(CspSubElementMaster.test_maxCapabilities_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_maxCapabilities_decorators
    def test_maxCapabilities(self, tango_context):
        """Test for maxCapabilities"""
        # PROTECTED REGION ID(CspSubElementMaster.test_maxCapabilities) ENABLED START #
        assert tango_context.device.maxCapabilities == ('FSP:27', 'VCC:197')
        # PROTECTED REGION END #    //  CspSubElementMaster.test_maxCapabilities

    # PROTECTED REGION ID(CspSubElementMaster.test_availableCapabilities_decorators) ENABLED START #
    # PROTECTED REGION END #    //  CspSubElementMaster.test_availableCapabilities_decorators
    def test_availableCapabilities(self, tango_context):
        """Test for availableCapabilities"""
        # PROTECTED REGION ID(CspSubElementMaster.test_availableCapabilities) ENABLED START #
        assert tango_context.device.availableCapabilities == ('FSP:27', 'VCC:197')
        # PROTECTED REGION END #    //  CspSubElementMaster.test_availableCapabilities

    def test_num_of_dev_completed_task(self, tango_context):
        """Test numOfCompletedTask attribute"""
        assert tango_context.device.numOfDevCompletedTask == 0

    def test_list_of_completed_task(self, tango_context):
        """Test listOfCompletedTask attribute"""
        assert tango_context.device.listOfDevCompletedTask == None

    def test_list_of_components(self, tango_context):
        """Test listOfComponents attribute"""
        assert tango_context.device.listOfComponents == None

    def test_timeout_expired(self, tango_context):
        """Test xxxTimeoutExpired flags attribute"""
        assert not tango_context.device.onCmdTimeoutExpired
        assert not tango_context.device.offCmdTimeoutExpired
        assert not tango_context.device.standbyCmdTimeoutExpired

    def test_failure_raised(self, tango_context):
        """Test xxxCmdFailure attributes"""
        assert not tango_context.device.onCmdFailure
        assert not tango_context.device.offCmdFailure
        assert not tango_context.device.standbyCmdFailure

    def test_failure_message(self, tango_context):
        """Test xxxFailureMessage attributes"""
        assert not tango_context.device.onFailureMessage
        assert not tango_context.device.offFailureMessage
        assert not tango_context.device.standbyFailureMessage

    def test_command_progress(self, tango_context):
        """Test xxxCommandProgress attributes"""
        assert tango_context.device.onCommandProgress == 0
        assert tango_context.device.offCommandProgress == 0
        assert tango_context.device.standbyCommandProgress == 0

    def test_command_duration_measured(self, tango_context):
        """Test xxxCmdDurationMeasured attributes"""
        assert tango_context.device.onCmdDurationMeasured == 0
        assert tango_context.device.offCmdDurationMeasured == 0
        assert tango_context.device.standbyCmdDurationMeasured == 0

    def test_command_duration_expected(self, tango_context):
        """Test xxxCmdDurationExpected attributes"""
        assert tango_context.device.onCmdDurationExpected == 30
        assert tango_context.device.offCmdDurationExpected == 30
        assert tango_context.device.standbyCmdDurationExpected == 30