From 2668b7d631679e79ab1e923bb5f8084b15d71afe Mon Sep 17 00:00:00 2001
From: Jan David Mol <mol@astron.nl>
Date: Wed, 3 Nov 2021 16:57:06 +0100
Subject: [PATCH] L2SS-470: Turn off a device before testing it. Moved common
device test code to a separate base class
---
.../integration_test/devices/base.py | 56 +++++++++++
.../devices/test_device_boot.py | 60 ++----------
.../devices/test_device_recv.py | 47 +--------
.../devices/test_device_sdp.py | 65 ++-----------
.../devices/test_device_sst.py | 95 ++++---------------
.../devices/test_device_unb2.py | 48 +---------
6 files changed, 97 insertions(+), 274 deletions(-)
create mode 100644 tangostationcontrol/tangostationcontrol/integration_test/devices/base.py
diff --git a/tangostationcontrol/tangostationcontrol/integration_test/devices/base.py b/tangostationcontrol/tangostationcontrol/integration_test/devices/base.py
new file mode 100644
index 000000000..555f7256e
--- /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 2341e1aa5..3a8dc8937 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 6d9353936..26e02ef31 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 f12fc0ae4..2df399ed4 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 210dc93de..38f3528f5 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 4a0382f2a..a35f70add 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")
--
GitLab