Testing with Twisted

Note

To ensure you are using compatible versions, install with the testfixtures[twisted] extra.

Due to its longevity, Twisted has its own structured logging framework built around Logger objects and log event dicts. TwistedSource is provided to work with this.

Testing logging

Pass a TwistedSource instance when constructing LogCapture. It replaces all existing Twisted log observers for the duration of the context manager and restores them on exit:

from twisted.logger import Logger
from testfixtures import LogCapture
from testfixtures.twisted import TwistedSource

log = Logger()

with LogCapture(TwistedSource()) as logs:
    log.info('hello {name}', name='world')

logs.check(('INFO', 'hello world'))

See Testing logging for the full LogCapture API, including check(), check_present(), actual(), and the entries attribute.

Inspecting raw events

Each Entry in entries exposes the underlying Twisted log event dict via its raw attribute:

with LogCapture(TwistedSource()) as logs:
    log.info('hello {greeting}', greeting='world')

>>> logs.entries[0].raw
{'greeting': 'world',..., 'log_level': <LogLevel=info>, 'log_namespace': '<unknown>', ...}

The event dict contains all the structured fields Twisted attaches to each log call, including log_namespace, log_source, log_level, log_format along with any keyword arguments passed to the log call. log_failure is present when logging a Failure.

Capturing specific fields

You can control which fields form the actual value by passing an attributes sequence to TwistedSource. Elements may be string keys into the event dict or callables that receive the event dict.

For example, to capture only the formatted message, pass formatEvent() as the sole element:

from twisted.logger import formatEvent

with LogCapture(TwistedSource(attributes=formatEvent)) as logs:
    log.info('hello {greeting}', greeting='world')

logs.check('hello world')

You can also mix string keys and callables:

with LogCapture(TwistedSource(attributes=('log_namespace', formatEvent))) as logs:
    Logger(namespace='myapp').info('started')

logs.check(('myapp', 'started'))

For full control, pass a single callable as attributes:

from testfixtures.twisted import INFO, WARN
logger = Logger(namespace='myapp')

def extract(event):
    return {'level': event['log_level'], 'message': formatEvent(event)}

with LogCapture(TwistedSource(attributes=extract)) as logs:
    logger.info('started')
    logger.warn('low memory')

logs.check(
    {'level': INFO, 'message': 'started'},
    {'level': WARN, 'message': 'low memory'},
)

In the above example, the Twisted log level objects are returned. For convenience in making assertions about these, the following constants are exported from testfixtures.twisted for use in assertions: DEBUG, INFO, WARN, ERROR and CRITICAL.

Filtering by level

To capture only events at or above a minimum level:

with LogCapture(TwistedSource(level='warn')) as capture:
    log.info('ignored')
    log.warn('captured')
capture.check(('WARN', 'captured'))

Combining with standard library logging

If your application uses both Twisted and the standard library’s logging, pass both a LoggingSource and a TwistedSource to one LogCapture to capture and check both:

import logging
from testfixtures import LogCapture, LoggingSource
from testfixtures.twisted import TwistedSource

with LogCapture(LoggingSource(), TwistedSource()) as logs:
    logging.warning('from stdlib')
    log.warn('from twisted')

logs.check(
    ('WARNING', 'from stdlib'),
    ('WARN', 'from twisted'),
)

Note

Twisted’s LogLevel.warn surfaces as 'WARN' while standard library logging uses 'WARNING'.

Exceptions and failures

When code logs a Failure, the underlying exception object is stored in the Entry’s exception attribute:

from testfixtures import LogCapture, compare
from testfixtures.twisted import TwistedSource

with LogCapture(TwistedSource()) as logs:
    try:
        raise ValueError('boom')
    except Exception:
        log.failure('something went wrong')

logs.check(
    ('CRITICAL', 'something went wrong')
)
compare(logs.entries[-1].exception, expected=ValueError('boom'))

Testing logging configuration

TwistedSource is good for checking that your code is logging the correct messages; just as important is checking that your application has correctly configured its Twisted log observers. If you have a setup_logging function such as this:

import sys
from twisted.logger import globalLogPublisher, textFileLogObserver

def setup_logging(stream=None) -> None:
    globalLogPublisher.addObserver(textFileLogObserver(stream or sys.stderr))

This can be tested with a unit test such as the following:

from twisted.logger import FileLogObserver
from testfixtures import Replacer, compare, like

def test_setup_logging() -> None:
    with Replacer() as replace:
        replace(
            globalLogPublisher._observers, [],
            container=globalLogPublisher, name='_observers',
        )
        setup_logging()
        compare(
            globalLogPublisher._observers,
            expected=[like(FileLogObserver)],
        )

Trial test case support

Since Twisted’s trial test runner has its own TestCase class on which you may well need to call flushLoggedErrors() in the event that you’re testing exception logging, a dedicated testfixtures.twisted.LogCapture is also provided, including a make() class method for creating, installing and adding a cleanup for the capture.

This sample test suite shows it all in action:

class TestLogCapture(TestCase):

    def test_simple(self):
        capture = TwistedLogCapture.make(self)
        log.info('er, {greeting}', greeting='hi')
        capture.check((INFO, 'er, hi'))

    def test_captured(self):
        capture = TwistedLogCapture.make(self)
        log.info('er, {greeting}', greeting='hi')
        assert len(capture.events) == 1
        compare(capture.events[0]['log_namespace'], expected='tests.test_twisted')

    def test_fields(self):
        capture = TwistedLogCapture.make(self, fields=('a', 'b'))
        log.info('{a}, {b}', a=1, b=2)
        log.info('{a}, {b}', a=3, b=4)
        capture.check(
            [1, 2],
            [3, 4],
        )

    def test_field(self):
        capture = TwistedLogCapture.make(self, fields=(formatEvent,))
        log.info('er, {greeting}', greeting='hi')
        capture.check('er, hi')

    def test_check_failure_test_minimal(self):
        capture = TwistedLogCapture.make(self)
        try:
            raise Exception('all gone wrong')
        except:
            log.failure('oh dear')
        capture.check_failure_text('all gone wrong')
        self.flushLoggedErrors()

    def test_check_failure_test_maximal(self):
        capture = TwistedLogCapture.make(self)
        try:
            raise TypeError('all gone wrong')
        except:
            log.failure('oh dear')
        log.info("don't look at me...")
        capture.check_failure_text(str(TypeError), index=0, attribute='type')
        self.flushLoggedErrors()
        self.flushLoggedErrors()

    def test_raise_logged_failure(self):
        capture = TwistedLogCapture.make(self)
        try:
            raise TypeError('all gone wrong')
        except:
            log.failure('oh dear')
        with ShouldRaise(Failure) as s:
            capture.raise_logged_failure()
        compare(s.raised.value, expected=TypeError('all gone wrong'))
        self.flushLoggedErrors()

    def test_raise_later_logged_failure(self):
        capture = TwistedLogCapture.make(self)
        try:
            raise ValueError('boom!')
        except:
            log.failure('oh dear')
        try:
            raise TypeError('all gone wrong')
        except:
            log.failure('what now?!')
        with ShouldRaise(Failure) as s:
            capture.raise_logged_failure(start_index=1)
        compare(s.raised.value, expected=TypeError('all gone wrong'))
        self.flushLoggedErrors()

    def test_order_doesnt_matter_ok(self):
        capture = TwistedLogCapture.make(self)
        log.info('Failed to send BAR')
        log.info('Sent FOO, length 1234')
        log.info('Sent 1 Messages')
        capture.check(
            (INFO, S(r'Sent FOO, length \d+')),
            (INFO, 'Failed to send BAR'),
            (INFO, 'Sent 1 Messages'),
            order_matters=False
        )

    def test_order_doesnt_matter_failure(self):
        capture = TwistedLogCapture.make(self)
        log.info('Failed to send BAR')
        log.info('Sent FOO, length 1234')
        log.info('Sent 1 Messages')
        with ShouldAssert(
            "same:\n"
            "[(<LogLevel=info>, 'Failed to send BAR'), (<LogLevel=info>, 'Sent 1 Messages')]\n"
            "\n"
            "in expected but not actual:\n"
            "[(<LogLevel=info>, <S:Sent FOO, length abc>)]\n"
            "\n"
            "in actual but not expected:\n"
            "[(<LogLevel=info>, 'Sent FOO, length 1234')]"
        ):
            capture.check(
                (INFO, S('Sent FOO, length abc')),
                (INFO, 'Failed to send BAR'),
                (INFO, 'Sent 1 Messages'),
                order_matters=False
            )

    def test_order_doesnt_matter_extra_in_expected(self):
        capture = TwistedLogCapture.make(self)
        log.info('Failed to send BAR')
        log.info('Sent FOO, length 1234')
        with ShouldAssert(
            "same:\n"
            "[(<LogLevel=info>, 'Sent FOO, length 1234'),\n"
            " (<LogLevel=info>, 'Failed to send BAR')]\n"
            "\n"
            "in expected but not actual:\n"
            "[(<LogLevel=info>, 'Sent 1 Messages')]"
        ):
            capture.check(
                (INFO, S('Sent FOO, length 1234')),
                (INFO, 'Failed to send BAR'),
                (INFO, 'Sent 1 Messages'),
                order_matters=False
            )

    def test_order_doesnt_matter_extra_in_actual(self):
        capture = TwistedLogCapture.make(self)
        log.info('Failed to send BAR')
        log.info('Sent 1 Messages')
        with ShouldAssert(
            "same:\n"
            "[(<LogLevel=info>, 'Failed to send BAR'), (<LogLevel=info>, 'Sent 1 Messages')]\n"
            "\n"
            "in expected but not actual:\n"
            "[(<LogLevel=info>, <S:Sent FOO, length abc>)]"
        ):
            capture.check(
                (INFO, S('Sent FOO, length abc')),
                (INFO, 'Failed to send BAR'),
                (INFO, 'Sent 1 Messages'),
                order_matters=False
            )