[CalendarServer-changes] [15302] twext/trunk
source_changes at macosforge.org
source_changes at macosforge.org
Mon Nov 9 16:15:44 PST 2015
Revision: 15302
http://trac.calendarserver.org//changeset/15302
Author: cdaboo at apple.com
Date: 2015-11-09 16:15:44 -0800 (Mon, 09 Nov 2015)
Log Message:
-----------
Use Twisted's logger rather than our own. Still need to tweak behavior a bit though.
Modified Paths:
--------------
twext/trunk/.project
twext/trunk/twext/enterprise/jobs/queue.py
twext/trunk/twext/enterprise/jobs/workitem.py
twext/trunk/twext/python/log.py
twext/trunk/twext/python/test/test_log.py
Modified: twext/trunk/.project
===================================================================
--- twext/trunk/.project 2015-11-10 00:06:26 UTC (rev 15301)
+++ twext/trunk/.project 2015-11-10 00:15:44 UTC (rev 15302)
@@ -7,7 +7,7 @@
<project>python-ldap</project>
<project>sqlparse</project>
<project>Twisted</project>
- <project>zope.interface-4.0.3</project>
+ <project>zope.interface</project>
</projects>
<buildSpec>
<buildCommand>
Modified: twext/trunk/twext/enterprise/jobs/queue.py
===================================================================
--- twext/trunk/twext/enterprise/jobs/queue.py 2015-11-10 00:06:26 UTC (rev 15301)
+++ twext/trunk/twext/enterprise/jobs/queue.py 2015-11-10 00:15:44 UTC (rev 15302)
@@ -610,7 +610,7 @@
except Exception as e:
# Could not mark as failed - break out of the next job loop
log.error(
- "workCheck: Failed to mark failed new job:{}, {exc}",
+ "workCheck: Failed to mark failed new job:{jobID}, {exc}",
jobID=nextJob.jobID,
exc=e,
)
Modified: twext/trunk/twext/enterprise/jobs/workitem.py
===================================================================
--- twext/trunk/twext/enterprise/jobs/workitem.py 2015-11-10 00:06:26 UTC (rev 15301)
+++ twext/trunk/twext/enterprise/jobs/workitem.py 2015-11-10 00:15:44 UTC (rev 15302)
@@ -246,7 +246,7 @@
weight = "unchanged"
log.info(
"updateWorkTypes: '{workType}' priority: '{priority}' weight: '{weight}' ",
- workType=workType, priority=priority,
+ workType=workType, priority=priority, weight=weight,
)
Modified: twext/trunk/twext/python/log.py
===================================================================
--- twext/trunk/twext/python/log.py 2015-11-10 00:06:26 UTC (rev 15301)
+++ twext/trunk/twext/python/log.py 2015-11-10 00:15:44 UTC (rev 15302)
@@ -15,341 +15,57 @@
# limitations under the License.
##
-"""
-Classes and functions to do granular logging.
+from twisted.logger import Logger as _Logger, LogLevel, LogPublisher, \
+ FileLogObserver, FilteringLogObserver, LogLevelFilterPredicate, \
+ formatEventAsClassicLogText, formatTime
+from twisted.python import log
-Example usage in a module C{some.module}::
-
- from twext.python.log import Logger
- log = Logger()
-
- def handleData(data):
- log.debug("Got data: {data!r}.", data=data)
-
-Or in a class::
-
- from twext.python.log import Logger
-
- class Foo(object):
- log = Logger()
-
- def oops(self, data):
- self.log.error("Oops! Invalid data from server: {data!r}",
- data=data)
-
-C{Logger}s have namespaces, for which logging can be configured independently.
-Namespaces may be specified by passing in a C{namespace} argument to L{Logger}
-when instantiating it, but if none is given, the logger will derive its own
-namespace by using the module name of the callable that instantiated it, or, in
-the case of a class, by using the fully qualified name of the class.
-
-In the first example above, the namespace would be C{some.module}, and in the
-second example, it would be C{some.module.Foo}.
-"""
-
-__all__ = [
- "InvalidLogLevelError",
- "LogLevel",
- "formatEvent",
- "Logger",
- "LegacyLogger",
- "ILogObserver",
- "LogPublisher",
- "PredicateResult",
- "ILogFilterPredicate",
- "FilteringLogObserver",
- "LogLevelFilterPredicate",
- "LegacyLogObserver",
- "replaceTwistedLoggers",
-]
-
-
-
-import sys
-from sys import stdout, stderr
-from string import Formatter
-import inspect
-import logging
-import time
-
-from zope.interface import Interface, implementer
-from twisted.python.constants import NamedConstant, Names
-from twisted.python.failure import Failure
-from twisted.python.reflect import safe_str, safe_repr
-import twisted.python.log
-from twisted.python.log import msg as twistedLogMessage
-from twisted.python.log import addObserver, removeObserver
-
-
-
-OBSERVER_REMOVED = (
- "Temporarily removing observer {observer} due to exception: {e}"
-)
-
-
-
-#
-# Log level definitions
-#
-
-class InvalidLogLevelError(Exception):
+class Logger(_Logger):
"""
- Someone tried to use a L{LogLevel} that is unknown to the logging system.
+ Logging object.
"""
- def __init__(self, level):
- """
- @param level: a L{LogLevel}
- """
- super(InvalidLogLevelError, self).__init__(str(level))
- self.level = level
+ filterPublisher = None
+ filterObserver = None
+ filterPredicate = LogLevelFilterPredicate(defaultLogLevel=LogLevel.info)
-
-class LogLevel(Names):
- """
- Constants denoting log levels:
-
- - C{debug}: Information of use to a developer of the software, not
- generally of interest to someone running the software unless they are
- attempting to diagnose a software issue.
-
- - C{info}: Informational events: Routine information about the status of
- an application, such as incoming connections, startup of a subsystem,
- etc.
-
- - C{warn}: Warnings events: Events that may require greater attention than
- informational events but are not a systemic failure condition, such as
- authorization failures, bad data from a network client, etc.
-
- - C{error}: Error conditions: Events indicating a systemic failure, such
- as unhandled exceptions, loss of connectivity to a back-end database,
- etc.
- """
- debug = NamedConstant()
- info = NamedConstant()
- warn = NamedConstant()
- error = NamedConstant()
- critical = NamedConstant()
-
@classmethod
- def levelWithName(cls, name):
+ def makeFilteredFileLogObserver(cls, stream, withTime=True):
"""
- @param name: the name of a L{LogLevel}
-
- @return: the L{LogLevel} with the specified C{name}
+ For a child process that has its stdout captured by the master process to be logged by the master,
+ we strip out the time from the log entry since the master process will always add one. Setting
+ C{withTime} to L{False} will ensure no time is generated.
"""
- try:
- return cls.lookupByName(name)
- except ValueError:
- raise InvalidLogLevelError(name)
+ assert (
+ cls.filterPublisher is None and
+ cls.filterObserver is None
+ ), "Only call this once"
+ timeFormat = formatTime if withTime else lambda _: u""
+ cls.filterObserver = FileLogObserver(stream, lambda event: formatEventAsClassicLogText(event, formatTime=timeFormat))
+ cls.filterPublisher = LogPublisher(cls.filterObserver)
+ return cls.filterPublisher
+
@classmethod
- def _priorityForLevel(cls, constant):
- """
- We want log levels to have defined ordering - the order of definition -
- but they aren't value constants (the only value is the name). This is
- arguably a bug in Twisted, so this is just a workaround for U{until
- this is fixed in some way
- <https://twistedmatrix.com/trac/ticket/6523>}.
- """
- return cls._levelPriorities[constant]
+ def addFilteredObserver(cls, observer):
+ log.addObserver(FilteringLogObserver(
+ observer,
+ [cls.filterPredicate]
+ ))
-LogLevel._levelPriorities = dict(
- (constant, idx) for (idx, constant) in
- (enumerate(LogLevel.iterconstants()))
-)
-
-
-
-#
-# Mappings to Python's logging module
-#
-pythonLogLevelMapping = {
- LogLevel.debug: logging.DEBUG,
- LogLevel.info: logging.INFO,
- LogLevel.warn: logging.WARNING,
- LogLevel.error: logging.ERROR,
- LogLevel.critical: logging.CRITICAL,
-}
-
-
-
-##
-# Loggers
-##
-
-def formatEvent(event):
- """
- Formats an event as a L{unicode}, using the format in
- C{event["log_format"]}.
-
- This implementation should never raise an exception; if the formatting
- cannot be done, the returned string will describe the event generically so
- that a useful message is emitted regardless.
-
- @param event: a logging event
-
- @return: a L{unicode}
- """
- try:
- format = event.get("log_format", None)
-
- if format is None:
- raise ValueError("No log format provided")
-
- # Make sure format is unicode.
- if isinstance(format, bytes):
- # If we get bytes, assume it's UTF-8 bytes
- format = format.decode("utf-8")
-
- elif isinstance(format, unicode):
- pass
-
- else:
- raise TypeError("Log format must be unicode or bytes, not {0!r}"
- .format(format))
-
- return formatWithCall(format, event)
-
- except BaseException as e:
- return formatUnformattableEvent(event, e)
-
-
-
-def formatUnformattableEvent(event, error):
- """
- Formats an event as a L{unicode} that describes the event generically and a
- formatting error.
-
- @param event: a logging event
- @type dict: L{dict}
-
- @param error: the formatting error
- @type error: L{Exception}
-
- @return: a L{unicode}
- """
- try:
- return (
- u"Unable to format event {event!r}: {error}"
- .format(event=event, error=error)
- )
- except BaseException:
- # Yikes, something really nasty happened.
- #
- # Try to recover as much formattable data as possible; hopefully at
- # least the namespace is sane, which will help you find the offending
- # logger.
- failure = Failure()
-
- text = ", ".join(" = ".join((safe_repr(key), safe_repr(value)))
- for key, value in event.items())
-
- return (
- u"MESSAGE LOST: unformattable object logged: {error}\n"
- u"Recoverable data: {text}\n"
- u"Exception during formatting:\n{failure}"
- .format(error=safe_repr(error), failure=failure, text=text)
- )
-
-
-
-class Logger(object):
- """
- Logging object.
- """
-
- publisher = lambda e: None
-
-
- @staticmethod
- def _namespaceFromCallingContext():
- """
- Derive a namespace from the module containing the caller's caller.
-
- @return: a namespace
- """
- return inspect.currentframe().f_back.f_back.f_globals["__name__"]
-
-
- def __init__(self, namespace=None, source=None):
- """
- @param namespace: The namespace for this logger. Uses a dotted
- notation, as used by python modules. If not C{None}, then the name
- of the module of the caller is used.
-
- @param source: The object which is emitting events to this
- logger; this is automatically set on instances of a class
- if this L{Logger} is an attribute of that class.
- """
- if namespace is None:
- namespace = self._namespaceFromCallingContext()
-
- self.namespace = namespace
- self.source = source
-
-
- def __get__(self, oself, type=None):
- """
- When used as a descriptor, i.e.::
-
- # athing.py
- class Something(object):
- log = Logger()
- def hello(self):
- self.log.info("Hello")
-
- a L{Logger}'s namespace will be set to the name of the class it is
- declared on. In the above example, the namespace would be
- C{athing.Something}.
-
- Additionally, it's source will be set to the actual object referring to
- the L{Logger}. In the above example, C{Something.log.source} would be
- C{Something}, and C{Something().log.source} would be an instance of
- C{Something}.
- """
- if oself is None:
- source = type
- else:
- source = oself
-
- return self.__class__(
- '.'.join([type.__module__, type.__name__]),
- source
- )
-
-
- def __repr__(self):
- return "<%s %r>" % (self.__class__.__name__, self.namespace)
-
-
def emit(self, level, format=None, **kwargs):
"""
- Emit a log event to all log observers at the given level.
-
- @param level: a L{LogLevel}
-
- @param format: a message format using new-style (PEP 3101)
- formatting. The logging event (which is a L{dict}) is
- used to render this format string.
-
- @param kwargs: additional keyword parameters to include with
- the event.
+ Fix {Logger.emit} to work with our legacy use and handle utf-8 properly.
"""
- # FIXME: Updated Twisted supports 'in' on constants container
- if level not in LogLevel.iterconstants():
- self.failure(
- "Got invalid log level {invalidLevel!r} in {logger}.emit().",
- Failure(InvalidLogLevelError(level)),
- invalidLevel=level,
- logger=self,
- )
- # level = LogLevel.error
- # FIXME: continue to emit?
- return
+ # For old style logging kwargs will be empty but the format string could contain "{}", so
+ # We turn the format string into a kwarg
+ if len(kwargs) == 0:
+ kwargs["msg"] = format
+ format = "{msg}"
+
# Assume that any L{bytes} in the format or kwargs are utf-8 encoded strings
if format is not None and isinstance(format, bytes):
format = format.decode("utf-8")
@@ -357,662 +73,30 @@
if isinstance(v, bytes):
kwargs[k] = v.decode("utf-8")
- kwargs.update(
- log_logger=self, log_level=level, log_namespace=self.namespace,
- log_source=self.source, log_format=format, log_time=time.time(),
- )
+ super(Logger, self).emit(level, format=format, **kwargs)
- self.publisher(kwargs)
-
- def failure(self, format, failure=None, level=LogLevel.error, **kwargs):
+ def levels(self):
"""
- Log an failure and emit a traceback.
-
- For example::
-
- try:
- frob(knob)
- except Exception:
- log.failure("While frobbing {knob}", knob=knob)
-
- or::
-
- d = deferredFrob(knob)
- d.addErrback(lambda f: log.failure, "While frobbing {knob}",
- f, knob=knob)
-
- @param format: a message format using new-style (PEP 3101)
- formatting. The logging event (which is a L{dict}) is
- used to render this format string.
-
- @param failure: a L{Failure} to log. If C{None}, a L{Failure} is
- created from the exception in flight.
-
- @param level: a L{LogLevel} to use.
-
- @param kwargs: additional keyword parameters to include with the
- event.
+ Get the L{LogLevelFilterPredicate} associated with this logger.
"""
- if failure is None:
- failure = Failure()
+ return self.filterPredicate
- self.emit(level, format, log_failure=failure, **kwargs)
+# Always replace Twisted's legacy log beginner with one that does LogLevel filtering
+class FilteringLogBeginnerWrapper(object):
+ def __init__(self, beginner):
+ self.beginner = beginner
-class LegacyLogger(object):
- """
- A logging object that provides some compatibility with the
- L{twisted.python.log} module.
- """
- def __init__(self, logger=None):
- if logger is None:
- self.newStyleLogger = Logger(Logger._namespaceFromCallingContext())
- else:
- self.newStyleLogger = logger
+ def beginLoggingTo(
+ self, observers, discardBuffer=False, redirectStandardIO=True
+ ):
+ new_observers = []
+ for observer in observers:
+ new_observers.append(FilteringLogObserver(observer, [Logger.filterPredicate]))
+ self.beginner.beginLoggingTo(new_observers, discardBuffer, redirectStandardIO)
- def __getattribute__(self, name):
- try:
- return super(LegacyLogger, self).__getattribute__(name)
- except AttributeError:
- return getattr(twisted.python.log, name)
-
-
- def msg(self, *message, **kwargs):
- """
- This method is API-compatible with L{twisted.python.log.msg} and exists
- for compatibility with that API.
- """
- if message:
- message = " ".join(map(safe_str, message))
- else:
- message = None
- return self.newStyleLogger.emit(LogLevel.info, message, **kwargs)
-
-
- def err(self, _stuff=None, _why=None, **kwargs):
- """
- This method is API-compatible with L{twisted.python.log.err} and exists
- for compatibility with that API.
- """
- if _stuff is None:
- _stuff = Failure()
- elif isinstance(_stuff, Exception):
- _stuff = Failure(_stuff)
-
- if isinstance(_stuff, Failure):
- self.newStyleLogger.emit(LogLevel.error, failure=_stuff, why=_why,
- isError=1, **kwargs)
- else:
- # We got called with an invalid _stuff.
- self.newStyleLogger.emit(LogLevel.error, repr(_stuff), why=_why,
- isError=1, **kwargs)
-
-
-
-def bindEmit(level):
- doc = """
- Emit a log event at log level L{{{level}}}.
-
- @param format: a message format using new-style (PEP 3101)
- formatting. The logging event (which is a L{{dict}}) is used to
- render this format string.
-
- @param kwargs: additional keyword parameters to include with the
- event.
- """.format(level=level.name)
-
- #
- # Attach methods to Logger
- #
- def log_emit(self, format=None, **kwargs):
- self.emit(level, format, **kwargs)
-
- log_emit.__doc__ = doc
-
- setattr(Logger, level.name, log_emit)
-
-
-
-def _bindLevels():
- for level in LogLevel.iterconstants():
- bindEmit(level)
-
-_bindLevels()
-
-
-#
-# Observers
-#
-
-class ILogObserver(Interface):
- """
- An observer which can handle log events.
- """
-
- def __call__(event):
- """
- Log an event.
-
- @type event: C{dict} with (native) C{str} keys.
-
- @param event: A dictionary with arbitrary keys as defined by
- the application emitting logging events, as well as keys
- added by the logging system, with are:
- ...
- """
-
-
-
- at implementer(ILogObserver)
-class LogPublisher(object):
- """
- I{ILogObserver} that fans out events to other observers.
-
- Keeps track of a set of L{ILogObserver} objects and forwards
- events to each.
- """
- log = Logger()
-
- def __init__(self, *observers):
- self._observers = set(observers)
-
-
- @property
- def observers(self):
- return frozenset(self._observers)
-
-
- def addObserver(self, observer):
- """
- Registers an observer with this publisher.
-
- @param observer: An L{ILogObserver} to add.
- """
- self._observers.add(observer)
-
-
- def removeObserver(self, observer):
- """
- Unregisters an observer with this publisher.
-
- @param observer: An L{ILogObserver} to remove.
- """
- try:
- self._observers.remove(observer)
- except KeyError:
- pass
-
-
- def __call__(self, event):
- for observer in self.observers:
- try:
- observer(event)
- except BaseException as e:
- #
- # We have to remove the offending observer because
- # we're going to badmouth it to all of its friends
- # (other observers) and it might get offended and
- # raise again, causing an infinite loop.
- #
- self.removeObserver(observer)
- try:
- self.log.failure(OBSERVER_REMOVED, observer=observer, e=e)
- except BaseException:
- pass
- finally:
- self.addObserver(observer)
-
-
-
-class PredicateResult(Names):
- """
- Predicate results.
- """
- yes = NamedConstant() # Log this
- no = NamedConstant() # Don't log this
- maybe = NamedConstant() # No opinion
-
-
-
-class ILogFilterPredicate(Interface):
- """
- A predicate that determined whether an event should be logged.
- """
-
- def __call__(event):
- """
- Determine whether an event should be logged.
-
- @returns: a L{PredicateResult}.
- """
-
-
-
- at implementer(ILogObserver)
-class FilteringLogObserver(object):
- """
- L{ILogObserver} that wraps another L{ILogObserver}, but filters
- out events based on applying a series of L{ILogFilterPredicate}s.
- """
-
- def __init__(self, observer, predicates):
- """
- @param observer: an L{ILogObserver} to which this observer
- will forward events.
-
- @param predicates: an ordered iterable of predicates to apply
- to events before forwarding to the wrapped observer.
- """
- self.observer = observer
- self.predicates = list(predicates)
-
-
- def shouldLogEvent(self, event):
- """
- Determine whether an event should be logged, based
- C{self.predicates}.
-
- @param event: an event
- """
- for predicate in self.predicates:
- result = predicate(event)
- if result == PredicateResult.yes:
- return True
- if result == PredicateResult.no:
- return False
- if result == PredicateResult.maybe:
- continue
- raise TypeError("Invalid predicate result: {0!r}".format(result))
- return True
-
-
- def __call__(self, event):
- if self.shouldLogEvent(event):
- self.observer(event)
-
-
-
- at implementer(ILogFilterPredicate)
-class LogLevelFilterPredicate(object):
- """
- L{ILogFilterPredicate} that filters out events with a log level
- lower than the log level for the event's namespace.
-
- Events that not not have a log level or namespace are also dropped.
- """
-
- def __init__(self):
- # FIXME: Make this a class variable. But that raises an
- # _initializeEnumerants constants error in Twisted 12.2.0.
- self.defaultLogLevel = LogLevel.info
-
- self._logLevelsByNamespace = {}
- self.clearLogLevels()
-
-
- def logLevelForNamespace(self, namespace):
- """
- @param namespace: a logging namespace, or C{None} for the default
- namespace.
-
- @return: the L{LogLevel} for the specified namespace.
- """
- if not namespace:
- return self._logLevelsByNamespace[None]
-
- if namespace in self._logLevelsByNamespace:
- return self._logLevelsByNamespace[namespace]
-
- segments = namespace.split(".")
- index = len(segments) - 1
-
- while index > 0:
- namespace = ".".join(segments[:index])
- if namespace in self._logLevelsByNamespace:
- return self._logLevelsByNamespace[namespace]
- index -= 1
-
- return self._logLevelsByNamespace[None]
-
-
- def setLogLevelForNamespace(self, namespace, level):
- """
- Sets the global log level for a logging namespace.
-
- @param namespace: a logging namespace
-
- @param level: the L{LogLevel} for the given namespace.
- """
- if level not in LogLevel.iterconstants():
- raise InvalidLogLevelError(level)
-
- if namespace:
- self._logLevelsByNamespace[namespace] = level
- else:
- self._logLevelsByNamespace[None] = level
-
-
- def clearLogLevels(self):
- """
- Clears all global log levels to the default.
- """
- self._logLevelsByNamespace.clear()
- self._logLevelsByNamespace[None] = self.defaultLogLevel
-
-
- def __call__(self, event):
- level = event.get("log_level", None)
- namespace = event.get("log_namespace", None)
-
- if (
- level is None or
- namespace is None or
- LogLevel._priorityForLevel(level) <
- LogLevel._priorityForLevel(self.logLevelForNamespace(namespace))
- ):
- return PredicateResult.no
-
- return PredicateResult.maybe
-
-
-
- at implementer(ILogObserver)
-class LegacyLogObserver(object):
- """
- L{ILogObserver} that wraps an L{ILegacyLogObserver}.
- """
-
- def __init__(self, legacyObserver):
- """
- @param legacyObserver: an L{ILegacyLogObserver} to which this
- observer will forward events.
- """
- self.legacyObserver = legacyObserver
-
-
- def __call__(self, event):
- prefix = "[{log_namespace}#{log_level.name}] ".format(**event)
-
- level = event["log_level"]
-
- #
- # Twisted's logging supports indicating a python log level, so let's
- # provide the equivalent to our logging levels.
- #
- if level in pythonLogLevelMapping:
- event["logLevel"] = pythonLogLevelMapping[level]
-
- # Format new style -> old style
- if event["log_format"]:
- #
- # Create an object that implements __str__() in order to
- # defer the work of formatting until it's needed by a
- # legacy log observer.
- #
- class LegacyFormatStub(object):
- def __str__(oself):
- return formatEvent(event).encode("utf-8")
-
- event["format"] = prefix + "%(log_legacy)s"
- event["log_legacy"] = LegacyFormatStub()
-
- # log.failure() -> isError blah blah
- if "log_failure" in event:
- event["failure"] = event["log_failure"]
- event["isError"] = 1
- event["why"] = "{prefix}{message}".format(
- prefix=prefix, message=formatEvent(event)
- )
-
- self.legacyObserver(**event)
-
-
-
-# FIXME: This could have a better name.
-class DefaultLogPublisher(object):
- """
- This observer sets up a set of chained observers as follows:
-
- 1. B{rootPublisher} - a L{LogPublisher}
-
- 2. B{filters}: a L{FilteringLogObserver} that filters out messages
- using a L{LogLevelFilterPredicate}
-
- 3. B{filteredPublisher} - a L{LogPublisher}
-
- 4. B{legacyLogObserver} - a L{LegacyLogObserver} wired up to
- L{twisted.python.log.msg}. This allows any observers registered
- with Twisted's logging (that is, most observers in presently use) to
- receive (filtered) events.
-
- The purpose of this class is to provide a default log observer with
- sufficient hooks to enable applications to add observers that can either
- receive all log messages, or only log messages that are configured to pass
- though the L{LogLevelFilterPredicate}::
-
- from twext.python.log import Logger, ILogObserver
-
- log = Logger()
-
- @implementer(ILogObserver)
- class AMPObserver(object):
- def __call__(self, event):
- # eg.: Hold events in a ring buffer and expose them via AMP.
- ...
-
- @implementer(ILogObserver)
- class FileObserver(object):
- def __call__(self, event):
- # eg.: Take events and write them into a file.
- ...
-
- # Send all events to the AMPObserver
- log.publisher.addObserver(AMPObserver(), filtered=False)
-
- # Send filtered events to the FileObserver
- log.publisher.addObserver(AMPObserver())
-
- With no observers added, the default behavior is that the legacy Twisted
- logging system sees messages as controlled by L{LogLevelFilterPredicate}.
- """
-
- def __init__(self):
- self.legacyLogObserver = LegacyLogObserver(twistedLogMessage)
- self.filteredPublisher = LogPublisher(self.legacyLogObserver)
- self.levels = LogLevelFilterPredicate()
- self.filters = FilteringLogObserver(
- self.filteredPublisher, (self.levels,)
- )
- self.rootPublisher = LogPublisher(self.filters)
-
-
- def addObserver(self, observer, filtered=True):
- """
- Registers an observer with this publisher.
-
- @param observer: An L{ILogObserver} to add.
-
- @param filtered: If true, registers C{observer} after filters are
- applied; otherwise C{observer} will get all events.
- """
- if filtered:
- self.filteredPublisher.addObserver(observer)
- self.rootPublisher.removeObserver(observer)
- else:
- self.rootPublisher.addObserver(observer)
- self.filteredPublisher.removeObserver(observer)
-
-
- def removeObserver(self, observer):
- """
- Unregisters an observer with this publisher.
-
- @param observer: An L{ILogObserver} to remove.
- """
- self.rootPublisher.removeObserver(observer)
- self.filteredPublisher.removeObserver(observer)
-
-
- def __call__(self, event):
- self.rootPublisher(event)
-
-
-
-Logger.publisher = DefaultLogPublisher()
-
-
-
-#
-# Utilities
-#
-
-class CallMapping(object):
- def __init__(self, submapping):
- self._submapping = submapping
-
-
- def __getitem__(self, key):
- callit = key.endswith(u"()")
- realKey = key[:-2] if callit else key
- value = self._submapping[realKey]
- if callit:
- value = value()
- return value
-
-
-
-def formatWithCall(formatString, mapping):
- """
- Format a string like L{unicode.format}, but:
-
- - taking only a name mapping; no positional arguments
-
- - with the additional syntax that an empty set of parentheses
- correspond to a formatting item that should be called, and its result
- C{str}'d, rather than calling C{str} on the element directly as
- normal.
-
- For example::
-
- >>> formatWithCall("{string}, {function()}.",
- ... dict(string="just a string",
- ... function=lambda: "a function"))
- 'just a string, a function.'
-
- @param formatString: A PEP-3101 format string.
- @type formatString: L{unicode}
-
- @param mapping: A L{dict}-like object to format.
-
- @return: The string with formatted values interpolated.
- @rtype: L{unicode}
- """
- return unicode(
- theFormatter.vformat(formatString, (), CallMapping(mapping))
- )
-
-theFormatter = Formatter()
-
-
-def replaceTwistedLoggers():
- """
- Visit all Python modules that have been loaded and:
-
- - replace L{twisted.python.log} with a L{LegacyLogger}
-
- - replace L{twisted.python.log.msg} with a L{LegacyLogger}'s C{msg}
-
- - replace L{twisted.python.log.err} with a L{LegacyLogger}'s C{err}
- """
- log = Logger()
-
- for moduleName, module in sys.modules.iteritems():
- # Oddly, this happens
- if module is None:
- continue
-
- # Don't patch Twisted's logging module
- if module in (twisted.python, twisted.python.log):
- continue
-
- # Don't patch this module
- if moduleName is __name__:
- continue
-
- try:
- for name, obj in module.__dict__.iteritems():
- try:
- newLogger = Logger(namespace=module.__name__)
- except AttributeError:
- # Can't look up __name__. A hack in the "six" module causes
- # this. Skip the module.
- # See https://trac.calendarserver.org/ticket/832
- continue
-
- legacyLogger = LegacyLogger(logger=newLogger)
-
- if obj is twisted.python.log:
- log.info(
- "Replacing Twisted log module object {0} in {1}"
- .format(name, module.__name__)
- )
- setattr(module, name, legacyLogger)
-
- elif obj is twisted.python.log.msg:
- log.info(
- "Replacing Twisted log.msg object {0} in {1}"
- .format(name, module.__name__)
- )
- setattr(module, name, legacyLogger.msg)
-
- elif obj is twisted.python.log.err:
- log.info(
- "Replacing Twisted log.err object {0} in {1}"
- .format(name, module.__name__)
- )
- setattr(module, name, legacyLogger.err)
- except RuntimeError as e:
- # Python could use more specific exceptions, eh.
- # What we mean to catch is:
- # RuntimeError: dictionary changed size during iteration
- log.error(
- "Unable to replace twisted loggers for module {module}: "
- "{error}",
- module=module, error=e
- )
-
-
-
-# FIXME: This may not be needed; look into removing it.
-
-class StandardIOObserver(object):
- """
- (Legacy) log observer that writes to standard I/O.
- """
- def emit(self, eventDict):
- text = None
-
- if eventDict["isError"]:
- output = stderr
- if "failure" in eventDict:
- text = eventDict["failure"].getTraceback()
- else:
- output = stdout
-
- if not text:
- text = formatEvent(eventDict) + "\n"
-
- output.write(text)
- output.flush()
-
-
- def start(self):
- addObserver(self.emit)
-
-
- def stop(self):
- removeObserver(self.emit)
+log.theLogPublisher._logBeginner = FilteringLogBeginnerWrapper(log.theLogPublisher._logBeginner)
Modified: twext/trunk/twext/python/test/test_log.py
===================================================================
--- twext/trunk/twext/python/test/test_log.py 2015-11-10 00:06:26 UTC (rev 15301)
+++ twext/trunk/twext/python/test/test_log.py 2015-11-10 00:15:44 UTC (rev 15302)
@@ -14,1016 +14,83 @@
# limitations under the License.
##
-from zope.interface.verify import verifyObject, BrokenMethodImplementation
-
-from twisted.python import log as twistedLogging
-from twisted.python.failure import Failure
+from cStringIO import StringIO
+from twext.python.log import Logger
+from twisted.logger import LogLevel, LogPublisher, textFileLogObserver
from twisted.trial import unittest
-from twext.python.log import (
- LogLevel, InvalidLogLevelError,
- pythonLogLevelMapping,
- formatEvent, formatUnformattableEvent, formatWithCall,
- Logger, LegacyLogger,
- ILogObserver, LogPublisher, DefaultLogPublisher,
- FilteringLogObserver, PredicateResult,
- LogLevelFilterPredicate, OBSERVER_REMOVED
-)
-
-
-
-defaultLogLevel = LogLevelFilterPredicate().defaultLogLevel
-clearLogLevels = Logger.publisher.levels.clearLogLevels
-logLevelForNamespace = Logger.publisher.levels.logLevelForNamespace
-setLogLevelForNamespace = Logger.publisher.levels.setLogLevelForNamespace
-
-
-
-class TestLogger(Logger):
- def emit(self, level, format=None, **kwargs):
- if False:
- print "*" * 60
- print "level =", level
- print "format =", format
- for key, value in kwargs.items():
- print key, "=", value
- print "*" * 60
-
- def observer(event):
- self.event = event
-
- twistedLogging.addObserver(observer)
- try:
- Logger.emit(self, level, format, **kwargs)
- finally:
- twistedLogging.removeObserver(observer)
-
- self.emitted = {
- "level": level,
- "format": format,
- "kwargs": kwargs,
- }
-
-
-
-class TestLegacyLogger(LegacyLogger):
- def __init__(self, logger=TestLogger()):
- LegacyLogger.__init__(self, logger=logger)
-
-
-
-class LogComposedObject(object):
+class LogPublisherTests(unittest.TestCase):
"""
- Just a regular object.
- """
- log = TestLogger()
-
- def __init__(self, state=None):
- self.state = state
-
-
- def __str__(self):
- return "<LogComposedObject {state}>".format(state=self.state)
-
-
-
-class SetUpTearDown(object):
- def setUp(self):
- super(SetUpTearDown, self).setUp()
- clearLogLevels()
-
-
- def tearDown(self):
- super(SetUpTearDown, self).tearDown()
- clearLogLevels()
-
-
-
-class LoggingTests(SetUpTearDown, unittest.TestCase):
- """
- General module tests.
- """
-
- def test_levelWithName(self):
- """
- Look up log level by name.
- """
- for level in LogLevel.iterconstants():
- self.assertIdentical(LogLevel.levelWithName(level.name), level)
-
-
- def test_levelWithInvalidName(self):
- """
- You can't make up log level names.
- """
- bogus = "*bogus*"
- try:
- LogLevel.levelWithName(bogus)
- except InvalidLogLevelError as e:
- self.assertIdentical(e.level, bogus)
- else:
- self.fail("Expected InvalidLogLevelError.")
-
-
- def test_defaultLogLevel(self):
- """
- Default log level is used.
- """
- self.failUnless(logLevelForNamespace(None), defaultLogLevel)
- self.failUnless(logLevelForNamespace(""), defaultLogLevel)
- self.failUnless(logLevelForNamespace("rocker.cool.namespace"),
- defaultLogLevel)
-
-
- def test_setLogLevel(self):
- """
- Setting and retrieving log levels.
- """
- setLogLevelForNamespace(None, LogLevel.error)
- setLogLevelForNamespace("twext.web2", LogLevel.debug)
- setLogLevelForNamespace("twext.web2.dav", LogLevel.warn)
-
- self.assertEquals(logLevelForNamespace(None),
- LogLevel.error)
- self.assertEquals(logLevelForNamespace("twisted"),
- LogLevel.error)
- self.assertEquals(logLevelForNamespace("twext.web2"),
- LogLevel.debug)
- self.assertEquals(logLevelForNamespace("twext.web2.dav"),
- LogLevel.warn)
- self.assertEquals(logLevelForNamespace("twext.web2.dav.test"),
- LogLevel.warn)
- self.assertEquals(logLevelForNamespace("twext.web2.dav.test1.test2"),
- LogLevel.warn)
-
-
- def test_setInvalidLogLevel(self):
- """
- Can't pass invalid log levels to setLogLevelForNamespace().
- """
- self.assertRaises(InvalidLogLevelError, setLogLevelForNamespace,
- "twext.web2", object())
-
- # Level must be a constant, not the name of a constant
- self.assertRaises(InvalidLogLevelError, setLogLevelForNamespace,
- "twext.web2", "debug")
-
-
- def test_clearLogLevels(self):
- """
- Clearing log levels.
- """
- setLogLevelForNamespace("twext.web2", LogLevel.debug)
- setLogLevelForNamespace("twext.web2.dav", LogLevel.error)
-
- clearLogLevels()
-
- self.assertEquals(logLevelForNamespace("twisted"), defaultLogLevel)
- self.assertEquals(logLevelForNamespace("twext.web2"), defaultLogLevel)
- self.assertEquals(logLevelForNamespace("twext.web2.dav"),
- defaultLogLevel)
- self.assertEquals(logLevelForNamespace("twext.web2.dav.test"),
- defaultLogLevel)
- self.assertEquals(logLevelForNamespace("twext.web2.dav.test1.test2"),
- defaultLogLevel)
-
-
- def test_namespace_default(self):
- """
- Default namespace is module name.
- """
- log = Logger()
- self.assertEquals(log.namespace, __name__)
-
-
- def test_formatWithCall(self):
- """
- L{formatWithCall} is an extended version of L{unicode.format} that will
- interpret a set of parentheses "C{()}" at the end of a format key to
- mean that the format key ought to be I{called} rather than stringified.
- """
- self.assertEquals(
- formatWithCall(
- u"Hello, {world}. {callme()}.",
- dict(world="earth", callme=lambda: "maybe")
- ),
- "Hello, earth. maybe."
- )
- self.assertEquals(
- formatWithCall(
- u"Hello, {repr()!r}.",
- dict(repr=lambda: "repr")
- ),
- "Hello, 'repr'."
- )
-
-
- def test_formatEvent(self):
- """
- L{formatEvent} will format an event according to several rules:
-
- - A string with no formatting instructions will be passed straight
- through.
-
- - PEP 3101 strings will be formatted using the keys and values of
- the event as named fields.
-
- - PEP 3101 keys ending with C{()} will be treated as instructions
- to call that key (which ought to be a callable) before
- formatting.
-
- L{formatEvent} will always return L{unicode}, and if given
- bytes, will always treat its format string as UTF-8 encoded.
- """
- def format(log_format, **event):
- event["log_format"] = log_format
- result = formatEvent(event)
- self.assertIdentical(type(result), unicode)
- return result
-
- self.assertEquals(u"", format(b""))
- self.assertEquals(u"", format(u""))
- self.assertEquals(u"abc", format("{x}", x="abc"))
- self.assertEquals(u"no, yes.",
- format("{not_called}, {called()}.",
- not_called="no", called=lambda: "yes"))
- self.assertEquals(u'S\xe1nchez', format("S\xc3\xa1nchez"))
- self.assertIn(u"Unable to format event", format(b"S\xe1nchez"))
- self.assertIn(u"Unable to format event",
- format(b"S{a}nchez", a=b"\xe1"))
- self.assertIn(u"S'\\xe1'nchez",
- format(b"S{a!r}nchez", a=b"\xe1"))
-
-
- def test_formatEventNoFormat(self):
- """
- Formatting an event with no format.
- """
- event = dict(foo=1, bar=2)
- result = formatEvent(event)
-
- self.assertIn("Unable to format event", result)
- self.assertIn(repr(event), result)
-
-
- def test_formatEventWeirdFormat(self):
- """
- Formatting an event with a bogus format.
- """
- event = dict(log_format=object(), foo=1, bar=2)
- result = formatEvent(event)
-
- self.assertIn("Log format must be unicode or bytes", result)
- self.assertIn(repr(event), result)
-
-
- def test_formatUnformattableEvent(self):
- """
- Formatting an event that's just plain out to get us.
- """
- event = dict(log_format="{evil()}", evil=lambda: 1 / 0)
- result = formatEvent(event)
-
- self.assertIn("Unable to format event", result)
- self.assertIn(repr(event), result)
-
-
- def test_formatUnformattableEventWithUnformattableKey(self):
- """
- Formatting an unformattable event that has an unformattable key.
- """
- event = {
- "log_format": "{evil()}",
- "evil": lambda: 1 / 0,
- Unformattable(): "gurk",
- }
- result = formatEvent(event)
- self.assertIn("MESSAGE LOST: unformattable object logged:", result)
- self.assertIn("Recoverable data:", result)
- self.assertIn("Exception during formatting:", result)
-
-
- def test_formatUnformattableEventWithUnformattableValue(self):
- """
- Formatting an unformattable event that has an unformattable value.
- """
- event = dict(
- log_format="{evil()}",
- evil=lambda: 1 / 0,
- gurk=Unformattable(),
- )
- result = formatEvent(event)
- self.assertIn("MESSAGE LOST: unformattable object logged:", result)
- self.assertIn("Recoverable data:", result)
- self.assertIn("Exception during formatting:", result)
-
-
- def test_formatUnformattableEventWithUnformattableErrorOMGWillItStop(self):
- """
- Formatting an unformattable event that has an unformattable value.
- """
- event = dict(
- log_format="{evil()}",
- evil=lambda: 1 / 0,
- recoverable="okay",
- )
- # Call formatUnformattableEvent() directly with a bogus exception.
- result = formatUnformattableEvent(event, Unformattable())
- self.assertIn("MESSAGE LOST: unformattable object logged:", result)
- self.assertIn(repr("recoverable") + " = " + repr("okay"), result)
-
-
-
-class LoggerTests(SetUpTearDown, unittest.TestCase):
- """
Tests for L{Logger}.
"""
- def test_repr(self):
+ def test_old_style(self):
"""
- repr() on Logger
+ L{Logger} handles old style log strings.
"""
- namespace = "bleargh"
- log = Logger(namespace)
- self.assertEquals(repr(log), "<Logger {0}>".format(repr(namespace)))
+ observer = LogPublisher()
+ observed = []
+ observer.addObserver(observed.append)
- def test_namespace_attribute(self):
- """
- Default namespace for classes using L{Logger} as a descriptor is the
- class name they were retrieved from.
- """
- obj = LogComposedObject()
- self.assertEquals(obj.log.namespace,
- "twext.python.test.test_log.LogComposedObject")
- self.assertEquals(LogComposedObject.log.namespace,
- "twext.python.test.test_log.LogComposedObject")
- self.assertIdentical(LogComposedObject.log.source, LogComposedObject)
- self.assertIdentical(obj.log.source, obj)
- self.assertIdentical(Logger().source, None)
+ sio = StringIO()
+ observer.addObserver(textFileLogObserver(sio))
+ logger = Logger(observer=observer)
- def test_sourceAvailableForFormatting(self):
- """
- On instances that have a L{Logger} class attribute, the C{log_source}
- key is available to format strings.
- """
- obj = LogComposedObject("hello")
- log = obj.log
- log.error("Hello, {log_source}.")
+ index = 0
+ logger.info("test")
+ self.assertEqual(observed[index]["log_level"], LogLevel.info)
+ self.assertEqual(observed[index]["log_format"], u"{msg}")
+ self.assertEqual(observed[index]["msg"], u"test")
+ self.assertEqual(sio.getvalue().splitlines()[index].split("#info] ")[1], "test")
- self.assertIn("log_source", log.event)
- self.assertEquals(log.event["log_source"], obj)
+ index += 1
+ logger.info("test {}")
+ self.assertEqual(observed[index]["log_level"], LogLevel.info)
+ self.assertEqual(observed[index]["log_format"], u"{msg}")
+ self.assertEqual(observed[index]["msg"], u"test {}")
+ self.assertEqual(sio.getvalue().splitlines()[index].split("#info] ")[1], "test {}")
- stuff = formatEvent(log.event)
- self.assertIn("Hello, <LogComposedObject hello>.", stuff)
+ index += 1
+ logger.info("test {foo}")
+ self.assertEqual(observed[index]["log_level"], LogLevel.info)
+ self.assertEqual(observed[index]["log_format"], u"{msg}")
+ self.assertEqual(observed[index]["msg"], u"test {foo}")
+ self.assertEqual(sio.getvalue().splitlines()[index].split("#info] ")[1], "test {foo}")
- def test_basic_Logger(self):
+ def test_utf8(self):
"""
- Test that log levels and messages are emitted correctly for
- Logger.
+ L{Logger} handles utf8 log strings and format args.
"""
- # FIXME: Need a basic test like this for logger attached to a class.
- # At least: source should not be None in that case.
+ observer = LogPublisher()
- log = TestLogger()
+ observed = []
+ observer.addObserver(observed.append)
- for level in LogLevel.iterconstants():
- format = "This is a {level_name} message"
- message = format.format(level_name=level.name)
+ sio = StringIO()
+ observer.addObserver(textFileLogObserver(sio))
- method = getattr(log, level.name)
- method(format, junk=message, level_name=level.name)
+ logger = Logger(observer=observer)
- # Ensure that test_emit got called with expected arguments
- self.assertEquals(log.emitted["level"], level)
- self.assertEquals(log.emitted["format"], format)
- self.assertEquals(log.emitted["kwargs"]["junk"], message)
+ index = 0
+ logger.info("t\xc3\xa9st")
+ self.assertEqual(observed[index]["log_level"], LogLevel.info)
+ self.assertEqual(observed[index]["log_format"], u"{msg}")
+ self.assertEqual(observed[index]["msg"], u"t\xe9st")
+ self.assertEqual(sio.getvalue().splitlines()[index].split("#info] ")[1], "t\xc3\xa9st")
- if level >= logLevelForNamespace(log.namespace):
- self.assertTrue(hasattr(log, "event"), "No event observed.")
- self.assertEquals(log.event["log_format"], format)
- self.assertEquals(log.event["log_level"], level)
- self.assertEquals(log.event["log_namespace"], __name__)
- self.assertEquals(log.event["log_source"], None)
+ index += 1
+ logger.info("{str}", str="t\xc3\xa9st")
+ self.assertEqual(observed[index]["log_level"], LogLevel.info)
+ self.assertEqual(observed[index]["log_format"], u"{str}")
+ self.assertEqual(observed[index]["str"], u"t\xe9st")
+ self.assertEqual(sio.getvalue().splitlines()[index].split("#info] ")[1], "t\xc3\xa9st")
- self.assertEquals(log.event["logLevel"],
- pythonLogLevelMapping[level])
-
- self.assertEquals(log.event["junk"], message)
-
- # FIXME: this checks the end of message because we do
- # formatting in emit()
- self.assertEquals(
- formatEvent(log.event),
- message
- )
- else:
- self.assertFalse(hasattr(log, "event"))
-
-
- def test_defaultFailure(self):
- """
- Test that log.failure() emits the right data.
- """
- log = TestLogger()
- try:
- raise RuntimeError("baloney!")
- except RuntimeError:
- log.failure("Whoops")
-
- #
- # log.failure() will cause trial to complain, so here we check that
- # trial saw the correct error and remove it from the list of things to
- # complain about.
- #
- errors = self.flushLoggedErrors(RuntimeError)
- self.assertEquals(len(errors), 1)
-
- self.assertEquals(log.emitted["level"], LogLevel.error)
- self.assertEquals(log.emitted["format"], "Whoops")
-
-
- def test_conflicting_kwargs(self):
- """
- Make sure that kwargs conflicting with args don't pass through.
- """
- log = TestLogger()
-
- log.warn(
- "*",
- log_format="#",
- log_level=LogLevel.error,
- log_namespace="*namespace*",
- log_source="*source*",
- )
-
- # FIXME: Should conflicts log errors?
-
- self.assertEquals(log.event["log_format"], "*")
- self.assertEquals(log.event["log_level"], LogLevel.warn)
- self.assertEquals(log.event["log_namespace"], log.namespace)
- self.assertEquals(log.event["log_source"], None)
-
-
- def test_logInvalidLogLevel(self):
- """
- Test passing in a bogus log level to C{emit()}.
- """
- log = TestLogger()
-
- log.emit("*bogus*")
-
- errors = self.flushLoggedErrors(InvalidLogLevelError)
- self.assertEquals(len(errors), 1)
-
-
-
-class LogPublisherTests(SetUpTearDown, unittest.TestCase):
- """
- Tests for L{LogPublisher}.
- """
-
- def test_interface(self):
- """
- L{LogPublisher} is an L{ILogObserver}.
- """
- publisher = LogPublisher()
- try:
- verifyObject(ILogObserver, publisher)
- except BrokenMethodImplementation as e:
- self.fail(e)
-
-
- def test_observers(self):
- """
- L{LogPublisher.observers} returns the observers.
- """
- o1 = lambda e: None
- o2 = lambda e: None
-
- publisher = LogPublisher(o1, o2)
- self.assertEquals(set((o1, o2)), set(publisher.observers))
-
-
- def test_addObserver(self):
- """
- L{LogPublisher.addObserver} adds an observer.
- """
- o1 = lambda e: None
- o2 = lambda e: None
- o3 = lambda e: None
-
- publisher = LogPublisher(o1, o2)
- publisher.addObserver(o3)
- self.assertEquals(set((o1, o2, o3)), set(publisher.observers))
-
-
- def test_removeObserver(self):
- """
- L{LogPublisher.removeObserver} removes an observer.
- """
- o1 = lambda e: None
- o2 = lambda e: None
- o3 = lambda e: None
-
- publisher = LogPublisher(o1, o2, o3)
- publisher.removeObserver(o2)
- self.assertEquals(set((o1, o3)), set(publisher.observers))
-
-
- def test_removeObserverNotRegistered(self):
- """
- L{LogPublisher.removeObserver} removes an observer that is not
- registered.
- """
- o1 = lambda e: None
- o2 = lambda e: None
- o3 = lambda e: None
-
- publisher = LogPublisher(o1, o2)
- publisher.removeObserver(o3)
- self.assertEquals(set((o1, o2)), set(publisher.observers))
-
-
- def test_fanOut(self):
- """
- L{LogPublisher} calls its observers.
- """
- event = dict(foo=1, bar=2)
-
- events1 = []
- events2 = []
- events3 = []
-
- o1 = lambda e: events1.append(e)
- o2 = lambda e: events2.append(e)
- o3 = lambda e: events3.append(e)
-
- publisher = LogPublisher(o1, o2, o3)
- publisher(event)
- self.assertIn(event, events1)
- self.assertIn(event, events2)
- self.assertIn(event, events3)
-
-
- def test_observerRaises(self):
- nonTestEvents = []
- Logger.publisher.addObserver(lambda e: nonTestEvents.append(e))
-
- event = dict(foo=1, bar=2)
- exception = RuntimeError("ARGH! EVIL DEATH!")
-
- events = []
-
- def observer(event):
- events.append(event)
- raise exception
-
- publisher = LogPublisher(observer)
- publisher(event)
-
- # Verify that the observer saw my event
- self.assertIn(event, events)
-
- # Verify that the observer raised my exception
- errors = self.flushLoggedErrors(exception.__class__)
- self.assertEquals(len(errors), 1)
- self.assertIdentical(errors[0].value, exception)
-
- # Verify that the exception was logged
- for event in nonTestEvents:
- if (
- event.get("log_format", None) == OBSERVER_REMOVED and
- getattr(event.get("failure", None), "value") is exception
- ):
- break
- else:
- self.fail("Observer raised an exception "
- "and the exception was not logged.")
-
-
- def test_observerRaisesAndLoggerHatesMe(self):
- nonTestEvents = []
- Logger.publisher.addObserver(lambda e: nonTestEvents.append(e))
-
- event = dict(foo=1, bar=2)
- exception = RuntimeError("ARGH! EVIL DEATH!")
-
- def observer(event):
- raise RuntimeError("Sad panda")
-
- class GurkLogger(Logger):
- def failure(self, *args, **kwargs):
- raise exception
-
- publisher = LogPublisher(observer)
- publisher.log = GurkLogger()
- publisher(event)
-
- # Here, the lack of an exception thus far is a success, of sorts
-
-
-
-class DefaultLogPublisherTests(SetUpTearDown, unittest.TestCase):
- def test_addObserver(self):
- o1 = lambda e: None
- o2 = lambda e: None
- o3 = lambda e: None
-
- publisher = DefaultLogPublisher()
- publisher.addObserver(o1)
- publisher.addObserver(o2, filtered=True)
- publisher.addObserver(o3, filtered=False)
-
- self.assertEquals(
- set((o1, o2, publisher.legacyLogObserver)),
- set(publisher.filteredPublisher.observers),
- "Filtered observers do not match expected set"
- )
- self.assertEquals(
- set((o3, publisher.filters)),
- set(publisher.rootPublisher.observers),
- "Root observers do not match expected set"
- )
-
-
- def test_addObserverAgain(self):
- o1 = lambda e: None
- o2 = lambda e: None
- o3 = lambda e: None
-
- publisher = DefaultLogPublisher()
- publisher.addObserver(o1)
- publisher.addObserver(o2, filtered=True)
- publisher.addObserver(o3, filtered=False)
-
- # Swap filtered-ness of o2 and o3
- publisher.addObserver(o1)
- publisher.addObserver(o2, filtered=False)
- publisher.addObserver(o3, filtered=True)
-
- self.assertEquals(
- set((o1, o3, publisher.legacyLogObserver)),
- set(publisher.filteredPublisher.observers),
- "Filtered observers do not match expected set"
- )
- self.assertEquals(
- set((o2, publisher.filters)),
- set(publisher.rootPublisher.observers),
- "Root observers do not match expected set"
- )
-
-
- def test_removeObserver(self):
- o1 = lambda e: None
- o2 = lambda e: None
- o3 = lambda e: None
-
- publisher = DefaultLogPublisher()
- publisher.addObserver(o1)
- publisher.addObserver(o2, filtered=True)
- publisher.addObserver(o3, filtered=False)
- publisher.removeObserver(o2)
- publisher.removeObserver(o3)
-
- self.assertEquals(
- set((o1, publisher.legacyLogObserver)),
- set(publisher.filteredPublisher.observers),
- "Filtered observers do not match expected set"
- )
- self.assertEquals(
- set((publisher.filters,)),
- set(publisher.rootPublisher.observers),
- "Root observers do not match expected set"
- )
-
-
- def test_filteredObserver(self):
- namespace = __name__
-
- event_debug = dict(log_namespace=namespace,
- log_level=LogLevel.debug, log_format="")
- event_error = dict(log_namespace=namespace,
- log_level=LogLevel.error, log_format="")
- events = []
-
- observer = lambda e: events.append(e)
-
- publisher = DefaultLogPublisher()
-
- publisher.addObserver(observer, filtered=True)
- publisher(event_debug)
- publisher(event_error)
- self.assertNotIn(event_debug, events)
- self.assertIn(event_error, events)
-
-
- def test_filteredObserverNoFilteringKeys(self):
- event_debug = dict(log_level=LogLevel.debug)
- event_error = dict(log_level=LogLevel.error)
- event_none = dict()
- events = []
-
- observer = lambda e: events.append(e)
-
- publisher = DefaultLogPublisher()
- publisher.addObserver(observer, filtered=True)
- publisher(event_debug)
- publisher(event_error)
- publisher(event_none)
- self.assertNotIn(event_debug, events)
- self.assertNotIn(event_error, events)
- self.assertNotIn(event_none, events)
-
-
- def test_unfilteredObserver(self):
- namespace = __name__
-
- event_debug = dict(log_namespace=namespace, log_level=LogLevel.debug,
- log_format="")
- event_error = dict(log_namespace=namespace, log_level=LogLevel.error,
- log_format="")
- events = []
-
- observer = lambda e: events.append(e)
-
- publisher = DefaultLogPublisher()
-
- publisher.addObserver(observer, filtered=False)
- publisher(event_debug)
- publisher(event_error)
- self.assertIn(event_debug, events)
- self.assertIn(event_error, events)
-
-
-
-class FilteringLogObserverTests(SetUpTearDown, unittest.TestCase):
- """
- Tests for L{FilteringLogObserver}.
- """
-
- def test_interface(self):
- """
- L{FilteringLogObserver} is an L{ILogObserver}.
- """
- observer = FilteringLogObserver(lambda e: None, ())
- try:
- verifyObject(ILogObserver, observer)
- except BrokenMethodImplementation as e:
- self.fail(e)
-
-
- def filterWith(self, *filters):
- events = [
- dict(count=0),
- dict(count=1),
- dict(count=2),
- dict(count=3),
- ]
-
- class Filters(object):
- @staticmethod
- def twoMinus(event):
- if event["count"] <= 2:
- return PredicateResult.yes
- return PredicateResult.maybe
-
- @staticmethod
- def twoPlus(event):
- if event["count"] >= 2:
- return PredicateResult.yes
- return PredicateResult.maybe
-
- @staticmethod
- def notTwo(event):
- if event["count"] == 2:
- return PredicateResult.no
- return PredicateResult.maybe
-
- @staticmethod
- def no(event):
- return PredicateResult.no
-
- @staticmethod
- def bogus(event):
- return None
-
- predicates = (getattr(Filters, f) for f in filters)
- eventsSeen = []
- trackingObserver = lambda e: eventsSeen.append(e)
- filteringObserver = FilteringLogObserver(trackingObserver, predicates)
- for e in events:
- filteringObserver(e)
-
- return [e["count"] for e in eventsSeen]
-
-
- def test_shouldLogEvent_noFilters(self):
- self.assertEquals(self.filterWith(), [0, 1, 2, 3])
-
-
- def test_shouldLogEvent_noFilter(self):
- self.assertEquals(self.filterWith("notTwo"), [0, 1, 3])
-
-
- def test_shouldLogEvent_yesFilter(self):
- self.assertEquals(self.filterWith("twoPlus"), [0, 1, 2, 3])
-
-
- def test_shouldLogEvent_yesNoFilter(self):
- self.assertEquals(self.filterWith("twoPlus", "no"), [2, 3])
-
-
- def test_shouldLogEvent_yesYesNoFilter(self):
- self.assertEquals(self.filterWith("twoPlus", "twoMinus", "no"),
- [0, 1, 2, 3])
-
-
- def test_shouldLogEvent_badPredicateResult(self):
- self.assertRaises(TypeError, self.filterWith, "bogus")
-
-
- def test_call(self):
- e = dict(obj=object())
-
- def callWithPredicateResult(result):
- seen = []
- observer = FilteringLogObserver(lambda e: seen.append(e),
- (lambda e: result,))
- observer(e)
- return seen
-
- self.assertIn(e, callWithPredicateResult(PredicateResult.yes))
- self.assertIn(e, callWithPredicateResult(PredicateResult.maybe))
- self.assertNotIn(e, callWithPredicateResult(PredicateResult.no))
-
-
-
-class LegacyLoggerTests(SetUpTearDown, unittest.TestCase):
- """
- Tests for L{LegacyLogger}.
- """
-
- def test_namespace_default(self):
- """
- Default namespace is module name.
- """
- log = TestLegacyLogger(logger=None)
- self.assertEquals(log.newStyleLogger.namespace, __name__)
-
-
- def test_passThroughAttributes(self):
- """
- C{__getattribute__} on L{LegacyLogger} is passing through to Twisted's
- logging module.
- """
- log = TestLegacyLogger()
-
- # Not passed through
- self.assertIn("API-compatible", log.msg.__doc__)
- self.assertIn("API-compatible", log.err.__doc__)
-
- # Passed through
- self.assertIdentical(log.addObserver, twistedLogging.addObserver)
-
-
- def test_legacy_msg(self):
- """
- Test LegacyLogger's log.msg()
- """
- log = TestLegacyLogger()
-
- message = "Hi, there."
- kwargs = {"foo": "bar", "obj": object()}
-
- log.msg(message, **kwargs)
-
- self.assertIdentical(log.newStyleLogger.emitted["level"],
- LogLevel.info)
- self.assertEquals(log.newStyleLogger.emitted["format"], message)
-
- for key, value in kwargs.items():
- self.assertIdentical(log.newStyleLogger.emitted["kwargs"][key],
- value)
-
- log.msg(foo="")
-
- self.assertIdentical(log.newStyleLogger.emitted["level"],
- LogLevel.info)
- self.assertIdentical(log.newStyleLogger.emitted["format"], None)
-
-
- def test_legacy_err_implicit(self):
- """
- Test LegacyLogger's log.err() capturing the in-flight exception.
- """
- log = TestLegacyLogger()
-
- exception = RuntimeError("Oh me, oh my.")
- kwargs = {"foo": "bar", "obj": object()}
-
- try:
- raise exception
- except RuntimeError:
- log.err(**kwargs)
-
- self.legacy_err(log, kwargs, None, exception)
-
-
- def test_legacy_err_exception(self):
- """
- Test LegacyLogger's log.err() with a given exception.
- """
- log = TestLegacyLogger()
-
- exception = RuntimeError("Oh me, oh my.")
- kwargs = {"foo": "bar", "obj": object()}
- why = "Because I said so."
-
- try:
- raise exception
- except RuntimeError as e:
- log.err(e, why, **kwargs)
-
- self.legacy_err(log, kwargs, why, exception)
-
-
- def test_legacy_err_failure(self):
- """
- Test LegacyLogger's log.err() with a given L{Failure}.
- """
- log = TestLegacyLogger()
-
- exception = RuntimeError("Oh me, oh my.")
- kwargs = {"foo": "bar", "obj": object()}
- why = "Because I said so."
-
- try:
- raise exception
- except RuntimeError:
- log.err(Failure(), why, **kwargs)
-
- self.legacy_err(log, kwargs, why, exception)
-
-
- def test_legacy_err_bogus(self):
- """
- Test LegacyLogger's log.err() with a bogus argument.
- """
- log = TestLegacyLogger()
-
- exception = RuntimeError("Oh me, oh my.")
- kwargs = {"foo": "bar", "obj": object()}
- why = "Because I said so."
- bogus = object()
-
- try:
- raise exception
- except RuntimeError:
- log.err(bogus, why, **kwargs)
-
- errors = self.flushLoggedErrors(exception.__class__)
- self.assertEquals(len(errors), 0)
-
- self.assertIdentical(log.newStyleLogger.emitted["level"],
- LogLevel.error)
- self.assertEquals(log.newStyleLogger.emitted["format"], repr(bogus))
- self.assertIdentical(log.newStyleLogger.emitted["kwargs"]["why"], why)
-
- for key, value in kwargs.items():
- self.assertIdentical(log.newStyleLogger.emitted["kwargs"][key],
- value)
-
-
- def legacy_err(self, log, kwargs, why, exception):
- #
- # log.failure() will cause trial to complain, so here we check that
- # trial saw the correct error and remove it from the list of things to
- # complain about.
- #
- errors = self.flushLoggedErrors(exception.__class__)
- self.assertEquals(len(errors), 1)
-
- self.assertIdentical(log.newStyleLogger.emitted["level"],
- LogLevel.error)
- self.assertEquals(log.newStyleLogger.emitted["format"], None)
- emittedKwargs = log.newStyleLogger.emitted["kwargs"]
- self.assertIdentical(emittedKwargs["failure"].__class__, Failure)
- self.assertIdentical(emittedKwargs["failure"].value, exception)
- self.assertIdentical(emittedKwargs["why"], why)
-
- for key, value in kwargs.items():
- self.assertIdentical(log.newStyleLogger.emitted["kwargs"][key],
- value)
-
-
-
-class Unformattable(object):
- """
- An object that raises an exception from C{__repr__}.
- """
-
- def __repr__(self):
- return str(1 / 0)
+ index += 1
+ logger.info("T\xc3\xa9st {str}", str="t\xc3\xa9st")
+ self.assertEqual(observed[index]["log_level"], LogLevel.info)
+ self.assertEqual(observed[index]["log_format"], u"T\xe9st {str}")
+ self.assertEqual(observed[index]["str"], u"t\xe9st")
+ self.assertEqual(sio.getvalue().splitlines()[index].split("#info] ")[1], "T\xc3\xa9st t\xc3\xa9st")
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20151109/2fd61982/attachment-0001.html>
More information about the calendarserver-changes
mailing list