Commit 4b7c8af6 authored by Elisabetta Giani's avatar Elisabetta Giani
Browse files

CT-69: code cleaning. Add assertpy package. Use number version for ska-python images.

parent ed062562

csp-lmc-mid/docker/Dockerfile

+2 −2
Original line number Diff line number Diff line 
FROM nexus.engageska-portugal.pt/ska-docker/ska-python-buildenv:latest AS buildenv

FROM nexus.engageska-portugal.pt/ska-docker/ska-python-runtime:latest AS runtime

FROM nexus.engageska-portugal.pt/ska-docker/ska-python-buildenv:9.3.1 AS buildenv

FROM nexus.engageska-portugal.pt/ska-docker/ska-python-runtime:9.3.1 AS runtime



# create ipython profile to so that itango doesn't fail if ipython hasn't run yet

RUN ipython profile create

csp-lmc-mid/tests/acceptance_tests/MidCspSubarrayConfiguration_test.py

+60 −56
Original line number Diff line number Diff line
@@ -14,6 +14,7 @@ import sys 
import os

import time

import logging

from pytest import fail



# Tango imports

import tango
@@ -25,21 +26,31 @@ import unittest 
from ska.base.control_model import ObsState

from utils import Poller, Probe



# TODO: remove call to sleep 



LOGGER = logging.getLogger(__name__)

# Device test case

@pytest.mark.usefixtures("midcsp_master", "midcsp_subarray01", "cbf_subarray01")





def prepare_configuration_string(filename="test_ConfigureScan_ADR4.json"):

    """Create the config string for CSP-CBF"""

    try:

        json_file = open(filename)

        configuration_string = json_file.read().replace("\n", "")

        return configuration_string

    except Exception as e:

        LOGGER.warning(f"Unable to locate file {filename}")





# Device test case

@pytest.mark.usefixtures("midcsp_master", "midcsp_subarray01", "cbf_subarray01")

class TestBase(unittest.TestCase):

    fixture_names = ()



    @pytest.fixture(autouse=True)

    def auto_injector_fixture(self, request):

        names = self.fixture_names

        for name in names:

            setattr(self, name, request.getfixturevalue(name))





class TestCspSubarrayConfiguration(TestBase):

    fixture_names = ("midcsp_master", "midcsp_subarray01", "cbf_subarray01")
@@ -58,10 +69,11 @@ class TestCspSubarrayConfiguration(TestBase): 
        self.midcsp_subarray01.AddReceptors([unassigned_receptors[0], ])

        receptor_membership_expected[unassigned_receptors[0] - 1] = 1

        # wait for the expected value of the CSP Master receptorMembership attribute

        prober_receptor_membership = Probe(self.midcsp_master, 'receptorMembership', receptor_membership_expected, f"Wrong state")

        prober_receptor_membership = Probe(self.midcsp_master, 'receptorMembership', receptor_membership_expected,

                                           f"Wrong value for receptor membership attribute")

        Poller(10, 0.2).check(prober_receptor_membership)

        # wait for the transition of the CSP subarray to ON

        prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.ON, f"Wrong state")

        prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.ON, f"Wrong CSP Subarray state")

        Poller(10, 0.2).check(prober_subarray_state)



    def _reset_subarray_to_disable_state(self):
@@ -71,14 +83,13 @@ class TestCspSubarrayConfiguration(TestBase): 
            if state == DevState.DISABLE:

                return

            if state == DevState.ON:

                num_of_receptors = len(self.midcsp_master.receptorMembership)

                receptor_membership_expected = [0] * num_of_receptors

                LOGGER.info("Remove all receptors")

                self.midcsp_subarray01.RemoveAllReceptors()

                # wait for the transition of the CSP subarray to OFF

                prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.OFF, f"Wrong state")

                # wait for the transition of the CSP sub-array to OFF

                prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.OFF,

                                              f"Wrong CSP Subarray state")

                Poller(7, 0.2).check(prober_subarray_state)

                prober_master_state = Probe(self.midcsp_master, 'state', DevState.ON, f"Wrong state")

                prober_master_state = Probe(self.midcsp_master, 'state', DevState.ON, f"Wrong CSP Master state")

                Poller(7, 0.2).check(prober_master_state)

            # Set the CSP subarray to OFF issuing the Standby command

            # on CSP Master
@@ -86,22 +97,15 @@ class TestCspSubarrayConfiguration(TestBase): 
            if state == DevState.OFF:

                LOGGER.info("CSP Subarray State is OFF...going to execute Standby")

                self.midcsp_master.Standby("")

                prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.DISABLE, f"Wrong state")

                prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.DISABLE,

                                              f"Wrong CSP Subarray state")

                Poller(5, 0.2).check(prober_subarray_state)

                prober_master_state = Probe(self.midcsp_master, 'state', DevState.STANDBY, f"Wrong state")

                prober_master_state = Probe(self.midcsp_master, 'state', DevState.STANDBY,

                                            f"Wrong CSP Master state")

                Poller(7, 0.2).check(prober_master_state)



        except tango.DevFailed as tango_err:

            LOGGER.warn(f"Unable to reset subarray to init state")



    def _prepare_configuration_string(self,filename="test_ConfigureScan_ADR4.json"):

        """Create the config string for CSP-CBF"""

        try:

            json_file = open(filename)

            configuration_string = json_file.read().replace("\n", "")

            return configuration_string

        except Exception as e:

            LOGGER.warn(f"Unable to locate file {filename}")

            LOGGER.warning(f"Unable to reset subarray to init state")



    def _setup_csp_subarray(self):

        """
@@ -114,29 +118,27 @@ class TestCspSubarrayConfiguration(TestBase): 
        LOGGER.info("Switch on Master")

        self.midcsp_master.On("")

        # wait for the transition of the CSP subarray to OFF

        prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.OFF, f"Wrong state")

        prober_subarray_state = Probe(self.midcsp_subarray01, 'state', DevState.OFF, f"Wrong CSP Subarray state")

        Poller(10, 0.2).check(prober_subarray_state)

        state = self.midcsp_subarray01.State()

        assert state in [DevState.OFF], "assuming that mid_csp_subarray_01 is OFF"

        self.__set_csp_subarray_to_on()





    def test_send_configure_to_cbf_and_json_stored(self):

        """

        Configure the CSP Subarray with a JSon string including

        the new ADR4 fields.

        """

        #prober_subarray_obstate = Probe(self.midcsp_subarray01, 'obsState', ObsState.IDLE, f"Wrong state")

        #Poller(5, 0.2).check(prober_subarray_obstate)

        self._setup_csp_subarray()

        configuration_string = self._prepare_configuration_string()

        configuration_string = prepare_configuration_string()

        state = self.midcsp_subarray01.State()

        assert state == DevState.ON, "assume the CSP subarray State is ON"

        # exercise the device

        LOGGER.debug(f"sending configure:{configuration_string}")

        LOGGER.info("Send configuration command")

        self.midcsp_subarray01.Configure(configuration_string)

        prober_subarray_obstate = Probe(self.midcsp_subarray01, 'obsState', ObsState.READY, f"Wrong CSP Subarray obsState")

        prober_subarray_obstate = Probe(self.midcsp_subarray01, 'obsState', ObsState.READY,

                                        f"Wrong CSP Subarray obsState")

        Poller(5, 0.2).check(prober_subarray_obstate)

        # check

        obs_state = self.midcsp_subarray01.obsState
@@ -144,7 +146,7 @@ class TestCspSubarrayConfiguration(TestBase): 
        stored_json = self.midcsp_subarray01.validScanConfiguration

        assert stored_json == configuration_string



    def test_configure_with_json_missing_configId(self):

    def test_configure_cspsubarray_WITH_json_missing_configId(self):

        """

        Configure the CSP Subarray sending a json configuration script 

        without the configID entry.
@@ -152,17 +154,18 @@ class TestCspSubarrayConfiguration(TestBase): 
        # setup the test

        # Set the CSP Subarray to ON-IDLE state

        self._setup_csp_subarray()

        configuration_string = self._prepare_configuration_string("test_ConfigureScan_without_configID.json")

        configuration_string = prepare_configuration_string("test_ConfigureScan_without_configID.json")



        # exercise the device

        with pytest.raises(tango.DevFailed) as df:

        with pytest.raises(tango.DevFailed):

            self.midcsp_subarray01.Configure(configuration_string)

            time.sleep(3)

        # check

        obs_state = self.midcsp_subarray01.obsState

        assert obs_state == ObsState.IDLE

        #assert obs_state == ObsState.IDLE, f"CSP Subarray obsState has wrong value ({obs_state})"

        assert_that(obs_state).described_by("AAA").is_equal_to(ObsState.IDLE)



    def test_configure_with_cspsubarray_in_wrong_state(self):

            self._reset_subarray_to_disable_state()

def test_configure_cspsubarray_WHEN_in_wrong_state(self):

        """

        Set the CSP Subarray in a wrong state and issue the

        Configure command on it.
@@ -170,7 +173,7 @@ class TestCspSubarrayConfiguration(TestBase): 
        # setup the test: Subarray DISABLE-IDLE

        self._setup_csp_subarray()

        self._reset_subarray_to_disable_state()

        configuration_string = self._prepare_configuration_string()

        configuration_string = prepare_configuration_string()

        init_state = self.midcsp_subarray01.State()



        # exercise the device
@@ -178,25 +181,26 @@ class TestCspSubarrayConfiguration(TestBase): 
            self.midcsp_subarray01.Configure(configuration_string)

        if df:

            err_msg = str(df.value.args[0].desc)

            # TODO: aggiungere quale device fallisce

            LOGGER.info(f"Command configure failed with error: {err_msg}")

        # check

        # Subarray final ObsState IDLE

        # Subarray final ObsState=IDLE

        obs_state = self.midcsp_subarray01.obsState

        end_state = self.midcsp_subarray01.State()

        assert obs_state == ObsState.IDLE 

        assert end_state == init_state 

        assert obs_state == ObsState.IDLE, f"Current ObsState should be IDLE"

        assert end_state == init_state, f"Current State should be the initial one"

        assert_that(obs_state).described_by("Current ObsState should be IDLE").is_not_equal_to(ObsState.IDLE)



    def test_configure_with_cbf_invalid_json(self):

    def test_configure_cbf_WITH_invalid_json(self):

        """

        Set the CSP Subarray in a wrong state and issue the

        Configure command on it.

        CSP Subarray sends an invalid json configuration to

        CBF Subarray.

        """

        # setup the test: Subarray DISABLE-IDLE

        self._setup_csp_subarray()

        configuration_string = self._prepare_configuration_string()

        init_state = self.midcsp_subarray01.State()



        configuration_string = prepare_configuration_string()

        # exercise the device

        LOGGER.info(f"Configuring CSP subarray01")

        self.midcsp_subarray01.Configure(configuration_string)

        # check

        # Subarray final ObsState IDLE

csp-lmc-mid/tests/acceptance_tests/test_requirements.txt

+1 −0
Original line number Diff line number Diff line
@@ -2,4 +2,5 @@ pytest-json-report 
pycodestyle

pytest-ordering

importlib

assertpy

mock

csp-lmc-mid/tests/acceptance_tests/utils.py

+8 −2
Original line number Diff line number Diff line
@@ -4,12 +4,16 @@ import numpy 
from tango import DeviceProxy, DevState



LOGGER = logging.getLogger(__name__)





class Timeout:

    def __init__(self, duration):

        self.endtime = time.time() + duration



    def has_expired(self):

        return time.time() > self.endtime





class Probe:

    def __init__(self, proxy, attr_name, expected_state, message):

        """
@@ -22,6 +26,7 @@ class Probe: 
 

    def get_attribute(self):

        return self.attr_name



    def sample(self):

        """

        extract the state of client and store it
@@ -33,11 +38,12 @@ class Probe: 
        """

        Check if the state satisfies this test condition

        """

        #if type(self.current_state) is numpy.ndarray:



        if isinstance(self.current_state, numpy.ndarray):

            return ((self.expected_state == self.current_state).all())

            return (self.expected_state == self.current_state).all()

        return self.expected_state == self.current_state





class Poller:

    def __init__(self, timeout, interval):

        self.timeout = timeout