Merge branch 'L2SS-452-corba-timeout' into 'master'

L2SS-452: DeviceProxy wrapper for integration tests

Closes L2SS-452

See merge request !162

devices/integration_test/device_proxy.py

+from tango import DeviceProxy
+
+
+class TestDeviceProxy(DeviceProxy):
+
+    def __init__(self, *args, **kwargs):
+        super(TestDeviceProxy, self).__init__(*args, **kwargs)
+        self.set_timeout_millis(10000)

devices/integration_test/devices/test_device_recv.py

@@ -7,11 +7,9 @@
 # Distributed under the terms of the APACHE license.
 # See LICENSE.txt for more info.
 
-import time
-
-from tango import DeviceProxy
 from tango._tango import DevState
 
+from integration_test.device_proxy import TestDeviceProxy
 from integration_test import base
 
 
@@ -22,7 +20,7 @@ class TestDeviceRECV(base.IntegrationTestCase):
 
     def tearDown(self):
         """Turn device Off in teardown to prevent blocking tests"""
-        d = DeviceProxy("LTS/RECV/1")
+        d = TestDeviceProxy("LTS/RECV/1")
 
         try:
             d.Off()
@@ -33,14 +31,14 @@ class TestDeviceRECV(base.IntegrationTestCase):
     def test_device_proxy_recv(self):
         """Test if we can successfully create a DeviceProxy and fetch state"""
 
-        d = DeviceProxy("LTS/RECV/1")
+        d = TestDeviceProxy("LTS/RECV/1")
 
         self.assertEqual(DevState.OFF, d.state())
 
     def test_device_recv_initialize(self):
         """Test if we can transition to standby"""
 
-        d = DeviceProxy("LTS/RECV/1")
+        d = TestDeviceProxy("LTS/RECV/1")
 
         d.Initialise()
 
@@ -49,7 +47,7 @@ class TestDeviceRECV(base.IntegrationTestCase):
     def test_device_recv_on(self):
         """Test if we can transition to on"""
 
-        d = DeviceProxy("LTS/RECV/1")
+        d = TestDeviceProxy("LTS/RECV/1")
 
         d.Initialise()
 

devices/integration_test/devices/test_device_sdp.py

@@ -7,11 +7,9 @@
 # Distributed under the terms of the APACHE license.
 # See LICENSE.txt for more info.
 
-import time
-
-from tango import DeviceProxy
 from tango._tango import DevState
 
+from integration_test.device_proxy import TestDeviceProxy
 from integration_test import base
 
 
@@ -23,7 +21,7 @@ class TestDeviceSDP(base.IntegrationTestCase):
 
     def tearDown(self):
         """Turn device Off in teardown to prevent blocking tests"""
-        d = DeviceProxy("LTS/SDP/1")
+        d = TestDeviceProxy("LTS/SDP/1")
 
         try:
             d.Off()
@@ -34,21 +32,21 @@ class TestDeviceSDP(base.IntegrationTestCase):
     def test_device_proxy_sdp(self):
         """Test if we can successfully create a DeviceProxy and fetch state"""
 
-        d = DeviceProxy("LTS/SDP/1")
+        d = TestDeviceProxy("LTS/SDP/1")
 
         self.assertEqual(DevState.OFF, d.state())
 
     def test_device_sdp_ping(self):
         """Test if we can successfully ping the device server"""
 
-        d = DeviceProxy("LTS/SDP/1")
+        d = TestDeviceProxy("LTS/SDP/1")
 
         self.assertGreater(d.ping(), 0)
 
     def test_device_sdp_initialize(self):
         """Test if we can transition to standby"""
 
-        d = DeviceProxy("LTS/SDP/1")
+        d = TestDeviceProxy("LTS/SDP/1")
 
         d.Initialise()
 
@@ -57,7 +55,7 @@ class TestDeviceSDP(base.IntegrationTestCase):
     def test_device_sdp_on(self):
         """Test if we can transition to on"""
 
-        d = DeviceProxy("LTS/SDP/1")
+        d = TestDeviceProxy("LTS/SDP/1")
 
         d.Initialise()
 
@@ -68,7 +66,7 @@ class TestDeviceSDP(base.IntegrationTestCase):
     def test_device_sdp_read_attribute(self):
         """Test if we can read an attribute obtained over OPC-UA"""
 
-        d = DeviceProxy("LTS/SDP/1")
+        d = TestDeviceProxy("LTS/SDP/1")
 
         d.initialise()
 

devices/integration_test/devices/test_device_sst.py

@@ -10,9 +10,9 @@ import socket
 import sys
 import time
 
-from tango import DeviceProxy
 from tango._tango import DevState
 
+from integration_test.device_proxy import TestDeviceProxy
 from integration_test import base
 
 
@@ -24,7 +24,7 @@ class TestDeviceSST(base.IntegrationTestCase):
 
     def tearDown(self):
         """Turn device Off in teardown to prevent blocking tests"""
-        d = DeviceProxy("LTS/SST/1")
+        d = TestDeviceProxy("LTS/SST/1")
 
         try:
             d.Off()
@@ -35,14 +35,14 @@ class TestDeviceSST(base.IntegrationTestCase):
     def test_device_proxy_sst(self):
         """Test if we can successfully create a DeviceProxy and fetch state"""
 
-        d = DeviceProxy("LTS/SST/1")
+        d = TestDeviceProxy("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 = TestDeviceProxy("LTS/SST/1")
 
         d.initialise()
 
@@ -53,7 +53,7 @@ class TestDeviceSST(base.IntegrationTestCase):
 
         port_property = {"Statistics_Client_TCP_Port": "4999"}
 
-        d = DeviceProxy("LTS/SST/1")
+        d = TestDeviceProxy("LTS/SST/1")
 
         self.assertEqual(DevState.OFF, d.state(),
                          "Prerequisite could not be met "
@@ -72,7 +72,7 @@ class TestDeviceSST(base.IntegrationTestCase):
     def test_device_sst_send_udp(self):
         port_property = {"Statistics_Client_TCP_Port": "4998"}
 
-        d = DeviceProxy("LTS/SST/1")
+        d = TestDeviceProxy("LTS/SST/1")
 
         self.assertEqual(DevState.OFF, d.state(),
                          "Prerequisite could not be met "
@@ -101,7 +101,7 @@ class TestDeviceSST(base.IntegrationTestCase):
 
         m_data = "Hello World!".encode("UTF-8")
 
-        d = DeviceProxy("LTS/SST/1")
+        d = TestDeviceProxy("LTS/SST/1")
 
         self.assertEqual(DevState.OFF, d.state(),
                          "Prerequisite could not be met "

devices/integration_test/devices/test_device_unb2.py

@@ -7,11 +7,9 @@
 # Distributed under the terms of the APACHE license.
 # See LICENSE.txt for more info.
 
-import time
-
-from tango import DeviceProxy
 from tango._tango import DevState
 
+from integration_test.device_proxy import TestDeviceProxy
 from integration_test import base
 
 
@@ -23,7 +21,7 @@ class TestDeviceUNB2(base.IntegrationTestCase):
 
     def tearDown(self):
         """Turn device Off in teardown to prevent blocking tests"""
-        d = DeviceProxy("LTS/UNB2/1")
+        d = TestDeviceProxy("LTS/UNB2/1")
 
         try:
             d.Off()
@@ -34,14 +32,14 @@ class TestDeviceUNB2(base.IntegrationTestCase):
     def test_device_proxy_unb2(self):
         """Test if we can successfully create a DeviceProxy and fetch state"""
 
-        d = DeviceProxy("LTS/UNB2/1")
+        d = TestDeviceProxy("LTS/UNB2/1")
 
         self.assertEqual(DevState.OFF, d.state())
 
     def test_device_unb2_initialize(self):
         """Test if we can transition to standby"""
 
-        d = DeviceProxy("LTS/UNB2/1")
+        d = TestDeviceProxy("LTS/UNB2/1")
 
         d.initialise()
 
@@ -50,7 +48,7 @@ class TestDeviceUNB2(base.IntegrationTestCase):
     def test_device_unb2_on(self):
         """Test if we can transition to on"""
 
-        d = DeviceProxy("LTS/UNB2/1")
+        d = TestDeviceProxy("LTS/UNB2/1")
 
         d.initialise()