MAINHACK

Viewing / Editing File: twisted.py

File is not writable. Editing disabled.
"""
Tools for helping to test Twisted applications.
"""
from pprint import pformat
from typing import Union, Sequence, Callable
from unittest import TestCase

from constantly import NamedConstant

from . import compare
from twisted.logger import globalLogPublisher, formatEvent, LogLevel


class LogCapture(object):
    """
    A helper for capturing stuff logged using Twisted's loggers.

    :param fields:
      A sequence of field names that :meth:`~LogCapture.check` will use to build
      "actual" events to compare against the expected events passed in.
      If items are strings, they will be treated as keys info the Twisted logging event.
      If they are callable, they will be called with the event as their only parameter.
      If only one field is specified, "actual" events will just be that one field;
      otherwise they will be a tuple of the specified fields.
    """

    def __init__(self, fields: Sequence[Union[str, Callable]] = ('log_level', formatEvent,)):
        #: The list of events captured.
        self.events = []
        self.fields = fields

    def __call__(self, event):
        self.events.append(event)

    def install(self):
        "Start capturing."
        self.original_observers = globalLogPublisher._observers
        globalLogPublisher._observers = [self]

    def uninstall(self):
        "Stop capturing."
        globalLogPublisher._observers = self.original_observers

    def check(self, *expected, order_matters: bool = True):
        """
        Check captured events against those supplied. Please see the ``fields`` parameter
        to the constructor to see how "actual" events are built.

        :param order_matters:
          This defaults to ``True``. If ``False``, the order of expected logging versus
          actual logging will be ignored.
        """
        actual = []
        for event in self.events:
            actual_event = tuple(field(event) if callable(field) else event.get(field)
                            for field in self.fields)
            if len(actual_event) == 1:
                actual_event = actual_event[0]
            actual.append(actual_event)
        if order_matters:
            compare(expected=expected, actual=actual)
        else:
            expected = list(expected)
            matched = []
            unmatched = []
            for entry in actual:
                try:
                    index = expected.index(entry)
                except ValueError:
                    unmatched.append(entry)
                else:
                    matched.append(expected.pop(index))
            if expected:
                raise AssertionError((
                    'entries not as expected:\n\n'
                    'expected and found:\n%s\n\n'
                    'expected but not found:\n%s\n\n'
                    'other entries:\n%s'
                ) % (pformat(matched), pformat(expected), pformat(unmatched)))

    def check_failure_text(self, expected: str, index: int = -1, attribute: str = 'value'):
        """
        Check the string representation of an attribute of a logged ``Failure`` is as expected.

        :param expected: The expected string representation.
        :param index: The index into :attr:`events` where the failure should have been logged.
        :param attribute: The attribute of the failure of which to find the string representation.
        """
        compare(expected, actual=str(getattr(self.events[index]['log_failure'], attribute)))

    def raise_logged_failure(self, start_index: int = 0):
        """
        A debugging tool that raises the first failure encountered in captured logging.

        :param start_index: The index into :attr:`events` from where to start looking for failures.
        """
        for event in self.events[start_index:]:
            failure = event.get('log_failure')
            if failure:
                raise failure

    @classmethod
    def make(cls, testcase: TestCase, **kw):
        """
        Instantiate, install and add a cleanup for a :class:`LogCapture`.

        :param testcase: This must be an instance of :class:`twisted.trial.unittest.TestCase`.
        :param kw: Any other parameters are passed directly to the :class:`LogCapture` constructor.
        :return: The :class:`LogCapture` instantiated by this method.
        """
        capture = cls(**kw)
        capture.install()
        testcase.addCleanup(capture.uninstall)
        return capture


#: Short reference to Twisted's ``LogLevel.debug``
DEBUG: NamedConstant = LogLevel.debug
#: Short reference to Twisted's ``LogLevel.info``
INFO: NamedConstant = LogLevel.info
#: Short reference to Twisted's ``LogLevel.warn``
WARN: NamedConstant = LogLevel.warn
#: Short reference to Twisted's ``LogLevel.error``
ERROR: NamedConstant = LogLevel.error
#: Short reference to Twisted's ``LogLevel.critical``
CRITICAL: NamedConstant = LogLevel.critical
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: twisted.py

Server Info

System Features

User Info
