# 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 robot.errors import DataError
from robot.model import Message as BaseMessage
from robot.utils import get_timestamp, is_unicode, unic
LEVELS = {
'NONE' : 7,
'SKIP' : 6,
'FAIL' : 5,
'ERROR' : 4,
'WARN' : 3,
'INFO' : 2,
'DEBUG' : 1,
'TRACE' : 0,
}
[docs]class AbstractLogger(object):
def __init__(self, level='TRACE'):
self._is_logged = IsLogged(level)
[docs] def set_level(self, level):
return self._is_logged.set_level(level)
[docs] def trace(self, msg):
self.write(msg, 'TRACE')
[docs] def debug(self, msg):
self.write(msg, 'DEBUG')
[docs] def info(self, msg):
self.write(msg, 'INFO')
[docs] def warn(self, msg):
self.write(msg, 'WARN')
[docs] def fail(self, msg):
html = False
if msg.startswith("*HTML*"):
html = True
msg = msg[6:].lstrip()
self.write(msg, 'FAIL', html)
[docs] def skip(self, msg):
html = False
if msg.startswith("*HTML*"):
html = True
msg = msg[6:].lstrip()
self.write(msg, 'SKIP', html)
[docs] def error(self, msg):
self.write(msg, 'ERROR')
[docs] def write(self, message, level, html=False):
self.message(Message(message, level, html))
[docs] def message(self, msg):
raise NotImplementedError(self.__class__)
[docs]class Message(BaseMessage):
__slots__ = ['_message']
def __init__(self, message, level='INFO', html=False, timestamp=None):
message = self._normalize_message(message)
level, html = self._get_level_and_html(level, html)
timestamp = timestamp or get_timestamp()
BaseMessage.__init__(self, message, level, html, timestamp)
def _normalize_message(self, msg):
if callable(msg):
return msg
if not is_unicode(msg):
msg = unic(msg)
if '\r\n' in msg:
msg = msg.replace('\r\n', '\n')
return msg
def _get_level_and_html(self, level, html):
level = level.upper()
if level == 'HTML':
return 'INFO', True
if level not in LEVELS:
raise DataError("Invalid log level '%s'." % level)
return level, html
@property
def message(self):
self.resolve_delayed_message()
return self._message
@message.setter
def message(self, message):
self._message = message
[docs] def resolve_delayed_message(self):
if callable(self._message):
self._message = self._message()
[docs]class IsLogged(object):
def __init__(self, level):
self.level = level.upper()
self._int_level = self._level_to_int(level)
def __call__(self, level):
return self._level_to_int(level) >= self._int_level
[docs] def set_level(self, level):
old = self.level
self.__init__(level)
return old
def _level_to_int(self, level):
try:
return LEVELS[level.upper()]
except KeyError:
raise DataError("Invalid log level '%s'." % level)
[docs]class AbstractLoggerProxy(object):
_methods = None
_no_method = lambda *args: None
def __init__(self, logger, method_names=None, prefix=None):
self.logger = logger
for name in method_names or self._methods:
# Allow extending classes to implement some of the messages themselves.
if hasattr(self, name):
if hasattr(logger, name):
continue
target = logger
else:
target = self
setattr(target, name, self._get_method(logger, name, prefix))
def _get_method(self, logger, name, prefix):
for method_name in self._get_method_names(name, prefix):
if hasattr(logger, method_name):
return getattr(logger, method_name)
return self._no_method
def _get_method_names(self, name, prefix):
names = [name, self._toCamelCase(name)] if '_' in name else [name]
if prefix:
names += [prefix + name for name in names]
return names
def _toCamelCase(self, name):
parts = name.split('_')
return ''.join([parts[0]] + [part.capitalize() for part in parts[1:]])