Source code for robot.running.outputcapture

#  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
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  See the License for the specific language governing permissions and
#  limitations under the License.

import sys
from io import StringIO

from robot.output import LOGGER
from robot.utils import console_decode, console_encode

[docs] class OutputCapturer: def __init__(self, library_import=False): self.library_import = library_import self.stdout = None self.stderr = None
[docs] def start(self): self.stdout = StreamCapturer(stdout=True) self.stderr = StreamCapturer(stdout=False) if self.library_import: LOGGER.enable_library_import_logging()
[docs] def stop(self): self._release_and_log() if self.library_import: LOGGER.disable_library_import_logging()
def __enter__(self): self.start() return self def __exit__(self, exc_type, exc_value, exc_trace): self.stop() return False def _release_and_log(self): stdout, stderr = self._release() if stdout: LOGGER.log_output(stdout) if stderr: LOGGER.log_output(stderr) sys.__stderr__.write(console_encode(stderr, stream=sys.__stderr__)) def _release(self): stdout = self.stdout.release() stderr = self.stderr.release() return stdout, stderr
[docs] class StreamCapturer: def __init__(self, stdout=True): if stdout: self._original = sys.stdout self._set_stream = self._set_stdout else: self._original = sys.stderr self._set_stream = self._set_stderr self._stream = StringIO() self._set_stream(self._stream) def _set_stdout(self, stream): sys.stdout = stream def _set_stderr(self, stream): sys.stderr = stream
[docs] def release(self): # Original stream must be restored before closing the current self._set_stream(self._original) try: return self._get_value(self._stream) finally: self._stream.close() self._avoid_at_exit_errors(self._stream)
def _get_value(self, stream): try: return console_decode(stream.getvalue()) except UnicodeError: # Error occurs if non-ASCII chars logged both as str and unicode. stream.buf = console_decode(stream.buf) stream.buflist = [console_decode(item) for item in stream.buflist] return stream.getvalue() def _avoid_at_exit_errors(self, stream): # Avoid ValueError at program exit when logging module tries to call # methods of streams it has intercepted that are already closed. # Which methods are called, and does logging silence possible errors, # depends on Python version. For related discussion see # stream.write = lambda s: None stream.flush = lambda: None