Merge branch 'nodb-tests' into 'develop'

Refactor testsuite to test without a tango DB

See merge request taurus-org/taurus!1180

.gitlab-ci.yml

@@ -0,0 +1,9 @@
+nodb:
+  stage: test
+  image:
+    name: registry.gitlab.com/taurus-org/taurus-docker:nodb
+  before_script:
+    - pip install -e .
+  script:
+    - pytest -n3 --forked -v lib/taurus
+

lib/taurus/core/epics/__init__.py

@@ -67,4 +67,10 @@
 Epics records may eventually be mapped as Devices.
 """
 from __future__ import absolute_import
-from .epicsfactory import *
+try:
+    from .epicsfactory import *
+except ImportError as e:
+    # This is a workaround to avoid breaking pytest collection
+    # TODO: the try-except can be removed when moving this to external plugin
+    from taurus import warning
+    warning("epics scheme not available: %r", e)

lib/taurus/core/tango/enums.py

@@ -79,7 +79,7 @@
     PyTango.DevULong: numpy.uint32,
     PyTango.DevLong64: numpy.int64,
     PyTango.DevULong64: numpy.uint64,
-    PyTango.DevString: numpy.str,
+    PyTango.DevString: str,
     PyTango.DevDouble: numpy.float64,
     PyTango.DevFloat: numpy.float32,
 }

lib/taurus/core/tango/tangodevice.py

@@ -208,8 +208,12 @@ def getDisplayValue(self, cache=True):
         return self.stateObj.read(cache=cache).rvalue.name
 
     def _createHWObject(self):
+        name = self.getFullName()
+        if name.startswith('tango-nodb:'):
+            # adapt the taurus full name to tango convention
+            name = 'tango' + name[10:] + '#dbase=no'
         try:
-            return DeviceProxy(self.getFullName())
+            return DeviceProxy(name)
         except DevFailed as e:
             self.warning('Could not create HW object: %s' % (e.args[0].desc))
             self.traceback()

lib/taurus/core/tango/tangofactory.py

@@ -90,9 +90,9 @@ class TangoFactory(Singleton, TaurusFactory, Logger):
           object, thus not being part of the model name itself
     """
 
-    #: the list of schemes that this factory supports. For this factory: 'tango'
-    #: is the only scheme
-    schemes = ("tango",)
+    #: the list of schemes that this factory supports. This factory
+    # supports 'tango' and 'tango-nodb' schemes
+    schemes = ("tango", "tango-nodb")
     caseSensitive = False
     elementTypesMap = {TaurusElementType.Authority: TangoAuthority,
                        TaurusElementType.Device: TangoDevice,
@@ -352,7 +352,10 @@ def getDevice(self, dev_name, create_if_needed=True, **kw):
 
         if d is None:
             try:
-                db = self.getAuthority(groups.get('authority'))
+                if groups['scheme'] == 'tango-nodb':
+                    db = None
+                else:
+                    db = self.getAuthority(groups.get('authority'))
                 dev_klass = self._getDeviceClass(
                     db=db, devname=groups['devname'])
                 kw['storeCallback'] = self._storeDevice

lib/taurus/core/tango/tangovalidator.py

@@ -51,11 +51,12 @@ class TangoAuthorityNameValidator(TaurusAuthorityNameValidator):
      - port: port number
     '''
 
-    scheme = 'tango'
+    scheme = '(tango|tango-nodb)'
     authority = r'//(?P<host>([\w\-_]+\.)*[\w\-_]+):(?P<port>\d{1,5})'
     path = '(?!)'
     query = '(?!)'
     fragment = '(?!)'
+    default_scheme = 'tango'
 
     def getUriGroups(self, name, strict=None):
         '''Reimplementation of getUriGroups to fix the host and authority
@@ -84,12 +85,13 @@ class TangoDeviceNameValidator(TaurusDeviceNameValidator):
     a string if the URI contains it.
     '''
 
-    scheme = 'tango'
+    scheme = '(tango|tango-nodb)'
     authority = TangoAuthorityNameValidator.authority
     path = r'/?(?P<devname>((?P<_devalias>[^/?#:]+)|' + \
            r'(?P<_devslashname>[^/?#:]+/[^/?#:]+/[^/?#:]+)))'
     query = '(?!)'
-    fragment = '(?!)'
+    fragment = '(?P<_fragmentdb>dbase=(yes|no)$)'
+    default_scheme = 'tango'
 
     def getUriGroups(self, name, strict=None):
         '''Reimplementation of getUriGroups to fix the host and authority
@@ -100,6 +102,9 @@ def getUriGroups(self, name, strict=None):
             fqdn = fqdn_no_alias(ret["host"])
             ret["host"] = fqdn
             ret["authority"] = "//{host}:{port}".format(**ret)
+        # use special scheme "tango-nodb" if the name contains #dbase=no
+        if ret is not None and ret.get('_fragmentdb', None) == 'dbase=no':
+            ret['scheme'] = 'tango-nodb'
         return ret
 
     def getNames(self, fullname, factory=None, queryAuth=True):
@@ -115,7 +120,7 @@ def getNames(self, fullname, factory=None, queryAuth=True):
         default_authority = None
         if factory is None:
             from taurus import Factory
-            factory = Factory(scheme=self.scheme)
+            factory = Factory(scheme='tango')
         default_authority = factory.get_default_tango_host()
 
         if default_authority is None:
@@ -130,7 +135,7 @@ def getNames(self, fullname, factory=None, queryAuth=True):
             groups['authority'] = authority = default_authority
 
         db = None
-        if queryAuth:
+        if queryAuth and groups['scheme'] != 'tango-nodb':
             try:
                 db = factory.getAuthority('tango:%s' % authority)
             except:
@@ -151,13 +156,13 @@ def getNames(self, fullname, factory=None, queryAuth=True):
             return None, None, _devalias
 
         # we can now construct everything. First the complete:
-        complete = 'tango:%(authority)s/%(_devslashname)s' % groups
+        complete = '{scheme}:{authority}/{_devslashname}'.format(**groups)
 
         # then the normal
         if authority.lower() == default_authority.lower():
-            normal = '%(_devslashname)s' % groups
+            normal = '{_devslashname}'.format(**groups)
         else:
-            normal = '%(authority)s/%(_devslashname)s' % groups
+            normal = '{authority}/{_devslashname}'.format(**groups)
 
         # and finally the short
         if _devalias is not None:
@@ -209,12 +214,14 @@ class TangoAttributeNameValidator(TaurusAttributeNameValidator):
     Note: brackets on the group name indicate that this group will only contain
     a string if the URI contains it.
     '''
-    scheme = 'tango'
+    scheme = '(tango|tango-nodb)'
     authority = TangoAuthorityNameValidator.authority
-    path = ('(?P<attrname>%s/(?P<_shortattrname>[^/?:#]+))' %
-            TangoDeviceNameValidator.path)
+    path = '(?P<attrname>{0}/(?P<_shortattrname>[^/?:#]+))'.format(
+        TangoDeviceNameValidator.path)
     query = '(?!)'
-    fragment = '(?P<cfgkey>[^# ]*)'
+    fragment = '((?P<_fragmentdb>dbase=(yes|no)$)|(?P<cfgkey>[^# ]*))'
+
+    default_scheme = 'tango'
 
     def getUriGroups(self, name, strict=None):
         '''Reimplementation of getUriGroups to fix the host and authority
@@ -225,6 +232,9 @@ def getUriGroups(self, name, strict=None):
             fqdn = fqdn_no_alias(ret["host"])
             ret["host"] = fqdn
             ret["authority"] = "//{host}:{port}".format(**ret)
+        # use special scheme "tango-nodb" if the name contains #dbase=no
+        if ret is not None and ret.get('_fragmentdb', None) == 'dbase=no':
+            ret['scheme'] = 'tango-nodb'
         return ret
 
     def getNames(self, fullname, factory=None, queryAuth=True, fragment=False):
@@ -242,13 +252,13 @@ def getNames(self, fullname, factory=None, queryAuth=True, fragment=False):
         devcomplete, devnormal, _ = v.getNames(devname, factory=factory,
                                                queryAuth=queryAuth)
         if devcomplete is not None:
-            complete = '%s/%s' % (devcomplete, short)
+            complete = '{}/{}'.format(devcomplete, short)
         if devnormal is not None:
-            normal = '%s/%s' % (devnormal, short)
+            normal = '{}/{}'.format(devnormal, short)
 
-        # return fragment if requested
+        # return fragment if requested (ignores the special #dbase=yes|no case)
         if fragment:
-            key = groups.get('fragment', None)
+            key = groups.get('cfgkey', None)
             return complete, normal, short, key
 
         return complete, normal, short

lib/taurus/core/tango/test/__init__.py

@@ -24,4 +24,8 @@
 #############################################################################
 
 from __future__ import absolute_import
-from .tgtestds import TangoSchemeTestLauncher, taurus_test_ds
+from .nodb import NamedDeviceTestContext, nodb_dev
+from .res.TangoSchemeTest import TangoSchemeTest
+
+# TODO: remove the following import of deprecated
+from .tgtestds import TangoSchemeTestLauncher, taurus_test_ds  # deprecated

lib/taurus/core/tango/test/nodb.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python
+
+#############################################################################
+##
+# This file is part of Taurus
+##
+# http://taurus-scada.org
+##
+# Copyright 2011 CELLS / ALBA Synchrotron, Bellaterra, Spain
+##
+# Taurus is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+##
+# Taurus is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+##
+# You should have received a copy of the GNU Lesser General Public License
+# along with Taurus.  If not, see <http://www.gnu.org/licenses/>.
+##
+#############################################################################
+
+"""Module containing base classes for database-less tango tests"""
+
+
+import pytest
+from .res.TangoSchemeTest import TangoSchemeTest
+from tango.test_context import DeviceTestContext
+from taurus.core.tango.tangovalidator import TangoDeviceNameValidator
+
+__all__ = [
+    'nodb_dev',
+    'NamedDeviceTestContext'
+]
+
+__docformat__ = 'restructuredtext'
+
+
+class NamedDeviceTestContext(DeviceTestContext):
+    """
+    Reimplementation of :class:`tango.test_context.DeviceTestContext that
+    returns the taurus full name of the Deviceinstead of a
+    :class:`tango.DeviceProxy`
+    """
+    def __enter__(self):
+        if not self.thread.is_alive():
+            self.start()
+        v = TangoDeviceNameValidator()
+        fullname, _, _ = v.getNames(self.get_device_access())
+        return fullname
+
+
+@pytest.fixture(scope="module")
+def nodb_dev():
+    """
+    A pytest fixture that launches TangoSchemeTest for the test
+    It provides the device name as the fixture value.
+
+    Usage::
+        from taurus.core.tango.test import nodb_dev
+
+        def test_foo(nodb_dev):
+            import taurus
+            d = taurus.Device(nodb_dev)
+            assert d["string_scalar"].rvalue == "hello world"
+
+    """
+    with NamedDeviceTestContext(
+            TangoSchemeTest, process=True, timeout=15) as full_name:
+        yield full_name