From 901c54de2fb4994f9f838d33f6335b4ff3450ac8 Mon Sep 17 00:00:00 2001
From: Jan David Mol <mol@astron.nl>
Date: Wed, 3 Nov 2021 12:41:21 +0100
Subject: [PATCH] L2SS-392: Testing against an actual server is integration,
not unit test
---
.../test_opcua_client_against_server.py | 163 ------------------
1 file changed, 163 deletions(-)
delete mode 100644 tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client_against_server.py
diff --git a/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client_against_server.py b/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client_against_server.py
deleted file mode 100644
index 15bb6fdb1..000000000
--- a/tangostationcontrol/tangostationcontrol/test/clients/test_opcua_client_against_server.py
+++ /dev/null
@@ -1,163 +0,0 @@
-import asyncua
-import numpy
-
-from tangostationcontrol.clients.opcua_client import OPCUAConnection
-
-from tangostationcontrol.test import base
-
-
-class TestClientServer(base.AsyncTestCase):
- """ Test the OPCUAConnection against an OPCUA server we instantiate ourselves. """
-
- async def setup_server(self, port):
- """ Setup a server on a dedicated port for the test, to allow
- the tests to be run in parallel. """
-
- # where we will run the server
- self.endpoint = f"opc.tcp://127.0.0.1:{port}"
- self.namespace = "http://example.com"
-
- # setup an OPC-UA server
- self.server = asyncua.Server()
- await self.server.init()
- self.server.set_endpoint(self.endpoint)
- self.server.set_server_name(f"Test server spawned by {__file__}")
-
- # create an interface
- idx = await self.server.register_namespace(self.namespace)
- obj = self.server.get_objects_node()
-
- # add double_R/double_RW
- double_R = await obj.add_variable(idx, "double_R", 42.0)
- double_RW = await obj.add_variable(idx, "double_RW", 42.0)
- await double_RW.set_writable()
-
- # add methods
- @asyncua.uamethod
- def multiply(parent, x, y):
- self.assertEqual(float, type(x))
- self.assertEqual(int, type(y))
- return x * y
-
- @asyncua.uamethod
- def procedure(parent):
- return
-
- @asyncua.uamethod
- def throws(parent):
- raise Exception("Expected test exception")
-
- multiply_method = await obj.add_method(idx, "multiply", multiply, [asyncua.ua.VariantType.Double, asyncua.ua.VariantType.Int64], [asyncua.ua.VariantType.Double])
- procedure_method = await obj.add_method(idx, "procedure", procedure, [], [])
- throws_method = await obj.add_method(idx, "throws", throws, [], [])
-
- # run the server
- await self.server.start()
-
- async def setUp(self):
- self.server = None
-
- async def tearDown(self):
- if self.server:
- await self.server.stop()
-
- def fault_func(self):
- raise Exception("FAULT")
-
- async def test_opcua_connection(self):
- await self.setup_server(14840)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
- finally:
- await test_client.stop()
-
- async def test_read_attribute(self):
- await self.setup_server(14841)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
-
- # setup the attribute
- class attribute(object):
- dim_x = 1
- dim_y = 0
- numpy_type = numpy.double
-
- prot_attr = await test_client.setup_protocol_attribute(["double_R"], attribute())
-
- # read it from the server
- self.assertEqual(42.0, await prot_attr.read_function())
- finally:
- await test_client.stop()
-
- async def test_write_attribute(self):
- await self.setup_server(14842)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
-
- # setup the attribute
- class attribute(object):
- dim_x = 1
- dim_y = 0
- numpy_type = numpy.double
-
- prot_attr = await test_client.setup_protocol_attribute(["double_RW"], attribute())
-
- # write it to the server and read it back to verify
- await prot_attr.write_function(123.0)
-
- self.assertEqual(123.0, await prot_attr.read_function())
- finally:
- await test_client.stop()
-
- async def test_method_without_args(self):
- await self.setup_server(14843)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
-
- self.assertEqual(None, await test_client._call_method(["procedure"]))
- finally:
- await test_client.stop()
-
- async def test_method_with_args(self):
- await self.setup_server(14843)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
-
- self.assertEqual(21.0, await test_client._call_method(["multiply"], numpy.double(3.0), numpy.int64(7)))
- finally:
- await test_client.stop()
-
- async def test_method_with_wrong_arg_types(self):
- await self.setup_server(14844)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
-
- with self.assertRaises(Exception):
- # correct signature is multiply(double,int64)
- _ = await test_client._call_method(["multiply"], numpy.double(3.0), numpy.double(7))
- finally:
- await test_client.stop()
-
- async def test_errorring_method(self):
- await self.setup_server(14845)
-
- test_client = OPCUAConnection(self.endpoint, self.namespace, 5, self.fault_func, self.loop)
- try:
- await test_client.start()
-
- with self.assertRaises(Exception):
- await test_client._call_method(["throws"])
- finally:
- await test_client.stop()
--
GitLab