Skip to content

Unittest

unittest integration for Pact logs.

PactTestMixin scopes the Pact test run ID to each test and prints the Pact log records collected during a test to stderr when the test fails or errors:

import unittest

from pact.testing.unittest import PactTestMixin


class ConsumerTest(PactTestMixin, unittest.TestCase):
    def test_something(self) -> None: ...

List the mixin before unittest.TestCase so that its run wraps the test.

Classes

PactTestMixin

Mixin for unittest.TestCase subclasses using Pact.

While a test runs, its id() is set as the Pact test run ID and every record on the pact logger is buffered. If the test fails or errors, the buffered records are written to stderr under a banner naming the test. Afterwards, the run ID is cleared or, when pytest runs the test case, set back to the test's node ID.

A logger only hands records at or above its effective level to its handlers, so when the pact logger has no explicit level it is lowered to forwarding_level for the duration of the test and restored afterwards. Handlers above the pact logger still receive only the records at or above its previous effective level. Handlers attached to the pact logger or below it receive every buffered record.

Methods:

run(result: unittest.TestResult | None = None) -> unittest.TestResult | None

Run the test with the Pact run ID set and Pact logs buffered.

PARAMETER DESCRIPTION
result

The result to record into. When omitted, the test's defaultTestResult() is used, and its startTestRun() and stopTestRun() are called around the test, as unittest.TestCase.run does.

TYPE: TestResult | None DEFAULT: None

RETURNS DESCRIPTION
TestResult | None

The result the test was recorded into.

Source code in src/pact/testing/unittest.py
def run(
    self,
    result: unittest.TestResult | None = None,
) -> unittest.TestResult | None:
    """
    Run the test with the Pact run ID set and Pact logs buffered.

    Args:
        result:
            The result to record into. When omitted, the test's
            `defaultTestResult()` is used, and its `startTestRun()` and
            `stopTestRun()` are called around the test, as
            `unittest.TestCase.run` does.

    Returns:
        The result the test was recorded into.
    """
    owns_result = result is None
    if result is None:
        result = self.defaultTestResult()  # type: ignore[attr-defined]
        result.startTestRun()
    test_id: str = self.id()  # type: ignore[attr-defined]
    proxy = _ResultProxy(result)

    pact_logger = logging.getLogger("pact")
    lowered = pact_logger.level == logging.NOTSET
    propagate = pact_logger.propagate
    handler = _BufferHandler(
        pact_logger.parent if lowered and propagate else None,
        pact_logger.getEffectiveLevel(),
    )

    pact_logging.set_test_run_id(test_id)
    pact_logger.addHandler(handler)
    if lowered:
        pact_logger.setLevel(pact_logging.forwarding_level())
        pact_logger.propagate = False
    try:
        super().run(proxy)  # type: ignore[misc]
        if proxy.failed:
            self._dump(test_id, handler.records)
        return result
    finally:
        if lowered:
            pact_logger.setLevel(logging.NOTSET)
            pact_logger.propagate = propagate
        pact_logger.removeHandler(handler)
        pact_logging.set_test_run_id(_enclosing_test_run_id())
        if owns_result:
            result.stopTestRun()