Merge branch 'master' into L2SS-843

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

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)