CT-206 Test that the transaction id is captured in log when applied to...

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

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():

    """

        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.

    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'):