L2SS-340: Very early sst device connectivity tests

devices/integration_test/devices/test_device_sst.py

+# -*- 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 socket
+import time
+
+from tango import DeviceProxy
+from tango._tango import DevState
+
+from integration_test import base
+
+
+class TestDeviceSST(base.IntegrationTestCase):
+
+    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 = DeviceProxy("LTS/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 = DeviceProxy("LTS/SST/1")
+
+        self.assertEqual(DevState.OFF, d.state())
+
+    def test_device_sst_initialize(self):
+        """Test if we can transition to standby"""
+
+        d = DeviceProxy("LTS/SST/1")
+
+        d.initialise()
+
+        self.assertEqual(DevState.STANDBY, d.state())
+
+    def test_device_sst_on(self):
+        """Test if we can transition to on"""
+
+        d = DeviceProxy("LTS/SST/1")
+
+        d.initialise()
+
+        d.on()
+
+        self.assertEqual(DevState.ON, d.state())
+
+    def test_device_sst_send_udp(self):
+
+        d = DeviceProxy("LTS/SST/1")
+
+        d.initialise()
+
+        self.assertEqual(DevState.STANDBY, d.state())
+
+        d.on()
+
+        self.assertEqual(DevState.ON, d.state())
+
+        s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s1.connect(("127.0.0.1", 5002))
+
+        # TODO(Corne): Change me into an actual SST packet
+        s1.send("Hello World!".encode("UTF-8"))

devices/test/clients/test_tcp_replicator.py

@@ -49,7 +49,7 @@ class TestTCPReplicator(base.TestCase):
         # Patch _process_queue and force match spec
         process_queue_patcher = mock.patch.object(
             self.m_tcp_replicator, '_process_queue',
-            spec=TCPReplicator._process_queue, return_value=self.dummy_task())
+            autospec=True, return_value=self.dummy_task())
         self.m_process_queue = process_queue_patcher.start()
         self.addCleanup(process_queue_patcher.stop)
 
@@ -155,7 +155,7 @@ class TestTCPReplicator(base.TestCase):
     def test_queue_start(self):
         replicator = self.m_tcp_replicator(self.m_queue)
 
-        self.m_process_queue.assert_called_once_with()
+        self.m_process_queue.assert_called_once_with(replicator)
 
     def test_transmit_queue(self):
         m_data = "Hello World!".encode('utf-8')