[pytango] 63/122: Move test utils in the tango package
Sandor Bodo-Merle
sbodomerle-guest at moszumanska.debian.org
Thu Sep 28 19:18:18 UTC 2017
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 cc16d0b4ded42b8bdfb123ec3def1b48a74e37b9
Author: Vincent Michel <vincent.michel at maxlab.lu.se>
Date: Wed Oct 12 16:24:56 2016 +0200
Move test utils in the tango package
---
test/context.py => tango/test_context.py | 100 +++++++++++++++++++++++++------
tango/test_utils.py | 63 +++++++++++++++++++
2 files changed, 146 insertions(+), 17 deletions(-)
diff --git a/test/context.py b/tango/test_context.py
similarity index 63%
rename from test/context.py
rename to tango/test_context.py
index c3073e4..9f2dc97 100644
--- a/test/context.py
+++ b/tango/test_context.py
@@ -1,32 +1,44 @@
-"""Contain the context to run a device without a database."""
+"""Provide a context to run a device without a database."""
+
+from __future__ import absolute_import
+
+__all__ = ["DeviceTestContext", "run_device_test_context"]
# Imports
import os
+import time
+import socket
import platform
import tempfile
-from socket import socket
-from functools import wraps
-from time import sleep, time
+import functools
+
+# Concurrency imports
from threading import Thread
from multiprocessing import Process
-# PyTango imports
-from tango.server import run
-from tango import DeviceProxy, Database, ConnectionFailed, DevFailed
+# CLI imports
+from ast import literal_eval
+from importlib import import_module
+from argparse import ArgumentParser
+# Local imports
+from .server import run
+from . import DeviceProxy, Database, ConnectionFailed, DevFailed
+
+
+# Helpers
-# Retry decorator
def retry(period, errors, pause=0.001):
"""Retry decorator."""
errors = tuple(errors)
def dec(func):
- @wraps(func)
+ @functools.wraps(func)
def wrapper(*args, **kwargs):
- stop = time() + period
+ stop = time.time() + period
first = True
- while first or time() < stop:
- sleep(pause)
+ while first or time.time() < stop:
+ time.sleep(pause)
try:
return func(*args, **kwargs)
except errors as exc:
@@ -37,17 +49,28 @@ def retry(period, errors, pause=0.001):
return dec
-# Get available port
def get_port():
- sock = socket()
+ sock = socket.socket()
sock.bind(('', 0))
res = sock.getsockname()[1]
del sock
return res
-# No database Tango context
-class TangoTestContext(object):
+def literal_dict(arg):
+ return dict(literal_eval(arg))
+
+
+def device(path):
+ """Get the device class from a given module."""
+ module_name, device_name = path.rsplit(".", 1)
+ module = import_module(module_name)
+ return getattr(module, device_name)
+
+
+# Device test context
+
+class DeviceTestContext(object):
""" Context to run a device without a database."""
nodb = "#dbase=no"
@@ -56,7 +79,7 @@ class TangoTestContext(object):
def __init__(self, device, device_cls=None, server_name=None,
instance_name=None, device_name=None, properties={},
- db=None, port=0, debug=0, daemon=False, process=False):
+ db=None, port=0, debug=5, daemon=False, process=False):
"""Inititalize the context to run a given device."""
# Argument
tangoclass = device.__name__
@@ -157,3 +180,46 @@ class TangoTestContext(object):
def __exit__(self, exc_type, exception, trace):
"""Exit method for context support."""
self.stop()
+
+
+# Command line interface
+
+def parse_command_line_args(args=None):
+ """Parse arguments given in command line."""
+ desc = "Run a given device on a given port."
+ parser = ArgumentParser(description=desc)
+ # Add arguments
+ msg = 'The device to run as a python path.'
+ parser.add_argument('device', metavar='DEVICE',
+ type=device, help=msg)
+ msg = "The port to use."
+ parser.add_argument('--port', metavar='PORT',
+ type=int, help=msg, default=0)
+ msg = "The debug level."
+ parser.add_argument('--debug', metavar='DEBUG',
+ type=int, help=msg, default=0)
+ msg = "The properties to set as python dict."
+ parser.add_argument('--prop', metavar='PROP',
+ type=literal_dict, help=msg, default='{}')
+ # Parse arguments
+ namespace = parser.parse_args(args)
+ return namespace.device, namespace.port, namespace.prop, namespace.debug
+
+
+def run_device_test_context(args=None):
+ device, port, properties, debug = parse_command_line_args(args)
+ context = DeviceTestContext(
+ device, properties=properties, port=port, debug=debug)
+ context.start()
+ msg = '{0} started on port {1} with properties {2}'
+ print(msg.format(device.__name__, context.port, properties))
+ print('Device access: {}'.format(context.get_device_access()))
+ print('Server access: {}'.format(context.get_server_access()))
+ context.join()
+ print("Done")
+
+
+# Main execution
+
+if __name__ == "__main__":
+ run_device_test_context()
diff --git a/tango/test_utils.py b/tango/test_utils.py
new file mode 100644
index 0000000..91e7ed3
--- /dev/null
+++ b/tango/test_utils.py
@@ -0,0 +1,63 @@
+"""Test utilities"""
+
+__all__ = ['DeviceTestContext', 'SimpleDevice']
+
+# Imports
+from six import add_metaclass
+
+# Local imports
+from . import utils
+from . import DevState, CmdArgType, GreenMode
+from .server import Device, DeviceMeta
+from .test_context import DeviceTestContext
+
+# Conditional imports
+try:
+ import pytest
+except ImportError:
+ pytest = None
+
+
+# Test devices
+
+ at add_metaclass(DeviceMeta)
+class SimpleDevice(Device):
+ def init_device(self):
+ self.set_state(DevState.ON)
+
+
+# Pytest fixtures
+
+if pytest:
+
+ @pytest.fixture(params=DevState.values.values())
+ def state(request):
+ return request.param
+
+ @pytest.fixture(params=utils._scalar_types)
+ def typed_values(request):
+ dtype = request.param
+ # Unsupported types
+ if dtype in [CmdArgType.DevInt, CmdArgType.ConstDevString,
+ CmdArgType.DevEncoded, CmdArgType.DevUChar]:
+ pytest.xfail('Should we support those types?')
+ # Supported types
+ if dtype in utils._scalar_str_types:
+ return dtype, ['hey hey', 'my my']
+ if dtype in utils._scalar_bool_types:
+ return dtype, [False, True]
+ if dtype in utils._scalar_int_types:
+ return dtype, [1, 2]
+ if dtype in utils._scalar_float_types:
+ return dtype, [2.71, 3.14]
+
+ @pytest.fixture(params=GreenMode.values.values())
+ def green_mode(request):
+ return request.param
+
+ @pytest.fixture(params=[
+ GreenMode.Synchronous,
+ GreenMode.Asyncio,
+ GreenMode.Gevent])
+ def server_green_mode(request):
+ return request.param
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/debian-science/packages/pytango.git
More information about the debian-science-commits
mailing list