TestRunnerAgent.py

Go to the documentation of this file.

#!/usr/bin/env python
# ----------------------------------------------------------------------------
# Copyright 2010 Orbitz WorldWide
#
# 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.
 
# Modified by Mikko Korpela under NSN copyrights
#  Copyright 2008-2015 Nokia Solutions and Networks
#
#  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.
 
# Ammended by Timothy Alexander <dragonfyre13@gmail.com>
# (StreamHandler class added)
#   Copyright 2013 Timothy Alexander
#   Licensed under the Apache License, Version 2.0
#      http://www.apache.org/licenses/LICENSE-2.0
 
#
# Modified by Mateusz Marzec under NSN copyrights
# Copyright 2015 Nokia Solutions and Networks
# * Licensed under the Apache License, Version 2.0,
# * see license.txt file for details.
#
 
# Ammended by Helio Guilherme <helioxentric@gmail.com>
#  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 copy
import os
import pickle
import platform
import sys
import socket
import threading
 
PLATFORM = platform.python_implementation()
 
try:
    import socketserver as SocketServer
except ImportError as e:
    raise e
 
try:
    # to find robot (we use provided lib)
    sys.path.append(os.path.join(os.path.dirname(__file__), '../../lib'))
    from robot.errors import ExecutionFailed
    from robot.running import EXECUTION_CONTEXTS
    from robot.running.signalhandler import STOP_SIGNAL_MONITOR
    from robot.utils import encoding
    from robot.utils.encoding import SYSTEM_ENCODING
except ImportError:
    encoding = None
    raise
 
try:
    import json
    
    _JSONAVAIL = True
except ImportError:
    
    _JSONAVAIL = False
 
try:
    from StringIO import StringIO
except ImportError:  # py3 <=3.6
    from io import StringIO
 
HOST = "localhost"
 
# Setting Output encoding to UTF-8 and ignoring the platform specs
# RIDE will expect UTF-8
# Set output encoding to UTF-8 for piped output streams
# DEBUG This was working in Linux always!
#if encoding:
#    encoding.OUTPUT_ENCODING = 'UTF-8'
# print("DEBUG: TestRunnerAgent encoding %s\n" % SYSTEM_ENCODING )
 
 
def _is_logged(level):
    current = EXECUTION_CONTEXTS.current
    if current is None:
        return True
    out = current.output
    if out is None:
        return True
    return out._xmllogger._log_message_is_logged(level)
 
 
 
class TestRunnerAgent:
    ROBOT_LISTENER_API_VERSION = 2
 
    def __init__(self, *args):
        self.portport = int(args[0])
        self.hosthost = HOST
        self.socksock = None
        self.filehandlerfilehandler = None
        self.streamhandlerstreamhandler = None
        self._connect_connect()
        self._send_pid_send_pid()
        self._create_debugger_create_debugger((len(args) >= 2) and (args[1] == 'True'))
        self._create_kill_server_create_kill_server()
        print("TestRunnerAgent: Running under %s %s\n" %
              (PLATFORM, sys.version.split()[0]))
 
    def _create_debugger(self, pause_on_failure):
        self._debugger_debugger = RobotDebugger(pause_on_failure)
 
    def _create_kill_server(self):
        self._killer_killer = RobotKillerServer(self._debugger_debugger)
        self._server_thread_server_thread = threading.Thread(
            target=self._killer_killer.serve_forever)
        # DEPRECATED: self._server_thread.setDaemon(True)
        self._server_thread_server_thread.daemon = True
        self._server_thread_server_thread.start()
        self._send_server_port_send_server_port(self._killer_killer.server_address[1])
 
    def _send_pid(self):
        self._send_socket_send_socket("pid", os.getpid())
 
    def _send_server_port(self, port):
        self._send_socket_send_socket("port", port)
 
    def start_test(self, name, attrs):
        self._send_socket_send_socket("start_test", name, attrs)
 
    def end_test(self, name, attrs):
        self._send_socket_send_socket("end_test", name, attrs)
 
    def start_suite(self, name, attrs):
        attrs_copy = copy.copy(attrs)
        del attrs_copy['doc']
        attrs_copy['is_dir'] = os.path.isdir(attrs['source'])
        self._send_socket_send_socket("start_suite", name, attrs_copy)
 
    def end_suite(self, name, attrs):
        attrs_copy = copy.copy(attrs)
        del attrs_copy['doc']
        attrs_copy['is_dir'] = os.path.isdir(attrs['source'])
        self._send_socket_send_socket("end_suite", name, attrs_copy)
 
    def start_keyword(self, name, attrs):
        # pass empty args, see https://github.com/nokia/RED/issues/32
 
        # we're cutting args from original attrs dict, because it may contain
        # objects which are not json-serializable and we don't need them anyway
        attrs_copy = copy.copy(attrs)
        del attrs_copy['args']
        del attrs_copy['doc']
        del attrs_copy['assign']
 
        self._send_socket_send_socket("start_keyword", name, attrs_copy)
        if self._debugger_debugger.is_breakpoint(name, attrs):  # must check original
            self._debugger_debugger.pause()
        paused = self._debugger_debugger.is_paused()
        if paused:
            self._send_socket_send_socket('paused')
        self._debugger_debugger.start_keyword()
        if paused:
            self._send_socket_send_socket('continue')
 
    def end_keyword(self, name, attrs):
        # pass empty args, see https://github.com/nokia/RED/issues/32
        attrs_copy = copy.copy(attrs)
        del attrs_copy['args']
        del attrs_copy['doc']
        del attrs_copy['assign']
 
        self._send_socket_send_socket("end_keyword", name, attrs_copy)
        self._debugger_debugger.end_keyword(attrs['status'] == 'PASS')
 
    def message(self, message):
        pass
 
    def log_message(self, message):
        if _is_logged(message['level']):
            self._send_socket("log_message", message)
 
    def log_file(self, path):
        self._send_socket_send_socket("log_file", path)
 
    def output_file(self, path):
        pass
 
    def report_file(self, path):
        self._send_socket("report_file", path)
 
    def summary_file(self, path):
        pass
 
    def debug_file(self, path):
        pass
 
    def close(self):
        self._send_socket("close")
        if self.sock:
            self.filehandler.close()
            self.sock.close()
 
    
    def _connect(self):
        try:
            self.socksock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socksock.connect((self.hosthost, self.portport))
            # Iron python does not return right object type if not binary mode
            self.filehandlerfilehandler = self.socksock.makefile('wb')
            self.streamhandlerstreamhandler = StreamHandler(self.filehandlerfilehandler)
        except socket.error as e:
            print('unable to open socket to "%s:%s" error: %s'
                  % (self.hosthost, self.portport, str(e)))
            self.socksock = None
            self.filehandlerfilehandler = None
 
    def _send_socket(self, name, *args):
        try:
            if self.filehandlerfilehandler:
                packet = (name, args)
                self.streamhandlerstreamhandler.dump(packet)
                self.filehandlerfilehandler.flush()
        except Exception:
            import traceback
            traceback.print_exc(file=sys.stdout)
            sys.stdout.flush()
            raise
 
 
class RobotDebugger():
 
    def __init__(self, pause_on_failure=False):
        self._state_state = 'running'
        self._keyword_level_keyword_level = 0
        self._pause_when_on_level_pause_when_on_level = -1
        self._pause_on_failure_pause_on_failure = pause_on_failure
        self._resume_resume = threading.Event()
 
    @staticmethod
    def is_breakpoint(name, attrs):
        if len(attrs['args']) > 0:
            return name == 'BuiltIn.Comment' and \
                   str(attrs['args'][0]).upper().startswith(u"PAUSE")
 
    def pause(self):
        self._resume_resume.clear()
        self._state_state = 'pause'
 
    def pause_on_failure(self, pause):
        self._pause_on_failure_pause_on_failure = pause
 
    def resume(self):
        self._state_state = 'running'
        self._pause_when_on_level_pause_when_on_level = -1
        self._resume_resume.set()
 
    def step_next(self):
        self._state_state = 'step_next'
        self._resume_resume.set()
 
    def step_over(self):
        self._state_state = 'step_over'
        self._resume_resume.set()
 
    def start_keyword(self):
        while self._state_state == 'pause':
            self._resume_resume.wait()
            self._resume_resume.clear()
        if self._state_state == 'step_next':
            self._state_state = 'pause'
        elif self._state_state == 'step_over':
            self._pause_when_on_level_pause_when_on_level = self._keyword_level_keyword_level
            self._state_state = 'resume'
        self._keyword_level_keyword_level += 1
 
    def end_keyword(self, passed=True):
        self._keyword_level_keyword_level -= 1
        if self._keyword_level_keyword_level == self._pause_when_on_level_pause_when_on_level or\
                (self._pause_on_failure_pause_on_failure and not passed):
            self._state_state = 'pause'
 
    def is_paused(self):
        return self._state_state == 'pause'
 
 
class RobotKillerServer(SocketServer.TCPServer):
    allow_reuse_address = True
 
    def __init__(self, debugger):
        SocketServer.TCPServer.__init__(self, ("", 0), RobotKillerHandler)
        self.debuggerdebugger = debugger
 
 
class RobotKillerHandler(SocketServer.StreamRequestHandler):
    def handle(self):
        data = self.request.makefile('r').read().strip()
        if data == 'kill':
            self._signal_kill_signal_kill()
        elif data == 'pause':
            self.server.debugger.pause()
        elif data == 'resume':
            self.server.debugger.resume()
        elif data == 'step_next':
            self.server.debugger.step_next()
        elif data == 'step_over':
            self.server.debugger.step_over()
        elif data == 'pause_on_failure':
            self.server.debugger.pause_on_failure(True)
        elif data == 'do_not_pause_on_failure':
            self.server.debugger.pause_on_failure(False)
 
    @staticmethod
    def _signal_kill():
        try:
            STOP_SIGNAL_MONITOR(1, '')
        except ExecutionFailed:
            pass
 
 
# NOTE: Moved to bottom of TestRunnerAgent per feedback in pull request,
#       so jybot doesn't encounter issues. Special imports at top of file.
 
class StreamError(Exception):
    pass
 
 
 
class EncodeError(StreamError):
    wrapped_exceptions = (pickle.PicklingError, )
 
 
 
class DecodeError(StreamError):
    # NOTE: No JSONDecodeError in json in stdlib for python >= 2.6
    wrapped_exceptions = (pickle.UnpicklingError,)
    if _JSONAVAIL:
        if hasattr(json, 'JSONDecodeError'):
            wrapped_exceptions = (pickle.UnpicklingError, json.JSONDecodeError)
 
 
def dump(obj, fp):
    StreamHandler(fp).dump(obj)
 
 
def load(fp):
    return StreamHandler(fp).load()
 
 
 
def dumps(obj):
    fp = StringIO()
    StreamHandler(fp).dump(obj)
    return fp.getvalue()
 
 
 
def loads(s):
    fp = StringIO(s)
    return StreamHandler(fp).load()
 
 
 
class StreamHandler():
    loads = staticmethod(loads)
    dumps = staticmethod(dumps)
 
    
    def __init__(self, fp):
        if _JSONAVAIL:
            self._json_encoder_json_encoder = json.JSONEncoder(separators=(',', ':'),
                                                  sort_keys=True).encode
            self._json_decoder_json_decoder = json.JSONDecoder(strict=False).decode
        else:
            def json_not_impl(dummy):
                raise NotImplementedError(
                    'Python should include json. Please check your Python installation.')
            self._json_decoder_json_decoder = staticmethod(json_not_impl)
            self._json_encoder_json_encoder = staticmethod(json_not_impl)
        self.fpfp = fp
 
    
    def dump(self, obj):
        # NOTE: Slightly less efficient than doing iterencode directly into the
        #       fp, however difference is negligable and reduces complexity of
        #       of the StreamHandler class (treating pickle and json the same)
        write_list = []
        if _JSONAVAIL:
            try:
                s = self._json_encoder_json_encoder(obj)
                write_list.append('J')
                write_list.extend([str(len(s)), '|', s])
            except:
                # Probably just failed to JSON-encode an object; try pickle.
                pass
        if not write_list:
            s = pickle.dumps(obj, pickle.HIGHEST_PROTOCOL)
            write_list.append('P')
            write_list.extend([str(len(s)), '|', s])
        self.fpfp.write(bytes(''.join(write_list), "UTF-8"))
 
    
    def load(self):
        header = self._load_header_load_header()
        msgtype = header[0]
        msglen = header[1:]
        if not msglen.isdigit():
            raise DecodeError('Message header not valid: %r' % header)
        msglen = int(msglen)
        buff = StringIO()
        # Don't use StringIO.len for sizing, reports string len not bytes
        buff.write(self.fpfp.read(msglen))
        try:
            if msgtype == 'J':
                return self._json_decoder_json_decoder(buff.getvalue())
            elif msgtype == 'P':
                return pickle.loads(buff.getvalue())
            else:
                raise DecodeError("Message type %r not supported" % msgtype)
        except DecodeError.wrapped_exceptions as e:
            raise DecodeError(str(e))
 
    
    def _load_header(self):
        buff = StringIO()
        while len(buff.getvalue()) == 0 or buff.getvalue()[-1] != '|':
            recv_char = self.fpfp.read(1)
            if not recv_char:
                raise EOFError('File/Socket closed while reading load header')
            buff.write(recv_char)
        return buff.getvalue()[:-1]