Testing TANGO devices using PyTest

apatkar · December 19, 2018, 4:16pm

Hi Everyone,

I am writing test cases for TANGO devices using Pytest. I have a few queries about the same.

1. Test TANGO clients

I have two TANGO devices, device A (a client) and device B(a server). Command invoked from device A, internally invokes command of device B and as a result some of the attributes of TANGO device B change.

Testing approach:
To test the command of device A, I create a proxy of device B (from test function of device A) and assert with changed attributes of device B. For the test to pass successfully, there is a dependency on Device B to be up and running.

Issue: Since the communication between Device A and Device B may take some time, I get Segmentation fault error. One possibility can be that the assert statement is checked while device B takes time to respond to the change. The error is resolved when the test is run again. I have also tried to add time.sleep(3) statement but still no luck.

I need help on how to overcome the Segmentation Fault issue. Also, how shall I test Device A without depending on Device B?

2. Access device properties inside test_context.py

The Tango Device A uses a device property during it’s initialization. Hence, there is a need to supply this device property in order to test Device A.

Issue:
It is observed that init function in test_context.py takes properties as input. However, while trying below solution, properties for the instance (tango_context) of device is not set. Hence, the device A’s initialization is getting failed which result in all the test cases to fail unless we define a default value for each property.

File: conftest.py

# some imports

from tango.test_context import DeviceTestContext

@pytest.fixture(scope="class")
def tango_context(request):

# code to get klass name

properties = {'DeviceB_FQDN': 'tango://test/device/b'}

tango_context = DeviceTestContext(klass, properties=properties)
tango_context.start()

3. Defining attribute properties for Forwarded Attributes

Some attributes of Device B are forwarded on Device A. However, it is mandatory to specify __root_attr attribute property to initialze device A successfully. Without this property, device A goes into Alarm state after initialization.

Is there any way to add attribute property and return test_context of Device A? Also, how to set default value for __root_attr in python?

4. Testing events

Comment on test_client.py says that:

"""Client tests that run against the standard TangoTest device.

Due to a TANGO 9 issue (#821), the device is run without any database.
Note that this means that various features won't work:

 * No device configuration via properties.
 * No event generated by the server.
 * No memorized attributes.
 * No device attribute configuration via the database.

So don't even try to test anything of the above as it will not work
and is even likely to crash the device (!)

"""

Is there any workaround to enable testing of events and device/attribute configuration through properties?

I understand that it is a long post but any inputs on above queries will be helpful.

Christmas wishes,
Apurva Patkar

ajoubert · December 21, 2018, 7:45pm

Hi Apurva

That is a lot of questions!

You will get a segfault if you are trying to use more than one DeviceTestContext instance in the same test. You have to set process=True to launch them in separate subprocesses instead. See the link to the PyTango PR in the code below.
If you are overriding init_device, you need to call the super classes’ init_device method first. My example uses the context handler to start the device, but calling start() directly also works.
Not sure.
The PyTango unit tests include testing of events, so it is possible using the DeviceTestContext, although there are some limitations like a port number has to be specified. See pytango/tests/test_event.py at develop · tango-controls/pytango · GitHub. Maybe the comments in test_client.py are just referring the case where an external device server is launched without using the test context?

(Sorry if code below isn’t syntax highlighted - it wasn’t in the preview. I used a “code=python” tag)

from tango import DeviceProxy, DevState
from tango.server import Device
from tango.server import command, attribute, device_property
from tango.test_utils import DeviceTestContext


def test_multiple_devices():

    class DeviceA(Device):
        b_address = device_property(dtype='str')

        @command(dtype_in='str')
        def SetAttrOnDeviceB(self, arg):
            dp = DeviceProxy(self.b_address)
            dp.SetAttr(arg)

    class DeviceB(Device):

        def init_device(self):
            self._attr = '123'

        @attribute(dtype='str')
        def attr(self):
            return self._attr

        @command(dtype_in='str')
        def SetAttr(self, arg):
            self._attr = arg

    # As we will launch multiple test devices, we need to set process=True
    # See https://github.com/tango-controls/pytango/pull/77
    context_b = DeviceTestContext(DeviceB, process=True)
    with context_b as proxy_b:
        # test accessing DeviceB directly
        assert proxy_b.attr == '123'
        proxy_b.SetAttr('456')
        assert proxy_b.attr == '456'

        # get address for DeviceB (changes for every test run)
        properties = {'b_address': context_b.get_device_access()}
        context_a = DeviceTestContext(DeviceA, properties=properties, process=True)
        with context_a as proxy_a:
            # test accessing DeviceB via DeviceA
            proxy_a.SetAttrOnDeviceB('789')
            assert proxy_b.attr == '789'


def test_device_property_on_init():

    class TestDevice(Device):
        my_property = device_property(dtype='str')

        def init_device(self):
            super(TestDevice, self).init_device()
            assert self.my_property == 'Testing123'

    # get address for DeviceB (changes for every test run)
    properties = {'my_property': 'Testing123'}
    with DeviceTestContext(TestDevice, properties=properties) as proxy:
        # Check state just to ensure device started up
        assert proxy.state() == DevState.UNKNOWN

apatkar · January 11, 2019, 11:05am

Hi Anton,

Thank you for the inputs; really helped us a lot. We are now able to provide property values while running the TANGO device server using DeviceTestContext.

I have few more queries related to the testing the events of TANGO devices!

I will take the example of TANGO device A (client) and TANGO device B (server). Command invoked from device A, internally invokes command of device B and as a result some of the attributes of TANGO device B change. Device A has subscribed change event on the related attributes of device B.

We are able to test change events subscribed by device A (running using DeviceTestContext only if device B is running with a database.

We also tried running device B, without a database (using DeviceTestContext). As suggested, we set process=True to use more than one DeviceTestContext instance in the same test. Following are some issues we faced with this approach:

    1. We are not able to test change events subscribed by device A. This is because the polling necessary for sending events was not set for Device B running without a database. How can we set the polling for Device B's attributes in order to test change events? Checked the [url=https://pytango.readthedocs.io/en/stable/server_api/attribute.html?highlight=set_polling_period]PyTango APIs[/url], but unable to find [i]set_polling_period()[/i] API for read-only attributes.
    
    2. Using [i]process=True[/i] while creating DeviceTestContexts of device A and device B, decreases the code coverage (measured using Pytest) significantly. The coverage reduced to 40%, which was 70% when [i]process=False[/i]. We are unable to identify why the coverage is reduced in this case and what can be done to stop this reduction.

Thanks and Regards,
Apurva Patkar

ajoubert · January 14, 2019, 9:08pm

Hi Apurva

Glad I could help.

I tried doing this, and it seems to be possible. Hopefully I understood your request correctly. The most import thing to note is that the DeviceContext instantiation must be provided with a port number for the zeromq events mechanism to work. You can see this by looking at the PyTango unit tests: pytango/tests/test_event.py at develop · tango-controls/pytango · GitHub
Maybe the coverage is lower because the tool that is measuring the coverage is not aware of the additional processes launched? Maybe you can find one that is multi-process aware.


import socket
import time

from tango import DeviceProxy, EventType
from tango.server import Device
from tango.server import command, attribute, device_property
from tango.test_utils import DeviceTestContext


def get_open_port():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("", 0))
    s.listen(1)
    port = s.getsockname()[1]
    s.close()
    return port


def test_subscribe_change_events():

    MAX_RETRIES = 10
    TIME_PER_RETRY = 0.1

    class DeviceA(Device):
        b_address = device_property(dtype='str')

        def init_device(self):
            super(DeviceA, self).init_device()
            self._sub_id = None
            self._sub_proxy = None
            self._results = []

        @attribute(dtype=('str',), max_dim_x=10)
        def results(self):
            return self._results

        def _event_callback(self, evt):
            if evt.attr_value:
                self._results.append(evt.attr_value.value)
            else:
                print('bad event: %s' % evt)

        @command(dtype_in='str')
        def SetAttrOnDeviceB(self, arg):
            dp = DeviceProxy(self.b_address)
            dp.SetAttr(arg)

        @command()
        def SubscribeToAttrOnDeviceB(self):
            assert self._sub_id is None, "Already subscribed!"
            dp = DeviceProxy(self.b_address)
            self._sub_id = dp.subscribe_event(
                "attr", EventType.CHANGE_EVENT, self._event_callback, wait=True)
            self._sub_proxy = dp
            # Note:
            # - need to keep a reference to the device proxy if we want to unsubscribe
            # - subscribing triggers _event_callback with the current value

        @command()
        def UnsubscribeFromAttrOnDeviceB(self):
            assert self._sub_id is not None, "Not subscribed!"
            # unsubscribing can only be done using the same device proxy instance
            self._sub_proxy.unsubscribe_event(self._sub_id)

    class DeviceB(Device):

        def init_device(self):
            super(DeviceB, self).init_device()
            self._attr = '123'
            self.set_change_event("attr", True, False)  # manually triggered

        @attribute(dtype='str')
        def attr(self):
            return self._attr

        @command(dtype_in='str')
        def SetAttr(self, arg):
            self._attr = arg
            self.push_change_event("attr", arg)

    # events from test context device only work if a port is specified (ZMQ related)
    port = get_open_port()
    context_b = DeviceTestContext(DeviceB, port=port, process=True)
    with context_b:
        # get address for DeviceB (changes for every test run)
        properties = {'b_address': context_b.get_device_access()}
        context_a = DeviceTestContext(DeviceA, properties=properties, process=True)
        with context_a as proxy_a:
            # set up the subscription once - this will provide the initial value
            proxy_a.SubscribeToAttrOnDeviceB()
            assert proxy_a.results == ('123', )

            # trigger events on DeviceB via DeviceA
            proxy_a.SetAttrOnDeviceB('456')
            proxy_a.SetAttrOnDeviceB('789')
            # wait for events
            for _ in range(MAX_RETRIES):
                results = proxy_a.results
                if len(results) >= 3:
                    break
                time.sleep(TIME_PER_RETRY)
            # Check the event values
            assert results == ('123', '456', '789')

            # unsubscribe (e.g., if the event is no longer required)
            proxy_a.UnsubscribeFromAttrOnDeviceB()

            # test the events have stopped
            proxy_a.SetAttrOnDeviceB('000')
            # wait for events
            for _ in range(MAX_RETRIES):
                new_results = proxy_a.results
                if len(new_results) > 3:
                    break
                time.sleep(TIME_PER_RETRY)
            # Check the event values haven't changed
            assert new_results == results