[pytango] 75/122: add basic client tests

[Sandor Bodo-Merle]

This is an automated email from the git hooks/post-receive script.

sbodomerle-guest pushed a commit to tag v9.2.1
in repository pytango.

commit e314ed4170dee41290595a73f2d67b470e8f1ba3
Author: Johan Forsberg <johan.forsberg at maxiv.lu.se>
Date:   Tue Oct 18 17:38:35 2016 +0200

    add basic client tests
---
 test/test_client.py | 221 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 221 insertions(+)

diff --git a/test/test_client.py b/test/test_client.py
new file mode 100644
index 0000000..261918c
--- /dev/null
+++ b/test/test_client.py
@@ -0,0 +1,221 @@
+"""Client tests that run against the standard TangoTest device.
+
+Due to a TANGO 9 issue (#821), the device is run without any database.
+Note that this means that various features won't work:
+
+ * No device configuration via properties.
+ * No event generated by the server.
+ * No memorized attributes.
+ * No device attribute configuration via the database.
+
+So don't even try to test anything of the above as it will not work
+and is even likely to crash the device (!)
+
+"""
+
+from distutils.spawn import find_executable
+from subprocess import Popen
+import platform
+from time import sleep
+
+import psutil
+import pytest
+from tango import DeviceProxy, DevFailed
+from tango import DeviceInfo, AttributeInfo, AttributeInfoEx
+from tango.utils import is_str_type, is_int_type, is_float_type, is_bool_type
+
+
+ATTRIBUTES = [
+    'ampli',
+    'boolean_scalar',
+    'double_scalar',
+    'double_scalar_rww',
+    'double_scalar_w',
+    'float_scalar',
+    'long64_scalar',
+    'long_scalar',
+    'long_scalar_rww',
+    'long_scalar_w',
+    'no_value',
+    'short_scalar',
+    'short_scalar_ro',
+    'short_scalar_rww',
+    'short_scalar_w',
+    'string_scalar',
+    'throw_exception',
+    'uchar_scalar',
+    'ulong64_scalar',
+    'ushort_scalar',
+    'ulong_scalar',
+    'boolean_spectrum',
+    'boolean_spectrum_ro',
+    'double_spectrum',
+    'double_spectrum_ro',
+    'float_spectrum',
+    'float_spectrum_ro',
+    'long64_spectrum_ro',
+    'long_spectrum',
+    'long_spectrum_ro',
+    'short_spectrum',
+    'short_spectrum_ro',
+    'string_spectrum',
+    'string_spectrum_ro',
+    'uchar_spectrum',
+    'uchar_spectrum_ro',
+    'ulong64_spectrum_ro',
+    'ulong_spectrum_ro',
+    'ushort_spectrum',
+    'ushort_spectrum_ro',
+    'boolean_image',
+    'boolean_image_ro',
+    'double_image',
+    'double_image_ro',
+    'float_image',
+    'float_image_ro',
+    'long64_image_ro',
+    'long_image',
+    'long_image_ro',
+    'short_image',
+    'short_image_ro',
+    'string_image',
+    'string_image_ro',
+    'uchar_image',
+    'uchar_image_ro',
+    'ulong64_image_ro',
+    'ulong_image_ro',
+    'ushort_image',
+    'ushort_image_ro',
+    'wave',
+    'State',
+    'Status',
+]
+
+
+# Helpers
+
+def get_ports(pid):
+    p = psutil.Process(pid)
+    conns = p.connections(kind="tcp")
+    # Sorting by family in order to make any IPv6 address go first.
+    # Otherwise there's a 50% chance that the proxy will just
+    # hang (presumably because it's connecting on the wrong port)
+    # This works on my machine, not sure if it's a general
+    # solution though.
+    conns = reversed(sorted(conns, key=lambda c: c.family))
+    return [c.laddr[1] for c in conns]
+
+
+def start_server(server, inst, device):
+    exe = find_executable(server)
+    cmd = ("{0} {1} -ORBendPoint giop:tcp::0 -nodb -dlist {2}"
+           .format(exe, inst, device))
+    proc = Popen(cmd.split())
+    proc.poll()
+    return proc
+
+
+def get_proxy(host, port, device):
+    access = "tango://{0}:{1}/{2}#dbase=no".format(
+        host, port, device)
+    return DeviceProxy(access)
+
+
+def wait_for_proxy(host, proc, device, retries=50, delay=0.1):
+    for i in range(retries):
+        ports = get_ports(proc.pid)
+        for port in ports:
+            try:
+                proxy = get_proxy(host, port, device)
+                proxy.ping()
+                return proxy
+            except DevFailed:
+                pass
+        sleep(delay)
+    else:
+        raise RuntimeError("TangoTest device did not start up!")
+
+
+# Fixtures
+
+ at pytest.fixture(scope="module")
+def tango_test(request):
+    server = "TangoTest"
+    inst = "test"
+    device = "sys/tg_test/17"
+    host = platform.node()
+    proc = start_server(server, inst, device)
+    proxy = wait_for_proxy(host, proc, device)
+
+    yield proxy
+
+    proc.terminate()
+    # let's not wait for it to exit, that takes too long :)
+
+
+ at pytest.fixture(params=ATTRIBUTES)
+def attribute(request):
+    return request.param
+
+
+ at pytest.fixture(params=[a for a in ATTRIBUTES
+                        if a not in ("no_value", "throw_exception")])
+def readable_attribute(request):
+    return request.param
+
+
+ at pytest.fixture(params=[a for a in ATTRIBUTES
+                        if "scalar" in a and
+                        a.split("_")[-1] not in ("ro", "rww")])
+def writable_scalar_attribute(request):
+    return request.param
+
+
+# Tests
+
+def test_ping(tango_test):
+    duration = tango_test.ping()
+    assert isinstance(duration, int)
+
+
+def test_info(tango_test):
+    info = tango_test.info()
+    assert isinstance(info, DeviceInfo)
+    assert info.dev_class == "TangoTest"
+    # ...
+
+
+def test_read_attribute(tango_test, readable_attribute):
+    "Check that readable attributes can be read"
+    tango_test.read_attribute(readable_attribute)
+
+
+def test_write_scalar_attribute(tango_test, writable_scalar_attribute):
+    "Check that writable scalar attributes can be written"
+    attr_name = writable_scalar_attribute
+    config = tango_test.get_attribute_config(writable_scalar_attribute)
+    if is_bool_type(config.data_type):
+        tango_test.write_attribute(attr_name, True)
+    elif is_int_type(config.data_type):
+        tango_test.write_attribute(attr_name, 76)
+    elif is_float_type(config.data_type):
+        tango_test.write_attribute(attr_name, -28.2)
+    elif is_str_type(config.data_type):
+        tango_test.write_attribute(attr_name, "hello")
+    else:
+        pytest.xfail("Not currently testing this type")
+
+
+def test_read_attribute_config(tango_test, attribute):
+    tango_test.get_attribute_config(attribute)
+
+
+def test_attribute_list_query(tango_test):
+    attrs = tango_test.attribute_list_query()
+    assert all(isinstance(a, AttributeInfo) for a in attrs)
+    assert set(a.name for a in attrs) == set(ATTRIBUTES)
+
+
+def test_attribute_list_query_ex(tango_test):
+    attrs = tango_test.attribute_list_query_ex()
+    assert all(isinstance(a, AttributeInfoEx) for a in attrs)
+    assert set(a.name for a in attrs) == set(ATTRIBUTES)

-- 
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/debian-science/packages/pytango.git