diff --git a/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client.py b/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client.py
index 5df197618c702ad9900c3c1bc7df624b7ffde3a0..ab20b238297af55adef923c7417273dc07e57dc6 100644
--- a/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client.py
+++ b/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client.py
@@ -153,18 +153,34 @@ class TestOPCua(base.AsyncTestCase):
self.assertTrue(hasattr(test, "write_function"), f"No write function found")
self.assertTrue(hasattr(test, "read_function"), f"No read function found")
+ def _get_test_value(self, dims, n_type):
+ """ get numpy array of the test value """
+ return numpy.zeros(dims, n_type)
+
+ def _wrap_dims(self, value, dims):
+ """ Wrap a value in the current number of dimensions """
+ if dims == 0:
+ return value
+ elif dims == 1:
+ return [value]
+ elif dims == 2:
+ return [[value]]
+
+ def _get_mock_value(self, value, n_type):
+ """ get opcua Varianttype array of the test value """
+ return asyncua.ua.uatypes.Variant(Value=value, VariantType=opcua_client.numpy_to_OPCua_dict[n_type])
+
async def test_read(self):
"""
This tests the read functions.
"""
+ async def get_flat_value():
+ return self._get_test_value(j, i.numpy_type).flatten()
+
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()
+ self._get_test_value(j, i.numpy_type)
m_node = asynctest.asynctest.CoroutineMock()
@@ -175,37 +191,28 @@ class TestOPCua(base.AsyncTestCase):
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()))
+ comp = val == self._get_test_value(j, i.numpy_type)
+ self.assertTrue(comp.all(), "Read value unequal to expected value: \n\t{} \n\t{}".format(val, self._get_test_value(j, i.numpy_type)))
async def test_read_unicode(self):
"""
Test whether unicode characters are replaced by '?'.
"""
+ async def get_unicode_value():
+ return self._wrap_dims(b'foo \xef\xbf\xbd bar'.decode('utf-8'), dims)
+
# test 0-2 dimensions of strings
for dims in range(0,2):
- # wrap a value in the current number of dimensions
- def wrap_dims(x):
- if dims == 0:
- return x
- elif dims == 1:
- return [x]
- elif dims == 2:
- return [[x]]
-
- # return a constructed value with unicode
- async def get_value():
- return wrap_dims(b'foo \xef\xbf\xbd bar'.decode('utf-8'))
m_node = asynctest.asynctest.CoroutineMock()
- m_node.get_value = get_value
+ m_node.get_value = get_unicode_value
# create the ProtocolAttribute to test
test = opcua_client.ProtocolAttribute(m_node, 1, 0, opcua_client.numpy_to_OPCua_dict[str])
# check if unicode is replaced by ?
val = await test.read_function()
- self.assertEqual(wrap_dims("foo ? bar"), val)
+ self.assertEqual(self._wrap_dims("foo ? bar", dims), val)
def test_type_map(self):
for numpy_type, opcua_type in opcua_client.numpy_to_OPCua_dict.items():
@@ -249,20 +256,24 @@ class TestOPCua(base.AsyncTestCase):
This allows the code to compare what values we want to write and what values would be given to a server.
"""
+ async def compare_values(val):
+ """ comparison function that replaces `set_data_value` inside the attributes write function """
+ # test valuest
+ val = val.tolist() if type(val) == numpy.ndarray else val
+ if j != DIMENSION_TESTS[0]:
+ comp = val.Value == self._get_mock_value(self._get_test_value(j, i.numpy_type).flatten(), i.numpy_type).Value
+ self.assertTrue(comp.all(),
+ "Array attempting to write unequal to expected array: \n\t got: {} \n\texpected: {}".format(val,self._get_mock_value(self._get_test_value(j, i.numpy_type), i.numpy_type)))
+ else:
+ comp = val == self._get_mock_value(self._get_test_value(j, i.numpy_type), i.numpy_type)
+ self.assertTrue(comp, "value attempting to write unequal to expected value: \n\tgot: {} \n\texpected: {}".format(val, self._get_mock_value(self._get_test_value(j, i.numpy_type), i.numpy_type)))
+
# 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
@@ -271,20 +282,8 @@ class TestOPCua(base.AsyncTestCase):
else:
test = opcua_client.ProtocolAttribute(m_node, j[1], j[0], opcua_client.numpy_to_OPCua_dict[i.numpy_type])
- # 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())
+ await test.write_function(self._get_test_value(j, i.numpy_type))
diff --git a/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py b/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py
index 3c3a38ef679118046e8b47cbfa070bc30ee78373..172cca59236dea2aa41ac098daf2e150a85797e5 100644
--- a/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py
+++ b/tangostationcontrol/tangostationcontrol/test/clients/test_snmp_client.py
@@ -143,6 +143,11 @@ class TestSNMP(base.TestCase):
"""
Attempts to write a value to an SNMP server, but instead intercepts it and compared whether the values is as expected.
"""
+
+ def loop_test(*value):
+ res_lst.append(value[1])
+ return None, None, None, server.get_return_val(i, server.DIM_LIST[j])
+
server = server_imitator()
@@ -161,11 +166,8 @@ class TestSNMP(base.TestCase):
snmp_attr = snmp_attribute(comm=m_comms, mib="test", name="test", idx=0, dtype=server.SNMP_TO_NUMPY_DICT[i], dim_x=server.DIM_LIST[j][0], dim_y=server.DIM_LIST[j][1])
res_lst = []
- def test(*value):
- res_lst.append(value[1])
- return None, None, None, server.get_return_val(i, server.DIM_LIST[j])
- hlapi.ObjectType = test
+ hlapi.ObjectType = loop_test
snmp_attr.write_function(set_val)