From d4729cd390037b2cba0ffe9b410db90a1ceed289 Mon Sep 17 00:00:00 2001
From: Gianluca Marotta <gianluca.marotta@inaf.it>
Date: Mon, 21 Dec 2020 19:53:18 +0100
Subject: [PATCH] CT-206 Test that the transaction id is captured in log when
applied to Configure and Assign Resources
---
.../csp_lmc_common/utils/test_utils.py | 67 +++++++++++++++++
.../tests/unit/cspsubarray_unit_test.py | 73 ++++++++++++++++++-
2 files changed, 138 insertions(+), 2 deletions(-)
create mode 100644 csp-lmc-common/csp_lmc_common/utils/test_utils.py
diff --git a/csp-lmc-common/csp_lmc_common/utils/test_utils.py b/csp-lmc-common/csp_lmc_common/utils/test_utils.py
new file mode 100644
index 0000000..c65e226
--- /dev/null
+++ b/csp-lmc-common/csp_lmc_common/utils/test_utils.py
@@ -0,0 +1,67 @@
+import time
+import logging
+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):
+ """
+ """
+ self.proxy = proxy
+ self.attr_name = attr_name
+ self.expected_state = expected_state
+ self.message = message
+ self.current_state = DevState.DISABLE
+
+ def get_attribute(self):
+ return self.attr_name
+
+ def sample(self):
+ """
+ extract the state of client and store it
+ """
+ device_attr = self.proxy.read_attribute(self.attr_name)
+ self.current_state = device_attr.value
+ #LOGGER.info("attr_name: {} current_state:{}".format(self.attr_name, self.current_state))
+
+ def is_satisfied(self):
+ """
+ Check if the state satisfies this test condition
+ """
+
+ if isinstance(self.current_state, numpy.ndarray):
+ 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
+ self.interval = interval
+
+ def check(self, probe: Probe):
+ """
+ Repeatedly check if the probe is satisfied.
+ Assert false when the timeout expires.
+ """
+ timer = Timeout(self.timeout)
+ probe.sample()
+ while not probe.is_satisfied():
+ if timer.has_expired():
+ LOGGER.debug("Check Timeout on:{}".format(probe.get_attribute()))
+ assert False, probe.message
+ time.sleep(self.interval)
+ probe.sample()
+ LOGGER.debug("Check success on: {}".format(probe.get_attribute()))
+ assert True
diff --git a/csp-lmc-common/tests/unit/cspsubarray_unit_test.py b/csp-lmc-common/tests/unit/cspsubarray_unit_test.py
index ba6939a..e25e80e 100644
--- a/csp-lmc-common/tests/unit/cspsubarray_unit_test.py
+++ b/csp-lmc-common/tests/unit/cspsubarray_unit_test.py
@@ -4,13 +4,15 @@ import sys
import mock
import pytest
import tango
+import logging
+import re
from mock import Mock, MagicMock
-
-from ska.base.control_model import HealthState, ObsState
+from ska.base.control_model import HealthState, ObsState, LoggingLevel
from ska.base.commands import ResultCode
from tango.test_context import DeviceTestContext
from tango import DevState, DevFailed
from csp_lmc_common.CspSubarray import CspSubarray
+from csp_lmc_common.utils.test_utils import Probe, Poller
def test_cspsubarray_state_and_obstate_value_after_initialization():
"""
@@ -147,6 +149,57 @@ def test_cspsbarray_state_after_On_forwarded_to_subelement_subarray():
assert tango_context.device.State() == DevState.ON
assert tango_context.device.obsState == ObsState.EMPTY
+def test_cspsubarray_transaction_id_in_log(capsys):
+ """
+ Test that when transaction_id decorator is applied to the Configure
+ and Assign Resources command, both transaction id are captured in log
+ """
+ device_under_test = CspSubarray
+ cbf_subarray_fqdn = 'mid_csp_cbf/sub_elt/subarray_01'
+ pss_subarray_fqdn = 'mid_csp_pss/sub_elt/subarray_01'
+ cbf_subarray_state_attr = 'obsState'
+ dut_properties = {
+ 'CspMaster':'mid_csp/elt/master',
+ 'CbfSubarray': cbf_subarray_fqdn,
+ 'PssSubarray': 'mid_csp_pss/sub_elt/subarray_01',
+ }
+ event_subscription_map = {}
+ cbf_subarray_device_proxy_mock = Mock()
+ pss_subarray_device_proxy_mock = Mock()
+ proxies_to_mock = {
+ cbf_subarray_fqdn: cbf_subarray_device_proxy_mock,
+ pss_subarray_fqdn: pss_subarray_device_proxy_mock,
+ }
+ cbf_subarray_device_proxy_mock.subscribe_event.side_effect = (
+ lambda attr_name, event_type, callback, *args, **kwargs: event_subscription_map.update({attr_name: callback}))
+
+ with fake_tango_system(device_under_test, initial_dut_properties=dut_properties, proxies_to_mock=proxies_to_mock) as tango_context:
+ cbf_subarray_device_proxy_mock.On.side_effect = return_ok
+ pss_subarray_device_proxy_mock.On.side_effect = return_ok
+ tango_context.device.On()
+ assert tango_context.device.State() == DevState.ON
+ tango_context.device.AssignResources('{"example":"band"}')
+ dummy_event = create_dummy_obs_event(cbf_subarray_fqdn, ObsState.IDLE)
+ event_subscription_map[cbf_subarray_state_attr](dummy_event)
+ assert tango_context.device.obsState == ObsState.IDLE
+ tango_context.device.Configure('{"id":"sbi-400-scienceA"}')
+ # a prober is needed since the duration of the Configure command is variable.
+ prober_obs_state = Probe(tango_context.device, "obsState", ObsState.READY, f"Configure command out of time")
+ Poller(10, 0.1).check(prober_obs_state)
+ assert_that_log_contains('ConfigureCommand',capsys)
+ assert_that_log_contains('AssignResourcesCommand', capsys)
+
+def assert_that_log_contains(name:str,capsys):
+ patterns = [f'|Transaction.*(?<=Enter\[{name}\])',f'|Transaction.*(?<=Exit\[{name}\])']
+ for pattern in patterns:
+ found = False
+ out, err = capsys.readouterr()
+ if re.match(pattern,out):
+ found = True
+ break
+ if not found:
+ raise AssertionError(f'pattern ({pattern}) not found in expected log messages')
+
def return_ok():
"""
Return a FAILED code in the execution of a device method.
@@ -187,6 +240,22 @@ def create_dummy_event(cbf_subarray_fqdn, event_value):
fake_event.device.dev_name.side_effect=(lambda *args, **kwargs: mock_event_dev_name(cbf_subarray_fqdn))
return fake_event
+def create_dummy_obs_event(cbf_subarray_fqdn, event_value):
+ """
+ Create a mocked event object to test the event callback method
+ associate to the attribute at subscription.
+ param: cbf_subarray_fqdn the CBF Subarray FQDN
+ event_value the expected value
+ return: the fake event
+ """
+ fake_event = Mock()
+ fake_event.err = False
+ fake_event.attr_name = f"{cbf_subarray_fqdn}/obsstate"
+ fake_event.attr_value.value = event_value
+ fake_event.attr_value.name = 'obsState'
+ fake_event.device.name = cbf_subarray_fqdn
+ fake_event.device.dev_name.side_effect=(lambda *args, **kwargs: mock_event_dev_name(cbf_subarray_fqdn))
+ return fake_event
@contextlib.contextmanager
def fake_tango_system(device_under_test, initial_dut_properties={}, proxies_to_mock={},
device_proxy_import_path='tango.DeviceProxy'):
--
GitLab