Select Git revision
test_opcua_client.py

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