Resolve CT-215

Closes CT-215

csp-lmc-mid/tests/integration/MidCspSubarray_test.py

@@ -16,7 +16,6 @@ import time
 import random
 import numpy as np
 import logging
-import unittest
 import pytest
 import json
 
@@ -45,7 +44,7 @@ def prepare_configuration_string(filename="test_ConfigureScan_ADR4.json"):
 
 # Device test case
 @pytest.mark.usefixtures("midcsp_master", "midcsp_subarray01", "cbf_subarray01")
-class TestBase(unittest.TestCase):
+class TestBase(object):
     fixture_names = ()
 
     @pytest.fixture(autouse=True)
@@ -93,7 +92,8 @@ class TestCspSubarray(TestBase):
             obs_state = self.midcsp_subarray01.obsState
             if obs_state == ObsState.IDLE:
                 self._release_all_receptors()
-
+                prober_obs_state = Probe(self.midcsp_subarray01, "obsState", ObsState.EMPTY, f"CSP Subarray not EMPTY")
+                Poller(4, 0.2).check(prober_obs_state)
     def _setup_subarray_off(self):
         """
         Set the subarray state to OFF-EMPTY
@@ -176,6 +176,7 @@ class TestCspSubarray(TestBase):
             'dish': { 'receptorIDList': list(map(str, argin))}}
         return json.dumps(param)
     
+    @pytest.mark.init_EMPTY
     def test_AFTER_initialization(self):
         """
         Test for State after CSP startup.
@@ -188,9 +189,116 @@ class TestCspSubarray(TestBase):
         Poller(4, 0.2).check(prober_subarray_state)
         prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.EMPTY, f"CSP Subarray not EMPTY")
         Poller(4, 0.2).check(prober_subarray_obsstate)
+   
+    @pytest.mark.init_IDLE
+    def test_re_initialization_when_IDLE(self):
+        """
+        Test for re-initialization of mid-csp device after a restart of
+        the Tango device. 
+        The CspSubarray aligns to the state/obsstate of CBFSubarray: ON/IDLE
+        """
+        self._setup_subarray()
+        json_config= self._build_resources([1,2])
+        self.midcsp_subarray01.AssignResources(json_config)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.IDLE, f"CSP Subarray not IDLE")
+        Poller(4, 0.2).check(prober_subarray_obsstate)
+        # open a proxy to the adminstrative server
+        dserver_proxy = tango.DeviceProxy(self.midcsp_subarray01.adm_name())
+        dserver_proxy. devrestart('mid_csp/elt/subarray_01')
+        prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not ON after re-initialization")
+        Poller(10, 0.2).check(prober_subarray_state)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.IDLE, f"CSP Subarray not IDLE after re-initialization")
+        Poller(4, 0.2).check(prober_subarray_obsstate)
 
+    @pytest.mark.init_READY
+    def test_re_initialization_when_READY(self):
+        """
+        Test for re-initialization of mid-csp device after the restart of
+        the Tango device. 
+        The CspSubarray aligns to the state/obsstate of CBFSubarray: ON/READY
+        """
+        self._setup_subarray()
+        self._configure_scan() 
+        # open a proxy to the adminstrative server
+        dserver_proxy = tango.DeviceProxy(self.midcsp_subarray01.adm_name())
+        dserver_proxy. devrestart('mid_csp/elt/subarray_01')
+        prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not ON after re-initialization")
+        Poller(10, 0.2).check(prober_subarray_state)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.READY, f"CSP Subarray not READY after re-initialization")
+        Poller(4, 0.2).check(prober_subarray_obsstate)
+    
+    @pytest.mark.init_SCANNING
+    @pytest.mark.parametrize("elapsed_time", [0.1, 0.2, 0.3, 0.4, 0.5,1,2])
+    def test_re_initialization_when_SCANNING(self, elapsed_time):
+        """
+        Test for re-initialization of mid-csp device after the of the
+        Tango device.
+        The CspSubarray aligns to the state/obsstate of CBFSubarray: ON/SCANNING.
+        The tests ends with the CSP Subarray in READY.
+        Test is executed several times with different wait time (dealys) before
+        sending the EndScan.
+        """
+        self._setup_subarray()
+        self._configure_scan()
+        self.midcsp_subarray01.Scan('2')
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.SCANNING, f'CSP Subarray not SCANNING')
+        Poller(10, 0.2).check(prober_subarray_obsstate)
+        # wait for a variable time
+        time.sleep(elapsed_time)
+        # open a proxy to the adminstrative server
+        dserver_proxy = tango.DeviceProxy(self.midcsp_subarray01.adm_name())
+        # restart the device to force a reinitialization
+        dserver_proxy. devrestart('mid_csp/elt/subarray_01')
+        prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not ON after re-initialization")
+        Poller(10, 0.2).check(prober_subarray_state)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.SCANNING, f"CSP Subarray not SCANNING after re-initialization")
+        Poller(4, 0.2).check(prober_subarray_obsstate)
+        # wait a variable time before sending the endscan
+        time.sleep(elapsed_time)
+        self.midcsp_subarray01.EndScan()
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.READY, f'CSP Subarray not SCANNING')
+        Poller(5, 0.2).check(prober_subarray_obsstate)
+    
+    @pytest.mark.init_ABORTED
+    def test_re_initialization_when_ABORTED(self):
+        """
+        Test for re-initialization of mid-csp device after the restart of the
+        Tango device.
+        The CspSubarray align to the state/obsstate of CBFSubarray: ON/ABORTED
+        """
+        self._setup_subarray()
+        self._assign_receptors()
+        self.midcsp_subarray01.Abort()
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.ABORTED, f'CSP Subarray not ABORTED')        
+        # open a proxy to the adminstrative server
+        dserver_proxy = tango.DeviceProxy(self.midcsp_subarray01.adm_name())
+        dserver_proxy. devrestart('mid_csp/elt/subarray_01')
+        prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not ON after re-initialization")
+        Poller(10, 0.2).check(prober_subarray_state)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.ABORTED, f"CSP Subarray not ABORTED after re-initialization")
+        Poller(4, 0.2).check(prober_subarray_obsstate)
+   
+    @pytest.mark.init_FAULT
+    def test_re_initialization_when_FAULT(self):
+        """
+        Test for re-initialization of mid-csp device after killing the Tango device.
+        The CspSubarray align to the state/obsstate of CBFSubarray: ON/IDLE
+        """
+        self._setup_subarray()
+        self._assign_receptors()
+        configuration_string = prepare_configuration_string("test_ConfigureScan_invalid_cbf_json.json")
+        self.midcsp_subarray01.Configure(configuration_string)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.FAULT, f'CSP Subarray not FAULT')
+        Poller(10, 0.2).check(prober_subarray_obsstate)
+        # open a proxy to the adminstrative server
+        dserver_proxy = tango.DeviceProxy(self.midcsp_subarray01.adm_name())
+        dserver_proxy. devrestart('mid_csp/elt/subarray_01')
+        prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not ON after re-initialization")
+        Poller(10, 0.2).check(prober_subarray_state)
+        prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.FAULT, f"CSP Subarray not FAULT after re-initialization")
+        Poller(4, 0.2).check(prober_subarray_obsstate)
+    
     def test_subarray_state_AFTER_on_command_execution(self):
-
         """
         Test for State after CSP startup.
         The CspSubarray State at start is OFF.
@@ -204,20 +312,29 @@ class TestCspSubarray(TestBase):
         prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not OFF")
         Poller(4, 0.2).check(prober_subarray_state)
 
+    @pytest.mark.off_resourcing
     def test_OffCommand_after_RESOURCING(self):
         """
         Test that the Off command send the device in OFF-EMPTY
         if it is ON-RESOURCING
         """
-        self._setup_subarray
+        self._setup_subarray()
+        subarray_state = self.midcsp_subarray01.State()
+        obs_state = self.midcsp_subarray01.obsState
+        LOGGER.info("CSPSubarray state before test:{}-{}".format(subarray_state, obs_state))
         json_config= self._build_resources([1,2])
         self.midcsp_subarray01.AssignResources(json_config)
+        subarray_state = self.midcsp_subarray01.State()
+        obs_state = self.midcsp_subarray01.obsState
+        LOGGER.info("CSPSubarray state before test:{}-{}".format(subarray_state, obs_state))
         self.midcsp_subarray01.Off()        
         prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.OFF, f"CSP Subarray not OFF")
         Poller(4, 0.2).check(prober_subarray_state)
         prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.EMPTY, f"CSP Subarray not EMPTY")
         Poller(4, 0.2).check(prober_subarray_obsstate)
         receptors = self.midcsp_subarray01.assignedReceptors
+        LOGGER.info(f'type of {type(receptors)}')
+        LOGGER.info(f'list of receptors{receptors}')
         assert not receptors.any(), f"CSP Subarray is not empty"
 
     def test_OffCommand_after_IDLE(self):
@@ -266,7 +383,8 @@ class TestCspSubarray(TestBase):
         Poller(10, 0.2).check(prober_subarray_obsstate)
         receptors = self.midcsp_subarray01.assignedReceptors
         assert not receptors.any(), f"CSP Subarray is not empty"
-    
+   
+    @pytest.mark.off
     def test_OffCommand_after_SCANNING(self):
         """
         Test that the Off command send the device in OFF-EMPTY
@@ -656,11 +774,15 @@ class TestCspSubarray(TestBase):
         prober_obs_state = Probe(self.midcsp_subarray01, "obsState", ObsState.IDLE, f"CSP Subarray not IDLE")
         Poller(4, 0.2).check(prober_obs_state)
 
-    @pytest.mark.csp_k8s
-    def test_send_abort_WHILE_in_configuring(self):
+    @pytest.mark.abort
+    #@pytest.mark.parametrize("elapsed_time", [0.4, 0.41, 0.42, 0.43, 0.44])
+    @pytest.mark.parametrize("elapsed_time", [0.42, 0.43, 0.44])
+    @pytest.mark.parametrize('execution_number', range(2))
+    def test_send_abort_WHILE_in_configuring(self, elapsed_time, execution_number):
         """
-        Configure the CSP Subarray with a JSon string including
-        the new ADR22 fields.
+        Test that the subarray is able to handle the Abort command when issued 
+        at any time.
+        The test is repeated several times with different delay times.
         """
         self._setup_subarray()
         self._assign_receptors()
@@ -671,7 +793,7 @@ class TestCspSubarray(TestBase):
         LOGGER.info(f"Configuring CSP subarray01")
         configuration_string = prepare_configuration_string()
         (result_code, msg) = self.midcsp_subarray01.Configure(configuration_string)
-        #time.sleep(0.1)
+        time.sleep(elapsed_time)
         self.midcsp_subarray01.Abort()
         # check
         prober_subarray_obstate = Probe(self.midcsp_subarray01, 'obsState', ObsState.ABORTED,
@@ -721,26 +843,20 @@ class TestCspSubarray(TestBase):
         prober_obs_state = Probe(self.midcsp_subarray01, "obsState", ObsState.EMPTY, f"CSP Subarray not EMPTY")
         Poller(4, 0.2).check(prober_obs_state)
 
-    '''
-    def test_configureScan_with_subarray_ready(self, midcsp_subarray01, midcsp_master):
+    def test_configureScan_WHEN_subarray_ready(self, midcsp_subarray01, midcsp_master):
         """
         Test that the Configure() command is issued when the Subarray
         state already READY
         """
+        self._configure_scan()
         obs_state = midcsp_subarray01.obsState
         assert obs_state == ObsState.READY
-        f = open(file_path +"/test_ConfigureScan_basic.json")
-        obs_state = midcsp_subarray01.obsState
-        try:
-            midcsp_subarray01.Configure(f.read().replace("\n", ""))
-        except tango.DevFailed as tango_err:
-            print("configure error:", tango_err.args[0].desc)
-        obs_state = midcsp_subarray01.obsState
-        f.close()
-        time.sleep(5)
-        obs_state = midcsp_subarray01.obsState
-        assert obs_state == ObsState.READY
+        configuration_string = prepare_configuration_string("test_ConfigureScan_basic.json")
+        self.midcsp_subarray01.Configure(configuration_string)
+        prober_obs_state = Probe(self.midcsp_subarray01, "obsState", ObsState.READY, f"CSP Subarray not READY")
+        Poller(4, 0.2).check(prober_obs_state)
 
+'''
     def test_remove_receptors_when_ready(self, midcsp_subarray01):
         """
         Test that the complete deallocation of receptors fails