Skip to content
Snippets Groups Projects

Resolve CT-215

Merged Gianluca Marotta requested to merge CT-215 into master
1 file
+ 33
1
Compare changes
  • Side-by-side
  • Inline
@@ -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
Loading