Source code for taurus.core.tango.tangovalidator

#!/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/>.
#
# ###########################################################################

"""This module contains the base taurus name validator classes
"""

__all__ = [
    "TangoAuthorityNameValidator",
    "TangoDeviceNameValidator",
    "TangoAttributeNameValidator",
]

__docformat__ = "restructuredtext"

from taurus.core.taurusvalidator import (
    TaurusAttributeNameValidator,
    TaurusDeviceNameValidator,
    TaurusAuthorityNameValidator,
)
from taurus.core.util.fqdn import fqdn_no_alias


# todo: I do not understand the behaviour of getNames for Auth, Dev and Attr in
#      the case when the fullname does not match the regexp. For Auth it
#      returns a 3-tuple, for devs a 2-tuple and for attrs and conf a single
#      None. This is not coherent to what the method returns when it matches
#      the regexp (always a 3-tuple)


[docs]class TangoAuthorityNameValidator(TaurusAuthorityNameValidator): """Validator for Tango authority names. Apart from the standard named groups (scheme, authority, path, query and fragment), the following named groups are created: - host: tango host name, without port. - port: port number """ scheme = "(tango|tango-nodb)" authority = r"//(?P<host>([\w\-_]+\.)*[\w\-_]+):(?P<port>\d{1,5})" path = "(?!)" query = "(?!)" fragment = "(?!)" default_scheme = "tango"
[docs] def getUriGroups(self, name, strict=None): """Reimplementation of getUriGroups to fix the host and authority name using fully qualified domain name for the host. """ ret = TaurusAuthorityNameValidator.getUriGroups(self, name, strict) if ret is not None: fqdn = fqdn_no_alias(ret["host"]) ret["host"] = fqdn ret["authority"] = "//{host}:{port}".format(**ret) return ret
[docs]class TangoDeviceNameValidator(TaurusDeviceNameValidator): """Validator for Tango device names. Apart from the standard named groups (scheme, authority, path, query and fragment), the following named groups are created: - devname: device name (either alias or slashed name) - [_devalias]: device alias - [_devslashname]: device name in slashed (a/b/c) form - [host] as in :class:`TangoAuthorityNameValidator` - [port] as in :class:`TangoAuthorityNameValidator` Note: brackets on the group name indicate that this group will only contain a string if the URI contains it. """ scheme = "(tango|tango-nodb)" authority = TangoAuthorityNameValidator.authority path = ( r"/?(?P<devname>((?P<_devalias>[^/?#:]+)|" + r"(?P<_devslashname>[^/?#:]+/[^/?#:]+/[^/?#:]+)))" ) query = "(?!)" fragment = "(?P<_fragmentdb>dbase=(yes|no)$)" default_scheme = "tango"
[docs] def getUriGroups(self, name, strict=None): """Reimplementation of getUriGroups to fix the host and authority name using fully qualified domain name for the host. """ ret = TaurusDeviceNameValidator.getUriGroups(self, name, strict) if ret is not None and ret.get("host", None) is not 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
[docs] def getNames(self, fullname, factory=None, queryAuth=True): """reimplemented from :class:`TaurusDeviceNameValidator`. It accepts an extra keyword arg `queryAuth` which, if set to False, will prevent the validator from trying to query a TaurusAuthority to obtain missing info such as the devslashname <--> devalias correspondence. """ groups = self.getUriGroups(fullname) if groups is None: return None default_authority = None if factory is None: from taurus import Factory factory = Factory(scheme="tango") default_authority = factory.get_default_tango_host() if default_authority is None: import PyTango host, port = PyTango.ApiUtil.get_env_var("TANGO_HOST").split(":") # Get the fully qualified domain name host = fqdn_no_alias(host) default_authority = "//{0}:{1}".format(host, port) authority = groups.get("authority") if authority is None: groups["authority"] = authority = default_authority db = None if queryAuth and groups["scheme"] != "tango-nodb": try: db = factory.getAuthority("tango:%s" % authority) except Exception: pass # note, since we validated, we either have alias or slashname (not # both) _devalias = groups.get("_devalias") _devslashname = groups.get("_devslashname") if _devslashname is None and db is not None: # get _devslashname from the alias using the DB _devslashname = db.getElementFullName(_devalias) groups["_devslashname"] = _devslashname if _devslashname is None: # if we still do not have a slashname, we can only give the short return None, None, _devalias # we can now construct everything. First the complete: complete = "{scheme}:{authority}/{_devslashname}".format(**groups) # then the normal if authority.lower() == default_authority.lower(): normal = "{_devslashname}".format(**groups) else: normal = "{authority}/{_devslashname}".format(**groups) # and finally the short if _devalias is not None: short = _devalias else: if db is not None: # get the alias from the DB (if it is defined) short = db.getElementAlias(_devslashname) or _devslashname else: short = _devslashname return complete, normal, short
@property def nonStrictNamePattern(self): """In non-strict mode, allow double-slash even if there is no Authority. (e.g., "tango://a/b/c" passes this non-strict form) """ pattern = ( r"^((?P<scheme>%(scheme)s)://)?" + r"((?P<authority>%(authority)s)(?=/))?" + r"(?P<path>%(path)s)" + r"(\?(?P<query>%(query)s))?" + r"(#%(fragment)s)?$" ) authority = r"(?P<host>([\w\-_]+\.)*[\w\-_]+):(?P<port>\d{1,5})" path = ( "/?(?P<devname>((?P<_devalias>([^/?#:]+))|" + "(?P<_devslashname>[^/?#:]+/[^/?#:]+/[^/?#:]+)))" ) return pattern % dict( scheme=self.scheme, authority=authority, path=path, query="(?!)", fragment="(?!)", )
[docs]class TangoAttributeNameValidator(TaurusAttributeNameValidator): """Validator for Tango attribute names. Apart from the standard named groups (scheme, authority, path, query and fragment), the following named groups are created: - attrname: attribute name including device name - _shortattrname: attribute name excluding device name - devname: as in :class:`TangoDeviceNameValidator` - [_devalias]: as in :class:`TangoDeviceNameValidator` - [_devslashname]: as in :class:`TangoDeviceNameValidator` - [host] as in :class:`TangoAuthorityNameValidator` - [port] as in :class:`TangoAuthorityNameValidator` - [cfgkey] same as fragment (for bck-compat use only) Note: brackets on the group name indicate that this group will only contain a string if the URI contains it. """ scheme = "(tango|tango-nodb)" authority = TangoAuthorityNameValidator.authority path = "(?P<attrname>{0}/(?P<_shortattrname>[^/?:#]+))".format( TangoDeviceNameValidator.path ) query = "(?!)" fragment = "((?P<_fragmentdb>dbase=(yes|no)$)|(?P<cfgkey>[^# ]*))" default_scheme = "tango"
[docs] def getUriGroups(self, name, strict=None): """Reimplementation of getUriGroups to fix the host and authority name using fully qualified domain name for the host. """ ret = TaurusAttributeNameValidator.getUriGroups(self, name, strict) if ret is not None and ret.get("host", None) is not 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
[docs] def getNames(self, fullname, factory=None, queryAuth=True, fragment=False): """Returns the complete and short names""" groups = self.getUriGroups(fullname) if groups is None: return None complete, normal, short = None, None, groups.get("_shortattrname") # reuse the getNames from the Device validator... devname = fullname.rsplit("#", 1)[0].rsplit("/", 1)[0] v = TangoDeviceNameValidator() devcomplete, devnormal, _ = v.getNames( devname, factory=factory, queryAuth=queryAuth ) if devcomplete is not None: complete = "{}/{}".format(devcomplete, short) if devnormal is not None: normal = "{}/{}".format(devnormal, short) # return fragment if requested (ignores the special #dbase=yes|no case) if fragment: key = groups.get("cfgkey", None) return complete, normal, short, key return complete, normal, short
@property def nonStrictNamePattern(self): """In non-strict mode, allow double-slash even if there is no Authority. Also allow old-style "?configuration[=cfgkey]" instead of fragment. If cfgkey is present, it is also stored in the "fragment" named group. For example, "tango://a/b/c/d?configuration=label" passes this non-strict form, and the named group "fragment" will contain "label" """ pattern = ( r"^((?P<scheme>%(scheme)s)://)?" + r"((?P<authority>%(authority)s)(?=/))?" + r"(?P<path>%(path)s)" + r"(\?(?P<query>%(query)s))?" + r"(#%(fragment)s)?$" ) authority = r"(?P<host>([\w\-_]+\.)*[\w\-_]+):(?P<port>\d{1,5})" query = "configuration(=(?P<fragment>(?P<cfgkey>[^# ]+)))?" return pattern % dict( scheme=self.scheme, authority=authority, path=self.path, query=query, fragment="(?!)", )