Remote.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.
 
from __future__ import absolute_import
 
try:
    import httplib
    import xmlrpclib
except ImportError:  # Py3
    import http.client as httplib
    import xmlrpc.client as xmlrpclib
import re
import socket
import sys
import time
 
try:
    from xml.parsers.expat import ExpatError
except ImportError:   # No expat in IronPython 2.7
    class ExpatError(Exception):
        pass
 
from robotide.lib.robot.errors import RemoteError
from robotide.lib.robot.utils import (is_bytes, is_dict_like, is_list_like, is_number,
                         is_string, timestr_to_secs, unic, DotDict,
                         IRONPYTHON, JYTHON, PY2)
 
 
class Remote():
    ROBOT_LIBRARY_SCOPE = 'TEST SUITE'
 
    
    def __init__(self, uri='http://127.0.0.1:8270', timeout=None):
        if '://' not in uri:
            uri = 'http://' + uri
        if timeout:
            timeout = timestr_to_secs(timeout)
        self._uri_uri = uri
        self._client_client = XmlRpcRemoteClient(uri, timeout)
 
    def get_keyword_names(self, attempts=2):
        for i in range(attempts):
            time.sleep(i)
            try:
                return self._client_client.get_keyword_names()
            except TypeError as err:
                error = err
        raise RuntimeError('Connecting remote server at %s failed: %s'
                           % (self._uri_uri, error))
 
    def get_keyword_arguments(self, name):
        try:
            return self._client_client.get_keyword_arguments(name)
        except TypeError:
            return ['*args']
 
    def get_keyword_types(self, name):
        try:
            return self._client_client.get_keyword_types(name)
        except TypeError:
            return None
 
    def get_keyword_tags(self, name):
        try:
            return self._client_client.get_keyword_tags(name)
        except TypeError:
            return None
 
    def get_keyword_documentation(self, name):
        try:
            return self._client_client.get_keyword_documentation(name)
        except TypeError:
            return None
 
    def run_keyword(self, name, args, kwargs):
        coercer = ArgumentCoercer()
        args = coercer.coerce(args)
        kwargs = coercer.coerce(kwargs)
        result = RemoteResult(self._client_client.run_keyword(name, args, kwargs))
        sys.stdout.write(result.output)
        if result.status != 'PASS':
            raise RemoteError(result.error, result.traceback, result.fatal,
                              result.continuable)
        return result.return_
 
 
class ArgumentCoercer():
    binary = re.compile('[\x00-\x08\x0B\x0C\x0E-\x1F]')
    non_ascii = re.compile('[\x80-\xff]')
 
    def coerce(self, argument):
        for handles, handler in [(is_string, self._handle_string_handle_string),
                                 (is_bytes, self._handle_bytes_handle_bytes),
                                 (is_number, self._pass_through_pass_through),
                                 (is_dict_like, self._coerce_dict_coerce_dict),
                                 (is_list_like, self._coerce_list_coerce_list),
                                 (lambda arg: True, self._to_string_to_string)]:
            if handles(argument):
                return handler(argument)
 
    def _handle_string(self, arg):
        if self._string_contains_binary_string_contains_binary(arg):
            return self._handle_binary_in_string_handle_binary_in_string(arg)
        return arg
 
    def _string_contains_binary(self, arg):
        return (self.binarybinary.search(arg) or
                is_bytes(arg) and self.non_asciinon_ascii.search(arg))
 
    def _handle_binary_in_string(self, arg):
        try:
            if not is_bytes(arg):
                arg = arg.encode('ASCII')
        except UnicodeError:
            raise ValueError('Cannot represent %r as binary.' % arg)
        return xmlrpclib.Binary(arg)
 
    def _handle_bytes(self, arg):
        # http://bugs.jython.org/issue2429
        if IRONPYTHON or JYTHON:
            arg = str(arg)
        return xmlrpclib.Binary(arg)
 
    def _pass_through(self, arg):
        return arg
 
    def _coerce_list(self, arg):
        return [self.coercecoerce(item) for item in arg]
 
    def _coerce_dict(self, arg):
        return dict((self._to_key_to_key(key), self.coercecoerce(arg[key])) for key in arg)
 
    def _to_key(self, item):
        item = self._to_string_to_string(item)
        self._validate_key_validate_key(item)
        return item
 
    def _to_string(self, item):
        item = unic(item) if item is not None else ''
        return self._handle_string_handle_string(item)
 
    def _validate_key(self, key):
        if isinstance(key, xmlrpclib.Binary):
            raise ValueError('Dictionary keys cannot be binary. Got %s%r.'
                             % ('b' if PY2 else '', key.data))
        if IRONPYTHON:
            try:
                key.encode('ASCII')
            except UnicodeError:
                raise ValueError('Dictionary keys cannot contain non-ASCII '
                                 'characters on IronPython. Got %r.' % key)
 
 
class RemoteResult():
 
    def __init__(self, result):
        if not (is_dict_like(result) and 'status' in result):
            raise RuntimeError('Invalid remote result dictionary: %s' % result)
        self.statusstatus = result['status']
        self.outputoutput = unic(self._get_get(result, 'output'))
        self.return_return_ = self._get_get(result, 'return')
        self.errorerror = unic(self._get_get(result, 'error'))
        self.tracebacktraceback = unic(self._get_get(result, 'traceback'))
        self.fatalfatal = bool(self._get_get(result, 'fatal', False))
        self.continuablecontinuable = bool(self._get_get(result, 'continuable', False))
 
    def _get(self, result, key, default=''):
        value = result.get(key, default)
        return self._convert_convert(value)
 
    def _convert(self, value):
        if isinstance(value, xmlrpclib.Binary):
            return bytes(value.data)
        if is_dict_like(value):
            return DotDict((k, self._convert_convert(v)) for k, v in value.items())
        if is_list_like(value):
            return [self._convert_convert(v) for v in value]
        return value
 
 
class XmlRpcRemoteClient():
 
    def __init__(self, uri, timeout=None):
        if uri.startswith('https://'):
            transport = TimeoutHTTPSTransport(timeout=timeout)
        else:
            transport = TimeoutHTTPTransport(timeout=timeout)
        self._server_server = xmlrpclib.ServerProxy(uri, encoding='UTF-8',
                                             transport=transport)
 
    def get_keyword_names(self):
        try:
            return self._server_server.get_keyword_names()
        except (socket.error, xmlrpclib.Error) as err:
            raise TypeError(err)
 
    def get_keyword_arguments(self, name):
        try:
            return self._server_server.get_keyword_arguments(name)
        except xmlrpclib.Error:
            raise TypeError
 
    def get_keyword_types(self, name):
        try:
            return self._server_server.get_keyword_types(name)
        except xmlrpclib.Error:
            raise TypeError
 
    def get_keyword_tags(self, name):
        try:
            return self._server_server.get_keyword_tags(name)
        except xmlrpclib.Error:
            raise TypeError
 
    def get_keyword_documentation(self, name):
        try:
            return self._server_server.get_keyword_documentation(name)
        except xmlrpclib.Error:
            raise TypeError
 
    def run_keyword(self, name, args, kwargs):
        run_keyword_args = [name, args, kwargs] if kwargs else [name, args]
        try:
            return self._server_server.run_keyword(*run_keyword_args)
        except xmlrpclib.Fault as err:
            message = err.faultString
        except socket.error as err:
            message = 'Connection to remote server broken: %s' % err
        except ExpatError as err:
            message = ('Processing XML-RPC return value failed. '
                       'Most often this happens when the return value '
                       'contains characters that are not valid in XML. '
                       'Original error was: ExpatError: %s' % err)
        raise RuntimeError(message)
 
 
# Custom XML-RPC timeouts based on
# http://stackoverflow.com/questions/2425799/timeout-for-xmlrpclib-client-requests
 
class TimeoutHTTPTransport(xmlrpclib.Transport):
    
    _connection_class = httplib.HTTPConnection
 
    def __init__(self, use_datetime=0, timeout=None):
        xmlrpclib.Transport.__init__(self, use_datetime)
        if not timeout:
            timeout = socket._GLOBAL_DEFAULT_TIMEOUT
        self.timeouttimeout = timeout
 
    def make_connection(self, host):
        if self._connection_connection and host == self._connection_connection[0]:
            return self._connection_connection[1]
        chost, self._extra_headers, x509 = self.get_host_info(host)
        self._connection_connection = host, self._connection_class_connection_class(chost, timeout=self.timeouttimeout)
        return self._connection_connection[1]
 
 
if IRONPYTHON:
 
    class TimeoutHTTPTransport(xmlrpclib.Transport):
 
        def __init__(self, use_datetime=0, timeout=None):
            xmlrpclib.Transport.__init__(self, use_datetime)
            if timeout:
                raise RuntimeError('Timeouts are not supported on IronPython.')
 
 
class TimeoutHTTPSTransport(TimeoutHTTPTransport):
    
    _connection_class = httplib.HTTPSConnection