# 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 sys
from robot.errors import DataError
from robot.utils import get_console_length, getshortdoc, isatty, pad_console_length
from .highlighting import HighlightingStream
from ..loggerapi import LoggerApi
[docs]
class VerboseOutput(LoggerApi):
def __init__(self, width=78, colors='AUTO', markers='AUTO', stdout=None,
stderr=None):
self.writer = VerboseWriter(width, colors, markers, stdout, stderr)
self.started = False
self.started_keywords = 0
self.running_test = False
[docs]
def start_suite(self, data, result):
if not self.started:
self.writer.suite_separator()
self.started = True
self.writer.info(data.full_name, result.doc, start_suite=True)
self.writer.suite_separator()
[docs]
def end_suite(self, data, result):
self.writer.info(data.full_name, result.doc)
self.writer.status(result.status)
self.writer.message(result.full_message)
self.writer.suite_separator()
[docs]
def start_test(self, data, result):
self.writer.info(result.name, result.doc)
self.running_test = True
[docs]
def end_test(self, data, result):
self.writer.status(result.status, clear=True)
self.writer.message(result.message)
self.writer.test_separator()
self.running_test = False
[docs]
def start_body_item(self, data, result):
self.started_keywords += 1
[docs]
def end_body_item(self, data, result):
self.started_keywords -= 1
if self.running_test and not self.started_keywords:
self.writer.keyword_marker(result.status)
[docs]
def message(self, msg):
if msg.level in ('WARN', 'ERROR'):
self.writer.error(msg.message, msg.level, clear=self.running_test)
[docs]
def result_file(self, kind, path):
self.writer.output(kind, path)
[docs]
class VerboseWriter:
_status_length = len('| PASS |')
def __init__(self, width=78, colors='AUTO', markers='AUTO', stdout=None,
stderr=None):
self.width = width
self.stdout = HighlightingStream(stdout or sys.__stdout__, colors)
self.stderr = HighlightingStream(stderr or sys.__stderr__, colors)
self._keyword_marker = KeywordMarker(self.stdout, markers)
self._last_info = None
[docs]
def info(self, name, doc, start_suite=False):
width, separator = self._get_info_width_and_separator(start_suite)
self._last_info = self._get_info(name, doc, width) + separator
self._write_info()
self._keyword_marker.reset_count()
def _write_info(self):
self.stdout.write(self._last_info)
def _get_info_width_and_separator(self, start_suite):
if start_suite:
return self.width, '\n'
return self.width - self._status_length - 1, ' '
def _get_info(self, name, doc, width):
if get_console_length(name) > width:
return pad_console_length(name, width)
doc = getshortdoc(doc, linesep=' ')
info = f'{name} :: {doc}' if doc else name
return pad_console_length(info, width)
[docs]
def suite_separator(self):
self._fill('=')
[docs]
def test_separator(self):
self._fill('-')
def _fill(self, char):
self.stdout.write(f'{char * self.width}\n')
[docs]
def status(self, status, clear=False):
if self._should_clear_markers(clear):
self._clear_status()
self.stdout.write('| ', flush=False)
self.stdout.highlight(status, flush=False)
self.stdout.write(' |\n')
def _should_clear_markers(self, clear):
return clear and self._keyword_marker.marking_enabled
def _clear_status(self):
self._clear_info()
self._write_info()
def _clear_info(self):
self.stdout.write(f"\r{' ' * self.width}\r")
self._keyword_marker.reset_count()
[docs]
def message(self, message):
if message:
self.stdout.write(message.strip() + '\n')
[docs]
def keyword_marker(self, status):
if self._keyword_marker.marker_count == self._status_length:
self._clear_status()
self._keyword_marker.reset_count()
self._keyword_marker.mark(status)
[docs]
def error(self, message, level, clear=False):
if self._should_clear_markers(clear):
self._clear_info()
self.stderr.error(message, level)
if self._should_clear_markers(clear):
self._write_info()
[docs]
def output(self, name, path):
self.stdout.write(f"{name+':':8} {path}\n")
[docs]
class KeywordMarker:
def __init__(self, highlighter, markers):
self.highlighter = highlighter
self.marking_enabled = self._marking_enabled(markers, highlighter)
self.marker_count = 0
def _marking_enabled(self, markers, highlighter):
options = {'AUTO': isatty(highlighter.stream),
'ON': True,
'OFF': False}
try:
return options[markers.upper()]
except KeyError:
raise DataError(f"Invalid console marker value '{markers}'. "
f"Available 'AUTO', 'ON' and 'OFF'.")
[docs]
def mark(self, status):
if self.marking_enabled:
marker, status = ('.', 'PASS') if status != 'FAIL' else ('F', 'FAIL')
self.highlighter.highlight(marker, status)
self.marker_count += 1
[docs]
def reset_count(self):
self.marker_count = 0