diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/base.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/base.py
new file mode 100644
index 0000000000000000000000000000000000000000..555f7256ea49d68465b1c45bf038d47b39beeb25
--- /dev/null
+++ b/tangostationcontrol/tangostationcontrol/integration_test/devices/base.py
@@ -0,0 +1,56 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+import time
+import unittest
+
+from tango._tango import DevState
+
+from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
+from tangostationcontrol.integration_test import base
+
+class AbstractTestBases:
+ """ Holder for abstract test base classes. If we define these at the top level,
+ the test runner will execute them. """
+
+ class TestDeviceBase(base.IntegrationTestCase):
+ __test__ = False
+
+ def setUp(self, name = ""):
+ #if name == "":
+ # raise unittest.SkipTest("This is a base class for other tests")
+
+ """Intentionally recreate the device object in each test"""
+ # create a proxy
+ self.name = name
+ self.proxy = TestDeviceProxy(self.name)
+
+ # make sure the device starts in Off
+ self.proxy.Off()
+
+ super().setUp()
+
+ def tearDown(self):
+ """Turn device Off in teardown to prevent blocking tests"""
+ self.proxy.Off()
+
+ def test_device_initialize(self):
+ """Test if we can transition to standby"""
+
+ self.proxy.initialise()
+
+ self.assertEqual(DevState.STANDBY, self.proxy.state())
+
+ def test_device_on(self):
+ """Test if we can transition to on"""
+
+ self.proxy.initialise()
+ self.proxy.on()
+
+ self.assertEqual(DevState.ON, self.proxy.state())
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_boot.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_boot.py
index 2341e1aa5f14a88695b056377d88e5174ffc32b8..3a8dc89374d79b60a5d8f0db0f6240df57899105 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_boot.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_boot.py
@@ -9,69 +9,25 @@
import time
-from tango._tango import DevState
+from .base import AbstractTestBases
-from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
-from tangostationcontrol.integration_test import base
-
-
-class TestDeviceBoot(base.IntegrationTestCase):
+class TestDeviceBoot(AbstractTestBases.TestDeviceBase):
def setUp(self):
- """Intentionally recreate the device object in each test"""
- super(TestDeviceBoot, self).setUp()
-
- def tearDown(self):
- """Turn device Off in teardown to prevent blocking tests"""
- d = TestDeviceProxy("STAT/Boot/1")
-
- try:
- d.Off()
- except Exception as e:
- """Failing to turn Off devices should not raise errors here"""
- print(f"Failed to turn device off in teardown {e}")
-
- def test_device_proxy_unb2(self):
- """Test if we can successfully create a DeviceProxy and fetch state"""
-
- d = TestDeviceProxy("STAT/Boot/1")
-
- self.assertEqual(DevState.OFF, d.state())
-
- def test_device_unb2_initialize(self):
- """Test if we can transition to standby"""
-
- d = TestDeviceProxy("STAT/Boot/1")
-
- d.initialise()
-
- self.assertEqual(DevState.STANDBY, d.state())
-
- def test_device_unb2_on(self):
- """Test if we can transition to on"""
-
- d = TestDeviceProxy("STAT/Boot/1")
-
- d.initialise()
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ super().setUp("STAT/Boot/1")
def test_device_unb2_initialise_station(self):
"""Test if we can initialise the station"""
- d = TestDeviceProxy("STAT/Boot/1")
-
- d.initialise()
- d.on()
+ self.proxy.initialise()
+ self.proxy.on()
- d.initialise_station()
+ self.proxy.boot()
# wait for a few seconds for the station to initialise
timeout = 10
- while d.initialising_station_R and timeout:
+ while self.proxy.booting_R and timeout:
time.sleep(1)
# check whether initialisation succeeded
- self.assertEqual(100, d.initialisation_progress_R, msg=f"Initialisation of station failed. Status: {d.initialisation_status_R}")
+ self.assertEqual(100, self.proxy.progress_R, msg=f"Initialisation of station failed. Status: {self.proxy.status_R}")
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_recv.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_recv.py
index 6d9353936537a116442409b9ce7c343fa4d803b1..26e02ef312214df4a9304eafd050ea1438002da2 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_recv.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_recv.py
@@ -7,50 +7,9 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from tango._tango import DevState
+from .base import AbstractTestBases
-from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
-from tangostationcontrol.integration_test import base
-
-
-class TestDeviceRECV(base.IntegrationTestCase):
+class TestDeviceRECV(AbstractTestBases.TestDeviceBase):
def setUp(self):
- super(TestDeviceRECV, self).setUp()
-
- def tearDown(self):
- """Turn device Off in teardown to prevent blocking tests"""
- d = TestDeviceProxy("STAT/RECV/1")
-
- try:
- d.Off()
- except Exception as e:
- """Failing to turn Off devices should not raise errors here"""
- print(f"Failed to turn device off in teardown {e}")
-
- def test_device_proxy_recv(self):
- """Test if we can successfully create a DeviceProxy and fetch state"""
-
- d = TestDeviceProxy("STAT/RECV/1")
-
- self.assertEqual(DevState.OFF, d.state())
-
- def test_device_recv_initialize(self):
- """Test if we can transition to standby"""
-
- d = TestDeviceProxy("STAT/RECV/1")
-
- d.Initialise()
-
- self.assertEqual(DevState.STANDBY, d.state())
-
- def test_device_recv_on(self):
- """Test if we can transition to on"""
-
- d = TestDeviceProxy("STAT/RECV/1")
-
- d.Initialise()
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ super().setUp("STAT/RECV/1")
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sdp.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sdp.py
index f12fc0ae4f9094c6c82a2dd9076f49cea7fd7ef1..2df399ed4607ea802abcf647c02078193f1b7f03 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sdp.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sdp.py
@@ -7,70 +7,17 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from tango import DeviceProxy
-from tango._tango import DevState
+from .base import AbstractTestBases
-from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
-from tangostationcontrol.integration_test import base
-
-
-class TestDeviceSDP(base.IntegrationTestCase):
+class TestDeviceSDP(AbstractTestBases.TestDeviceBase):
def setUp(self):
- """Intentionally recreate the device object in each test"""
- super(TestDeviceSDP, self).setUp()
-
- def tearDown(self):
- """Turn device Off in teardown to prevent blocking tests"""
- d = TestDeviceProxy("STAT/SDP/1")
-
- try:
- d.Off()
- except Exception as e:
- """Failing to turn Off devices should not raise errors here"""
- print(f"Failed to turn device off in teardown {e}")
-
- def test_device_proxy_sdp(self):
- """Test if we can successfully create a DeviceProxy and fetch state"""
-
- d = TestDeviceProxy("STAT/SDP/1")
-
- self.assertEqual(DevState.OFF, d.state())
-
- def test_device_sdp_ping(self):
- """Test if we can successfully ping the device server"""
-
- d = TestDeviceProxy("STAT/SDP/1")
-
- self.assertGreater(d.ping(), 0)
-
- def test_device_sdp_initialize(self):
- """Test if we can transition to standby"""
-
- d = TestDeviceProxy("STAT/SDP/1")
-
- d.Initialise()
-
- self.assertEqual(DevState.STANDBY, d.state())
-
- def test_device_sdp_on(self):
- """Test if we can transition to on"""
-
- d = TestDeviceProxy("STAT/SDP/1")
-
- d.Initialise()
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ super().setUp("STAT/SDP/1")
def test_device_sdp_read_attribute(self):
"""Test if we can read an attribute obtained over OPC-UA"""
- d = TestDeviceProxy("STAT/SDP/1")
-
- d.initialise()
-
- d.on()
+ self.proxy.initialise()
+ self.proxy.on()
- self.assertListEqual([True]*16, list(d.TR_fpga_communication_error_R))
+ self.assertListEqual([True]*16, list(self.proxy.TR_fpga_communication_error_R))
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sst.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sst.py
index 210dc93defc98c51f9fb9a2b0497a52e5547e5a7..38f3528f531660704baa00f70a7073974256a19f 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sst.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_sst.py
@@ -1,3 +1,4 @@
+
# -*- coding: utf-8 -*-
#
# This file is part of the LOFAR 2.0 Station Software
@@ -12,81 +13,36 @@ import time
from tango._tango import DevState
-from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
-from tangostationcontrol.integration_test import base
-
+from .base import AbstractTestBases
-class TestDeviceSST(base.IntegrationTestCase):
+class TestDeviceSST(AbstractTestBases.TestDeviceBase):
def setUp(self):
- """Intentionally recreate the device object in each test"""
- super(TestDeviceSST, self).setUp()
-
- def tearDown(self):
- """Turn device Off in teardown to prevent blocking tests"""
- d = TestDeviceProxy("STAT/SST/1")
-
- try:
- d.Off()
- except Exception as e:
- """Failing to turn Off devices should not raise errors here"""
- print(f"Failed to turn device off in teardown {e}")
-
- def test_device_proxy_sst(self):
- """Test if we can successfully create a DeviceProxy and fetch state"""
-
- d = TestDeviceProxy("STAT/SST/1")
-
- self.assertEqual(DevState.OFF, d.state())
+ super().setUp("STAT/SST/1")
- def test_device_sst_initialize(self):
- """Test if we can transition to standby"""
-
- d = TestDeviceProxy("STAT/SST/1")
-
- d.initialise()
-
- self.assertEqual(DevState.STANDBY, d.state())
-
- def test_device_sst_on(self):
+ def test_device_on(self):
"""Test if we can transition to on"""
port_property = {"Statistics_Client_TCP_Port": "4999"}
+ self.proxy.put_property(port_property)
+ self.proxy.initialise()
- d = TestDeviceProxy("STAT/SST/1")
-
- self.assertEqual(DevState.OFF, d.state(),
- "Prerequisite could not be met "
- "this test can not continue")
-
- d.put_property(port_property)
+ self.assertEqual(DevState.STANDBY, self.proxy.state())
- d.initialise()
+ self.proxy.on()
- self.assertEqual(DevState.STANDBY, d.state())
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ self.assertEqual(DevState.ON, self.proxy.state())
def test_device_sst_send_udp(self):
port_property = {"Statistics_Client_TCP_Port": "4998"}
+ self.proxy.put_property(port_property)
+ self.proxy.initialise()
- d = TestDeviceProxy("STAT/SST/1")
-
- self.assertEqual(DevState.OFF, d.state(),
- "Prerequisite could not be met "
- "this test can not continue")
-
- d.put_property(port_property)
+ self.assertEqual(DevState.STANDBY, self.proxy.state())
- d.initialise()
+ self.proxy.on()
- self.assertEqual(DevState.STANDBY, d.state())
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ self.assertEqual(DevState.ON, self.proxy.state())
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s1.connect(("device-sst", 5001))
@@ -98,24 +54,14 @@ class TestDeviceSST(base.IntegrationTestCase):
def test_device_sst_connect_tcp_receive(self):
port_property = {"Statistics_Client_TCP_Port": "5101"}
+ self.proxy.put_property(port_property)
+ self.proxy.initialise()
- m_data = "Hello World!".encode("UTF-8")
+ self.assertEqual(DevState.STANDBY, self.proxy.state())
- d = TestDeviceProxy("STAT/SST/1")
+ self.proxy.on()
- self.assertEqual(DevState.OFF, d.state(),
- "Prerequisite could not be met "
- "this test can not continue")
-
- d.put_property(port_property)
-
- d.initialise()
-
- self.assertEqual(DevState.STANDBY, d.state())
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ self.assertEqual(DevState.ON, self.proxy.state())
time.sleep(2)
@@ -128,6 +74,7 @@ class TestDeviceSST(base.IntegrationTestCase):
time.sleep(2)
# TODO(Corne): Change me into an actual SST packet
+ m_data = "Hello World!".encode("UTF-8")
s1.send(m_data)
time.sleep(2)
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_unb2.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_unb2.py
index 4a0382f2aea7c370562ff4d23e37c85c7963a436..a35f70adde7af58408882e3b5ee3256a4838db84 100644
--- a/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_unb2.py
+++ b/tangostationcontrol/tangostationcontrol/integration_test/devices/test_device_unb2.py
@@ -7,51 +7,9 @@
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
-from tango._tango import DevState
+from .base import AbstractTestBases
-from tangostationcontrol.integration_test.device_proxy import TestDeviceProxy
-from tangostationcontrol.integration_test import base
-
-
-class TestDeviceUNB2(base.IntegrationTestCase):
+class TestDeviceUNB2(AbstractTestBases.TestDeviceBase):
def setUp(self):
- """Intentionally recreate the device object in each test"""
- super(TestDeviceUNB2, self).setUp()
-
- def tearDown(self):
- """Turn device Off in teardown to prevent blocking tests"""
- d = TestDeviceProxy("STAT/UNB2/1")
-
- try:
- d.Off()
- except Exception as e:
- """Failing to turn Off devices should not raise errors here"""
- print(f"Failed to turn device off in teardown {e}")
-
- def test_device_proxy_unb2(self):
- """Test if we can successfully create a DeviceProxy and fetch state"""
-
- d = TestDeviceProxy("STAT/UNB2/1")
-
- self.assertEqual(DevState.OFF, d.state())
-
- def test_device_unb2_initialize(self):
- """Test if we can transition to standby"""
-
- d = TestDeviceProxy("STAT/UNB2/1")
-
- d.initialise()
-
- self.assertEqual(DevState.STANDBY, d.state())
-
- def test_device_unb2_on(self):
- """Test if we can transition to on"""
-
- d = TestDeviceProxy("STAT/UNB2/1")
-
- d.initialise()
-
- d.on()
-
- self.assertEqual(DevState.ON, d.state())
+ super().setUp("STAT/UNB2/1")