testlibraries.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
 
import inspect
import os
 
from robotide.lib.robot.errors import DataError
from robotide.lib.robot.libraries import STDLIBS
from robotide.lib.robot.output import LOGGER
from robotide.lib.robot.utils import (getdoc, get_error_details, Importer, is_java_init,
                         is_java_method, JYTHON, normalize, seq2str2, unic,
                         is_list_like, PY2, PYPY, type_name)
 
from .arguments import EmbeddedArguments
from .context import EXECUTION_CONTEXTS
from .dynamicmethods import (GetKeywordArguments, GetKeywordDocumentation,
                             GetKeywordNames, GetKeywordTags, RunKeyword)
from .handlers import Handler, InitHandler, DynamicHandler, EmbeddedArgumentsHandler
from .handlerstore import HandlerStore
from .libraryscopes import LibraryScope
from .outputcapture import OutputCapturer
 
 
if JYTHON:
    from java.lang import Object
else:
    Object = None
 
 
def TestLibrary(name, args=None, variables=None, create_handlers=True):
    if name in STDLIBS:
        import_name = 'robot.libraries.' + name
    else:
        import_name = name
    with OutputCapturer(library_import=True):
        importer = Importer('test library')
        libcode, source = importer.import_class_or_module(import_name,
                                                          return_source=True)
    libclass = _get_lib_class(libcode)
    lib = libclass(libcode, name, args or [], source, variables)
    if create_handlers:
        lib.create_handlers()
    return lib
 
 
def _get_lib_class(libcode):
    if inspect.ismodule(libcode):
        return _ModuleLibrary
    if GetKeywordNames(libcode):
        if RunKeyword(libcode):
            return _DynamicLibrary
        else:
            return _HybridLibrary
    return _ClassLibrary
 
 
class _BaseTestLibrary():
    
    _failure_level = 'INFO'
 
    def __init__(self, libcode, name, args, source, variables):
        if os.path.exists(name):
            name = os.path.splitext(os.path.basename(os.path.abspath(name)))[0]
        self.versionversion = self._get_version_get_version(libcode)
        self.namename = name
        self.orig_nameorig_name = name  # Stores original name when importing WITH NAME
        self.sourcesource = source
        self.handlershandlers = HandlerStore(self.namename, HandlerStore.TEST_LIBRARY_TYPE)
        self.has_listenerhas_listener = None  # Set when first instance is created
        self._doc_doc = None
        self.doc_formatdoc_format = self._get_doc_format_get_doc_format(libcode)
        self.scopescope = LibraryScope(libcode, self)
        self.initinit = self._create_init_handler_create_init_handler(libcode)
        self.positional_args, self.named_args \
            = self.initinit.resolve_arguments(args, variables)
        self._libcode_libcode = libcode
        self._libinst_libinst = None
 
    def __len__(self):
        return len(self.handlershandlers)
 
    @property
    doc = property
    
    def doc(self):
        if self._doc_doc is None:
            self._doc_doc = getdoc(self.get_instanceget_instance())
        return self._doc_doc
 
    def create_handlers(self):
        self._create_handlers_create_handlers(self.get_instanceget_instance())
        self.reset_instancereset_instance()
 
    def reload(self):
        self.handlershandlers = HandlerStore(self.namename, HandlerStore.TEST_LIBRARY_TYPE)
        self._create_handlers_create_handlers(self.get_instanceget_instance())
 
    def start_suite(self):
        self.scopescope.start_suite()
 
    def end_suite(self):
        self.scopescope.end_suite()
 
    def start_test(self):
        self.scopescope.start_test()
 
    def end_test(self):
        self.scopescope.end_test()
 
    def _get_version(self, libcode):
        return self._get_attr_get_attr(libcode, 'ROBOT_LIBRARY_VERSION') \
            or self._get_attr_get_attr(libcode, '__version__')
 
    def _get_attr(self, object, attr, default='', upper=False):
        value = unic(getattr(object, attr, default))
        if upper:
            value = normalize(value, ignore='_').upper()
        return value
 
    def _get_doc_format(self, libcode):
        return self._get_attr_get_attr(libcode, 'ROBOT_LIBRARY_DOC_FORMAT', upper=True)
 
    def _create_init_handler(self, libcode):
        return InitHandler(self, self._resolve_init_method_resolve_init_method(libcode))
 
    def _resolve_init_method(self, libcode):
        init = getattr(libcode, '__init__', None)
        return init if init and self._valid_init_valid_init(init) else lambda: None
 
    def _valid_init(self, method):
        # https://bitbucket.org/pypy/pypy/issues/2462/
        if PYPY:
            if PY2:
                return method.__func__ is not object.__init__.__func__
            return method is not object.__init__
        return (inspect.ismethod(method) or     # PY2
                inspect.isfunction(method) or   # PY3
                is_java_init(method))
 
    def reset_instance(self, instance=None):
        prev = self._libinst_libinst
        if not self.scopescope.is_global:
            self._libinst_libinst = instance
        return prev
 
    def get_instance(self, create=True):
        if not create:
            return self._libinst_libinst
        if self._libinst_libinst is None:
            self._libinst_libinst = self._get_instance_get_instance(self._libcode_libcode)
        if self.has_listenerhas_listener is None:
            self.has_listenerhas_listener = bool(self.get_listenersget_listeners(self._libinst_libinst))
        return self._libinst_libinst
 
    def _get_instance(self, libcode):
        with OutputCapturer(library_import=True):
            try:
                return libcode(*self.positional_args, **dict(self.named_args))
            except:
                self._raise_creating_instance_failed_raise_creating_instance_failed()
 
    def get_listeners(self, libinst=None):
        if libinst is None:
            libinst = self.get_instanceget_instance()
        listeners = getattr(libinst, 'ROBOT_LIBRARY_LISTENER', None)
        if listeners is None:
            return []
        if is_list_like(listeners):
            return listeners
        return [listeners]
 
    def register_listeners(self):
        if self.has_listenerhas_listener:
            try:
                listeners = EXECUTION_CONTEXTS.current.output.library_listeners
                listeners.register(self.get_listenersget_listeners(), self)
            except DataError as err:
                self.has_listenerhas_listener = False
                # Error should have information about suite where the
                # problem occurred but we don't have such info here.
                LOGGER.error("Registering listeners for library '%s' failed: %s"
                             % (self.namename, err))
 
    def unregister_listeners(self, close=False):
        if self.has_listenerhas_listener:
            listeners = EXECUTION_CONTEXTS.current.output.library_listeners
            listeners.unregister(self, close)
 
    def close_global_listeners(self):
        if self.scopescope.is_global:
            for listener in self.get_listenersget_listeners():
                self._close_listener_close_listener(listener)
 
    def _close_listener(self, listener):
        method = (getattr(listener, 'close', None) or
                  getattr(listener, '_close', None))
        try:
            if method:
                method()
        except:
            message, details = get_error_details()
            name = getattr(listener, '__name__', None) or type_name(listener)
            LOGGER.error("Calling method '%s' of listener '%s' failed: %s"
                         % (method.__name__, name, message))
            LOGGER.info("Details:\n%s" % details)
 
    def _create_handlers(self, libcode):
        try:
            names = self._get_handler_names_get_handler_names(libcode)
        except:
            message, details = get_error_details()
            raise DataError("Getting keyword names from library '%s' failed: %s"
                            % (self.namename, message), details)
        for name in names:
            method = self._try_to_get_handler_method_try_to_get_handler_method(libcode, name)
            if method:
                handler, embedded = self._try_to_create_handler_try_to_create_handler(name, method)
                if handler:
                    try:
                        self.handlershandlers.add(handler, embedded)
                    except DataError as err:
                        LOGGER.error("Error in test library '%s': "
                                     "Creating keyword '%s' failed: %s"
                                     % (self.namename, handler.name, err.message))
                    else:
                        LOGGER.debug("Created keyword '%s'" % handler.name)
 
    def _get_handler_names(self, libcode):
        return [name for name in dir(libcode)
                if name[:1] != '_' or self._has_robot_name_has_robot_name(name, libcode)]
 
    def _has_robot_name(self, name, libcode):
        try:
            handler = self._get_handler_method_get_handler_method(libcode, name)
        except DataError:
            return False
        return hasattr(handler, 'robot_name')
 
    def _try_to_get_handler_method(self, libcode, name):
        try:
            return self._get_handler_method_get_handler_method(libcode, name)
        # TODO: RF 3.1: Catch only DataError or at least consider others errors.
        # Don't want to do that in a minor release.
        except:
            self._report_adding_keyword_failed_report_adding_keyword_failed(name)
            return None
 
    def _report_adding_keyword_failed(self, name, message=None, details=None,
                                      level=None):
        if not message:
            message, details = get_error_details()
        LOGGER.write("Adding keyword '%s' to library '%s' failed: %s"
                     % (name, self.namename, message), level or self._failure_level_failure_level)
        if details:
            LOGGER.debug('Details:\n%s' % details)
 
    def _get_handler_method(self, libcode, name):
        method = getattr(libcode, name)
        if not inspect.isroutine(method):
            raise DataError('Not a method or function')
        return method
 
    def _try_to_create_handler(self, name, method):
        try:
            handler = self._create_handler_create_handler(name, method)
        except DataError as err:
            self._report_adding_keyword_failed_report_adding_keyword_failed(name, err.message, level='ERROR')
            return None, False
        # TODO: RF 3.1: Catch only DataError or at least consider others errors.
        # Don't want to do that in a minor release.
        except:
            self._report_adding_keyword_failed_report_adding_keyword_failed(name)
            return None, False
        try:
            return self._get_possible_embedded_args_handler_get_possible_embedded_args_handler(handler)
        except DataError as err:
            self._report_adding_keyword_failed_report_adding_keyword_failed(handler.name, err.message,
                                               level='ERROR')
            return None, False
 
    def _create_handler(self, handler_name, handler_method):
        return Handler(self, handler_name, handler_method)
 
    def _get_possible_embedded_args_handler(self, handler):
        embedded = EmbeddedArguments(handler.name)
        if embedded:
            self._validate_embedded_count_validate_embedded_count(embedded, handler.arguments)
            return EmbeddedArgumentsHandler(embedded.name, handler), True
        return handler, False
 
    def _validate_embedded_count(self, embedded, arguments):
        if not (arguments.minargs <= len(embedded.args) <= arguments.maxargs):
            raise DataError('Embedded argument count does not match number of '
                            'accepted arguments.')
 
    def _raise_creating_instance_failed(self):
        msg, details = get_error_details()
        if self.positional_args or self.named_args:
            args = self.positional_args \
                + ['%s=%s' % item for item in self.named_args]
            args_text = 'arguments %s' % seq2str2(args)
        else:
            args_text = 'no arguments'
        raise DataError("Initializing test library '%s' with %s failed: %s\n%s"
                        % (self.namename, args_text, msg, details))
 
 
class _ClassLibrary(_BaseTestLibrary):
 
    def _get_handler_method(self, libinst, name):
        # Type is checked before using getattr to avoid calling properties,
        # most importantly bean properties generated by Jython (issue 188).
        for item in (libinst,) + inspect.getmro(libinst.__class__):
            if item in (object, Object):
                continue
            if hasattr(item, '__dict__') and name in item.__dict__:
                self._validate_handler_validate_handler(item.__dict__[name])
                return getattr(libinst, name)
        raise DataError('No non-implicit implementation found')
 
    def _validate_handler(self, handler):
        if not inspect.isroutine(handler):
            raise DataError('Not a method or function')
        if self._is_implicit_java_or_jython_method_is_implicit_java_or_jython_method(handler):
            raise DataError('Implicit methods are ignored')
 
    def _is_implicit_java_or_jython_method(self, handler):
        if not is_java_method(handler):
            return False
        for signature in handler.argslist[:handler.nargs]:
            cls = signature.declaringClass
            if not (cls is Object or cls.__module____module__ == 'org.python.proxies'):
                return False
        return True
 
 
class _ModuleLibrary(_BaseTestLibrary):
 
    def _get_handler_method(self, libcode, name):
        method = _BaseTestLibrary._get_handler_method(self, libcode, name)
        if hasattr(libcode, '__all__') and name not in libcode.__all__:
            raise DataError('Not exposed as a keyword')
        return method
 
    def get_instance(self, create=True):
        if not create:
            return self._libcode_libcode
        if self.has_listenerhas_listenerhas_listener is None:
            self.has_listenerhas_listenerhas_listener = bool(self.get_listenersget_listeners(self._libcode_libcode))
        return self._libcode_libcode
 
    def _create_init_handler(self, libcode):
        return InitHandler(self, lambda: None)
 
 
class _HybridLibrary(_BaseTestLibrary):
    
    _failure_level = 'ERROR'
 
    def _get_handler_names(self, instance):
        return GetKeywordNames(instance)()
 
 
class _DynamicLibrary(_BaseTestLibrary):
    
    _failure_level = 'ERROR'
 
    def __init__(self, libcode, name, args, source, variables=None):
        _BaseTestLibrary.__init__(self, libcode, name, args, source, variables)
 
    @property
    doc = property
    
    def doc(self):
        if self._doc_doc_doc is None:
            self._doc_doc_doc = (self._get_kw_doc_get_kw_doc('__intro__') or
                         _BaseTestLibrary.doc.fget(self))
        return self._doc_doc_doc
 
    def _get_kw_doc(self, name):
        getter = GetKeywordDocumentation(self.get_instanceget_instance())
        return getter(name)
 
    def _get_kw_args(self, name):
        getter = GetKeywordArguments(self.get_instanceget_instance())
        return getter(name)
 
    def _get_kw_tags(self, name):
        getter = GetKeywordTags(self.get_instanceget_instance())
        return getter(name)
 
    def _get_handler_names(self, instance):
        return GetKeywordNames(instance)()
 
    def _get_handler_method(self, instance, name):
        return RunKeyword(instance)
 
    def _create_handler(self, name, method):
        argspec = self._get_kw_args_get_kw_args(name)
        tags = self._get_kw_tags_get_kw_tags(name)
        doc = self._get_kw_doc_get_kw_doc(name)
        return DynamicHandler(self, name, method, doc, argspec, tags)
 
    def _create_init_handler(self, libcode):
        docgetter = lambda: self._get_kw_doc_get_kw_doc('__init__')
        return InitHandler(self, self._resolve_init_method_resolve_init_method(libcode), docgetter)