Skip to content
Snippets Groups Projects
Select Git revision
  • 74acc0675f72b04143980da7c1c1da064797e43f
  • master default protected
  • deploy-components-parallel
  • fix-chrony-exporter
  • L2SS-2407-swap-iers-caltable-monitoring-port
  • L2SS-2357-fix-ruff
  • sync-up-with-meta-pypcc
  • stabilise-landing-page
  • all-stations-lofar2
  • v0.39.7-backports
  • Move-sdptr-to-v1.5.0
  • fix-build-ubuntu
  • tokens-in-env-files
  • fix-build
  • L2SS-2214-deploy-cdb
  • fix-missing-init
  • add-power-hardware-apply
  • L2SS-2129-Add-Subrack-Routine
  • Also-listen-internal-to-rpc
  • fix-build-dind
  • L2SS-2153--Improve-Error-Handling
  • v0.52.8-rc1 protected
  • v0.55.5 protected
  • v0.55.4 protected
  • 0.55.2.dev0
  • 0.55.1.dev0
  • 0.55.0.dev0
  • v0.54.0 protected
  • 0.53.2.dev0
  • 0.53.1.dev0
  • v0.52.3-r2 protected
  • remove-snmp-client
  • v0.52.3 protected
  • v0.52.3dev0 protected
  • 0.53.1dev0
  • v0.52.2-rc3 protected
  • v0.52.2-rc2 protected
  • v0.52.2-rc1 protected
  • v0.52.1.1 protected
  • v0.52.1 protected
  • v0.52.1-rc1 protected
41 results

test_opcua_client.py

  • Jan David Mol's avatar
    L2SS-412: Use asynctest.TestCase to simplify asyncio tests, move event_loop...
    Jan David Mol authored
    L2SS-412: Use asynctest.TestCase to simplify asyncio tests, move event_loop out of ProtocolAttribute
    74acc067
    History
    Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    test_opcua_client.py 11.49 KiB
    import numpy
    from clients.opcua_client import OPCUAConnection
    from clients import opcua_client
    
    import asyncua
    import io
    import asyncio
    
    from unittest import mock
    import unittest
    
    from test import base
    import asynctest
    
    
    class attr_props:
        def __init__(self, numpy_type):
            self.numpy_type = numpy_type
    
    
    attr_test_types = [
        attr_props(numpy_type=str),
        attr_props(numpy_type=numpy.bool_),
        attr_props(numpy_type=numpy.float32),
        attr_props(numpy_type=numpy.float64),
        attr_props(numpy_type=numpy.double),
        attr_props(numpy_type=numpy.uint8),
        attr_props(numpy_type=numpy.uint16),
        attr_props(numpy_type=numpy.uint32),
        attr_props(numpy_type=numpy.uint64),
        attr_props(numpy_type=numpy.int16),
        attr_props(numpy_type=numpy.int32),
        attr_props(numpy_type=numpy.int64)
    ]
    
    scalar_shape = (1,)
    spectrum_shape = (4,)
    image_shape = (2, 3)
    dimension_tests = [scalar_shape, spectrum_shape, image_shape]
    
    
    class TestOPCua(base.AsyncTestCase):
        @asynctest.patch.object(OPCUAConnection, "ping")
        @asynctest.patch.object(opcua_client, "Client")
        async def test_opcua_connection(self, m_opc_client, m_ping):
            """
            This tests verifies whether the correct connection steps happen. It checks whether we can init an OPCUAConnection object
            Whether we can set the namespace, and the OPCua client.
            """
    
            m_opc_client_members = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.get_namespace_index = asynctest.asynctest.CoroutineMock(return_value=42)
            m_opc_client_members.connect = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.disconnect = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.send_hello = asynctest.asynctest.CoroutineMock()
            m_opc_client.return_value = m_opc_client_members
    
            test_client = OPCUAConnection("opc.tcp://localhost:4874/freeopcua/server/", "http://lofar.eu", 5, mock.Mock(), self.loop)
            try:
                await test_client.start()
    
                m_opc_client.assert_called_once()  # makes sure the actual freeOPCua client object is created only once
    
                # this also implies test_client.connect() is called
                m_opc_client_members.get_namespace_index.assert_called_once_with("http://lofar.eu")
                self.assertEqual(42, test_client.name_space_index)
            finally:
                await test_client.stop()
    
    
        @asynctest.patch.object(OPCUAConnection, "ping")
        @asynctest.patch.object(opcua_client, "Client")
        @mock.patch.object(opcua_client, 'ProtocolAttribute')
        async def test_opcua_attr_setup(self, m_protocol_attr, m_opc_client, m_ping):
            """
            This tests covers the correct creation of read/write functions.
            In normal circumstances called by he attribute wrapper.
            Will be given 'comms_annotation', for OPCua that will be a node path and can access the attributes type and dimensions
    
            Test succeeds if there are no errors.
            """
    
            m_opc_client_members = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.get_namespace_index = asynctest.asynctest.CoroutineMock(return_value=2)
            m_opc_client_members.connect = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.disconnect = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.send_hello = asynctest.asynctest.CoroutineMock()
            m_objects_node = asynctest.Mock()
            m_objects_node.get_child = asynctest.asynctest.CoroutineMock()
            m_opc_client_members.get_objects_node = asynctest.Mock(return_value=m_objects_node)
            m_opc_client.return_value = m_opc_client_members
    
            for i in attr_test_types:
                class mock_attr:
                    def __init__(self, dtype, x, y):
                        self.numpy_type = dtype
                        self.dim_x = x
                        self.dim_y = y
    
                for j in dimension_tests:
                    if len(j) == 1:
                        dim_x = j[0]
                        dim_y = 0
                    else:
                        dim_x = j[1]
                        dim_y = j[0]
    
                    # create a fake attribute with only the required variables in it.
                    m_attribute = mock_attr(i.numpy_type, dim_x, dim_y)
    
                    # pretend like there is a running OPCua server with a node that has this name
                    m_annotation = ["2:PCC", f"2:testNode_{str(i.numpy_type)}_{str(dim_x)}_{str(dim_y)}"]
    
                    test_client = OPCUAConnection("opc.tcp://localhost:4874/freeopcua/server/", "http://lofar.eu", 5, mock.Mock(), self.loop)
                    try:
                        await test_client.start()
                        await test_client.setup_attribute(m_annotation, m_attribute)
                    finally:
                        await test_client.stop()
    
                    # success if there are no errors.
    
        def test_protocol_attr(self):
            """
            This tests finding an OPCua node and returning a valid object with read/write functions.
            (This step is normally initiated by the attribute_wrapper)
            """
    
            # for all datatypes
            for i in attr_test_types:
                # for all dimensions
                for j in dimension_tests:
    
                    node = mock.Mock()
    
                    # handle scalars slightly differently
                    if len(j) == 1:
                        dims = (j[0], 0)
                    else:
                        dims = (j[1], j[0])
    
                    ua_type = opcua_client.numpy_to_OPCua_dict[i.numpy_type]
                    test = opcua_client.ProtocolAttribute(node, dims[0], dims[1], ua_type)
                    print(test.dim_y, test.dim_x, test.ua_type)
    
                    """
                    Part of the test already includes simply not throwing an exception, but for the sake coverage these asserts have also
                    been added.
                    """
                    self.assertTrue(test.dim_y == dims[1], f"Dimensionality error, ProtocolAttribute.dim_y got: {test.dim_y} expected: {dims[1]}")
                    self.assertTrue(test.dim_x == dims[0], f"Dimensionality error, ProtocolAttribute.dim_y got: {test.dim_x} expected: {dims[0]}")
                    self.assertTrue(test.ua_type == ua_type, f"type error. Got: {test.ua_type} expected: {ua_type}")
                    self.assertTrue(hasattr(test, "write_function"), f"No write function found")
                    self.assertTrue(hasattr(test, "read_function"), f"No read function found")
    
        async def test_read(self):
            """
            This tests the read functions.
            """
    
            for j in dimension_tests:
                for i in attr_test_types:
                    def get_test_value():
                        return numpy.zeros(j, i.numpy_type)
    
                    async def get_flat_value():
                        return get_test_value().flatten()
    
                    m_node = asynctest.asynctest.CoroutineMock()
    
                    if len(j) == 1:
                        test = opcua_client.ProtocolAttribute(m_node, j[0], 0, opcua_client.numpy_to_OPCua_dict[i.numpy_type])
                    else:
                        test = opcua_client.ProtocolAttribute(m_node, j[1], j[0], opcua_client.numpy_to_OPCua_dict[i.numpy_type])
                    m_node.get_value = get_flat_value
                    val = await test.read_function()
    
                    comp = val == get_test_value()
                    self.assertTrue(comp.all(), "Read value unequal to expected value: \n\t{} \n\t{}".format(val, get_test_value()))
    
        def test_type_map(self):
            for numpy_type, opcua_type in opcua_client.numpy_to_OPCua_dict.items():
                # derive a default value that can get lost in a type translation
                if numpy_type in [str, numpy.str]:
                  default_value = "foo"
                elif numpy_type == numpy.bool_:
                  default_value = True
                else:
                  # integer or float type
                  # integers: numpy will drop the decimals for us
                  # floats: make sure we chose a value that has an exact binary representation
                  default_value = 42.25
    
                # apply our mapping
                v = asyncua.ua.uatypes.Variant(Value=numpy_type(default_value), VariantType=opcua_type)
    
                try:
                    # try to convert it to binary to force opcua to parse the value as the type
                    binary = asyncua.ua.ua_binary.variant_to_binary(v)
    
                    # reinterpret the resulting binary to obtain what opcua made of our value
                    binary_stream = io.BytesIO(binary)
                    reparsed_v = asyncua.ua.ua_binary.variant_from_binary(binary_stream)
                except Exception as e:
                    raise Exception(f"Conversion {numpy_type} -> {opcua_type} failed.") from e
    
                # did the value get lost in translation?
                self.assertEqual(v.Value, reparsed_v.Value, msg=f"Conversion {numpy_type} -> {opcua_type} failed.")
    
                # does the OPC-UA type have the same datasize (and thus, precision?)
                if numpy_type not in [str, numpy.str]:
                    self.assertEqual(numpy_type().itemsize, getattr(asyncua.ua.ua_binary.Primitives, opcua_type.name).size, msg=f"Conversion {numpy_type} -> {opcua_type} failed: precision mismatch")
    
    
    
        async def test_write(self):
            """
            Test the writing of values by instantiating a ProtocolAttribute attribute, and calling the write function.
            but the opcua function that writes to the server has been changed to the compare_values function.
            This allows the code to compare what values we want to write and what values would be given to a server.
            """
    
            # for all dimensionalities
            for j in dimension_tests:
    
                #for all datatypes
                for i in attr_test_types:
    
                    # get numpy array of the test value
                    def get_test_value():
                        return numpy.zeros(j, i.numpy_type)
    
                    # get opcua Varianttype array of the test value
                    def get_mock_value(value):
                        return asyncua.ua.uatypes.Variant(Value=value, VariantType=opcua_client.numpy_to_OPCua_dict[i.numpy_type])
    
                    m_node = asynctest.asynctest.CoroutineMock()
    
                    # create the protocolattribute
                    if len(j) == 1:
                        test = opcua_client.ProtocolAttribute(m_node, j[0], 0, opcua_client.numpy_to_OPCua_dict[i.numpy_type], self.loop)
                    else:
                        test = opcua_client.ProtocolAttribute(m_node, j[1], j[0], opcua_client.numpy_to_OPCua_dict[i.numpy_type], self.loop)
    
                    # comparison function that replaces `set_data_value` inside the attributes write function
                    async def compare_values(val):
                        # test valuest
                        val = val.tolist() if type(val) == numpy.ndarray else val
                        if j != dimension_tests[0]:
                            comp = val.Value == get_mock_value(get_test_value().flatten()).Value
                            self.assertTrue(comp.all(),
                                            "Array attempting to write unequal to expected array: \n\t got: {} \n\texpected: {}".format(val,get_mock_value(get_test_value())))
                        else:
                            comp = val == get_mock_value(get_test_value())
                            self.assertTrue(comp, "value attempting to write unequal to expected value: \n\tgot: {} \n\texpected: {}".format(val, get_mock_value(get_test_value())))
    
                    # replace the `set_data_value`, usualy responsible for communicating with the server with the `compare_values` function.
                    m_node.set_data_value = compare_values
    
                    # call the write function with the test values
                    await test.write_function(get_test_value())