Skip to content
Snippets Groups Projects
Commit 2668b7d6 authored by Jan David Mol's avatar Jan David Mol
Browse files

L2SS-470: Turn off a device before testing it. Moved common device test code...

L2SS-470: Turn off a device before testing it. Moved common device test code to a separate base class
parent ad37b39d
No related branches found
No related tags found
1 merge request!176L2SS-470: Fix boot
# -*- 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())
......@@ -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}")
......@@ -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")
......@@ -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))
# -*- 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)
......
......@@ -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")
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment