Skip to content

V3

Pact Python V3.

This module provides a preview of the next major version of Pact Python. It is subject to breaking changes and may have bugs; however, it is available for testing and feedback. If you encounter any issues, please report them on GitHub, and if you have any feedback, please let us know on either the GitHub discussions or on Slack.

The next major release will use the Pact Rust library to provide full support for all Pact features, and bring feature parity between the Python library and the other Pact libraries.

Migration Plan

This change will introduce some breaking changes where needed, but it will be done in a staged manner to give everyone the opportunity to migrate.

🚧 Stage 1 (from v2.2)

  • The main Pact Python library remains the same. Bugs and minor features will continue to be added to the existing library, but no new major features will be added as the focus will be on the new library.
  • The new library is exposed within pact.v3 and can be used alongside the existing library. During this stage, no guarantees are made about the stability of the pact.v3 module.
  • Users are not recommended to use the new library in any production critical code at this stage, but are encouraged to try it out and provide feedback.
  • The existing library will raise PendingDeprecationWarning warnings when it is used (if these warnings are enabled).

🛠 Stage 2 (from v2.3, tbc)

  • The library within pact.v3 is considered generally stable and users are encouraged to start migrating to it.
  • A detailed migration guide will be provided.
  • The existing library will raise DeprecationWarning warnings when it is used to help raise awareness of the upcoming change.
  • This stage will likely last a few months to give everyone the opportunity to migrate.

🚀 Stage 3 (from v3)

  • The pact.v3 module is renamed to pact

    • People who have previously migrated to pact.v3 should be able to do a s/pact.v3/pact/ and have everything work.
    • If the previous stage identifies any breaking changes as necessary, they will be made at this point and a detailed migration guide will be provided.
  • The existing library is moved to the pact.v2 scope.

    • ‼ This will be a very major and breaking change. Previous code running against v2 of Pact Python will not work against v3 of Pact Python.
    • Users still wanting to use the v2 library will need to update their code to use the new pact.v2 module. A s/pact/pact.v2/ should be sufficient.
    • The pact.v2 module will be considered deprecated, and will eventually be removed in a future release. No new features and only critical bug fixes will be made to this part of the library.

Classes

Pact(consumer: str, provider: str)

A Pact between a consumer and a provider.

This class defines a Pact between a consumer and a provider. It is the central class in Pact's framework, and is responsible for defining the interactions between the two parties.

One Pact instance should be created for each provider that a consumer interacts with. The methods on this class are used to define the broader attributes of the Pact, such as the consumer and provider names, the Pact specification, any plugins that are used, and any metadata that is attached to the Pact.

Each interaction between the consumer and the provider is defined through the upon_receiving method, which returns a sub-class of Interaction.

PARAMETER DESCRIPTION
consumer

Name of the consumer.

TYPE: str

provider

Name of the provider.

TYPE: str

Source code in src/pact/v3/pact.py
def __init__(
    self,
    consumer: str,
    provider: str,
) -> None:
    """
    Initialise a new Pact.

    Args:
        consumer:
            Name of the consumer.

        provider:
            Name of the provider.
    """
    if not consumer:
        msg = "Consumer name cannot be empty."
        raise ValueError(msg)
    if not provider:
        msg = "Provider name cannot be empty."
        raise ValueError(msg)

    self._consumer = consumer
    self._provider = provider
    self._interactions: set[Interaction] = set()
    self._handle: pact.v3.ffi.PactHandle = pact.v3.ffi.new_pact(
        consumer,
        provider,
    )

Attributes

consumer: str property

Consumer name.

provider: str property

Provider name.

specification: pact.v3.ffi.PactSpecification property

Pact specification version.

Functions

interactions(kind: Literal['HTTP', 'Sync', 'Async'] = 'HTTP') -> Generator[pact.v3.ffi.SynchronousHttp, None, None] | Generator[pact.v3.ffi.SynchronousMessage, None, None] | Generator[pact.v3.ffi.AsynchronousMessage, None, None]
interactions(
    kind: Literal["HTTP"],
) -> Generator[pact.v3.ffi.SynchronousHttp, None, None]
interactions(
    kind: Literal["Sync"],
) -> Generator[pact.v3.ffi.SynchronousMessage, None, None]
interactions(
    kind: Literal["Async"],
) -> Generator[pact.v3.ffi.AsynchronousMessage, None, None]

Return an iterator over the Pact's interactions.

The kind is used to specify the type of interactions that will be iterated over.

Source code in src/pact/v3/pact.py
def interactions(
    self,
    kind: Literal["HTTP", "Sync", "Async"] = "HTTP",
) -> (
    Generator[pact.v3.ffi.SynchronousHttp, None, None]
    | Generator[pact.v3.ffi.SynchronousMessage, None, None]
    | Generator[pact.v3.ffi.AsynchronousMessage, None, None]
):
    """
    Return an iterator over the Pact's interactions.

    The kind is used to specify the type of interactions that will be
    iterated over.
    """
    # TODO: Add an iterator for `All` interactions.
    # https://github.com/pact-foundation/pact-python/issues/451
    if kind == "HTTP":
        yield from pact.v3.ffi.pact_handle_get_sync_http_iter(self._handle)
    elif kind == "Sync":
        yield from pact.v3.ffi.pact_handle_get_sync_message_iter(self._handle)
    elif kind == "Async":
        yield from pact.v3.ffi.pact_handle_get_async_message_iter(self._handle)
    else:
        msg = f"Unknown interaction type: {kind}"
        raise ValueError(msg)
    return  # Ensures that the parent object outlives the generator
serve(addr: str = 'localhost', port: int = 0, transport: str = 'http', transport_config: str | None = None, *, raises: bool = True, verbose: bool = True) -> PactServer

Return a mock server for the Pact.

This function configures a mock server for the Pact. The mock server is then started when the Pact is entered into a with block:

pact = Pact("consumer", "provider")
with pact.serve() as srv:
    ...
PARAMETER DESCRIPTION
addr

Address to bind the mock server to. Defaults to localhost.

TYPE: str DEFAULT: 'localhost'

port

Port to bind the mock server to. Defaults to 0, which will select a random port.

TYPE: int DEFAULT: 0

transport

Transport to use for the mock server. Defaults to HTTP.

TYPE: str DEFAULT: 'http'

transport_config

Configuration for the transport. This is specific to the transport being used and should be a JSON string.

TYPE: str | None DEFAULT: None

raises

Whether to raise an exception if there are mismatches between the Pact and the server. If set to False, then the mismatches must be handled manually.

TYPE: bool DEFAULT: True

verbose

Whether or not to print the mismatches to the logger. This works independently of raises.

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
PactServer

A PactServer instance.

Source code in src/pact/v3/pact.py
def serve(  # noqa: PLR0913
    self,
    addr: str = "localhost",
    port: int = 0,
    transport: str = "http",
    transport_config: str | None = None,
    *,
    raises: bool = True,
    verbose: bool = True,
) -> PactServer:
    """
    Return a mock server for the Pact.

    This function configures a mock server for the Pact. The mock server
    is then started when the Pact is entered into a `with` block:

    ```python
    pact = Pact("consumer", "provider")
    with pact.serve() as srv:
        ...
    ```

    Args:
        addr:
            Address to bind the mock server to. Defaults to `localhost`.

        port:
            Port to bind the mock server to. Defaults to `0`, which will
            select a random port.

        transport:
            Transport to use for the mock server. Defaults to `HTTP`.

        transport_config:
            Configuration for the transport. This is specific to the
            transport being used and should be a JSON string.

        raises:
            Whether to raise an exception if there are mismatches between
            the Pact and the server. If set to `False`, then the mismatches
            must be handled manually.

        verbose:
            Whether or not to print the mismatches to the logger. This works
            independently of `raises`.

    Returns:
        A [`PactServer`][pact.v3.pact.PactServer] instance.
    """
    return PactServer(
        self._handle,
        addr,
        port,
        transport,
        transport_config,
        raises=raises,
        verbose=verbose,
    )
upon_receiving(description: str, interaction: Literal['HTTP', 'Sync', 'Async'] = 'HTTP') -> HttpInteraction | AsyncMessageInteraction | SyncMessageInteraction
upon_receiving(
    description: str, interaction: Literal["HTTP"] = ...
) -> HttpInteraction
upon_receiving(
    description: str, interaction: Literal["Async"]
) -> AsyncMessageInteraction
upon_receiving(
    description: str, interaction: Literal["Sync"]
) -> SyncMessageInteraction

Create a new Interaction.

PARAMETER DESCRIPTION
description

Description of the interaction. This must be unique within the Pact.

TYPE: str

interaction

Type of interaction. Defaults to HTTP. This must be one of HTTP, Async, or Sync.

TYPE: Literal['HTTP', 'Sync', 'Async'] DEFAULT: 'HTTP'

Source code in src/pact/v3/pact.py
def upon_receiving(
    self,
    description: str,
    interaction: Literal["HTTP", "Sync", "Async"] = "HTTP",
) -> HttpInteraction | AsyncMessageInteraction | SyncMessageInteraction:
    """
    Create a new Interaction.

    Args:
        description:
            Description of the interaction. This must be unique
            within the Pact.

        interaction:
            Type of interaction. Defaults to `HTTP`. This must be one of
            `HTTP`, `Async`, or `Sync`.
    """
    if interaction == "HTTP":
        return HttpInteraction(self._handle, description)
    if interaction == "Async":
        return AsyncMessageInteraction(self._handle, description)
    if interaction == "Sync":
        return SyncMessageInteraction(self._handle, description)

    msg = f"Invalid interaction type: {interaction}"
    raise ValueError(msg)
using_plugin(name: str, version: str | None = None) -> Self

Add a plugin to be used by the test.

Plugins extend the functionality of Pact.

PARAMETER DESCRIPTION
name

Name of the plugin.

TYPE: str

version

Version of the plugin. This is optional and can be None.

TYPE: str | None DEFAULT: None

Source code in src/pact/v3/pact.py
def using_plugin(self, name: str, version: str | None = None) -> Self:
    """
    Add a plugin to be used by the test.

    Plugins extend the functionality of Pact.

    Args:
        name:
            Name of the plugin.

        version:
            Version of the plugin. This is optional and can be `None`.
    """
    pact.v3.ffi.using_plugin(self._handle, name, version)
    return self
verify(handler: Callable[[str | bytes | None, dict[str, str]], None], kind: Literal['Async', 'Sync'], *, raises: bool = True) -> list[InteractionVerificationError] | None
verify(
    handler: Callable[
        [str | bytes | None, dict[str, str]], None
    ],
    kind: Literal["Async", "Sync"],
    *,
    raises: Literal[True] = True,
) -> None
verify(
    handler: Callable[
        [str | bytes | None, dict[str, str]], None
    ],
    kind: Literal["Async", "Sync"],
    *,
    raises: Literal[False],
) -> list[InteractionVerificationError]

Verify message interactions.

This function is used to ensure that the consumer is able to handle the messages that are defined in the Pact. The handler function is called for each message in the Pact.

The end-user is responsible for defining the handler function and verifying that the messages are handled correctly. For example, if the handler is meant to call an API, then the API call should be mocked out and once the verification is complete, the mock should be verified. Any exceptions raised by the handler will be caught and reported as mismatches.

PARAMETER DESCRIPTION
handler

The function that will be called for each message in the Pact.

The first argument to the function is the message body, either as a string or byte array.

The second argument is the metadata for the message. If there is no metadata, then this will be an empty dictionary.

TYPE: Callable[[str | bytes | None, dict[str, str]], None]

kind

The type of message interaction. This must be one of Async or Sync.

TYPE: Literal['Async', 'Sync']

raises

Whether or not to raise an exception if the handler fails to process a message. If set to False, then the function will return a list of errors.

TYPE: bool DEFAULT: True

Source code in src/pact/v3/pact.py
def verify(
    self,
    handler: Callable[[str | bytes | None, dict[str, str]], None],
    kind: Literal["Async", "Sync"],
    *,
    raises: bool = True,
) -> list[InteractionVerificationError] | None:
    """
    Verify message interactions.

    This function is used to ensure that the consumer is able to handle the
    messages that are defined in the Pact. The `handler` function is called
    for each message in the Pact.

    The end-user is responsible for defining the `handler` function and
    verifying that the messages are handled correctly. For example, if the
    handler is meant to call an API, then the API call should be mocked out
    and once the verification is complete, the mock should be verified. Any
    exceptions raised by the handler will be caught and reported as
    mismatches.

    Args:
        handler:
            The function that will be called for each message in the Pact.

            The first argument to the function is the message body, either as
            a string or byte array.

            The second argument is the metadata for the message. If there
            is no metadata, then this will be an empty dictionary.

        kind:
            The type of message interaction. This must be one of `Async`
            or `Sync`.

        raises:
            Whether or not to raise an exception if the handler fails to
            process a message. If set to `False`, then the function will
            return a list of errors.
    """
    errors: list[InteractionVerificationError] = []
    for message in self.interactions(kind):
        request: pact.v3.ffi.MessageContents | None = None
        if isinstance(message, pact.v3.ffi.SynchronousMessage):
            request = message.request_contents
        elif isinstance(message, pact.v3.ffi.AsynchronousMessage):
            request = message.contents
        else:
            msg = f"Unknown message type: {type(message).__name__}"
            raise TypeError(msg)

        if request is None:
            warnings.warn(
                f"Message '{message.description}' has no contents",
                stacklevel=2,
            )
            continue

        body = request.contents
        metadata = {pair.key: pair.value for pair in request.metadata}

        try:
            handler(body, metadata)
        except Exception as e:  # noqa: BLE001
            errors.append(InteractionVerificationError(message.description, e))

    if raises:
        if errors:
            raise PactVerificationError(errors)
        return None
    return errors
with_metadata(namespace: str, metadata: dict[str, str]) -> Self

Set additional metadata for the Pact.

A common use for this function is to add information about the client library (name, version, hash, etc.) to the Pact.

PARAMETER DESCRIPTION
namespace

Namespace for the metadata. This is used to group the metadata together.

TYPE: str

metadata

Key-value pairs of metadata to add to the Pact.

TYPE: dict[str, str]

Source code in src/pact/v3/pact.py
def with_metadata(
    self,
    namespace: str,
    metadata: dict[str, str],
) -> Self:
    """
    Set additional metadata for the Pact.

    A common use for this function is to add information about the client
    library (name, version, hash, etc.) to the Pact.

    Args:
        namespace:
            Namespace for the metadata. This is used to group the metadata
            together.

        metadata:
            Key-value pairs of metadata to add to the Pact.
    """
    for k, v in metadata.items():
        pact.v3.ffi.with_pact_metadata(self._handle, namespace, k, v)
    return self
with_specification(version: str | pact.v3.ffi.PactSpecification) -> Self

Set the Pact specification version.

The Pact specification version indicates the features which are supported by the Pact, and certain default behaviours.

PARAMETER DESCRIPTION
version

Pact specification version. The can be either a string or a PactSpecification instance.

The version string is case insensitive and has an optional v prefix.

TYPE: str | PactSpecification

Source code in src/pact/v3/pact.py
def with_specification(
    self,
    version: str | pact.v3.ffi.PactSpecification,
) -> Self:
    """
    Set the Pact specification version.

    The Pact specification version indicates the features which are
    supported by the Pact, and certain default behaviours.

    Args:
        version:
            Pact specification version. The can be either a string or a
            [`PactSpecification`][pact.v3.ffi.PactSpecification] instance.

            The version string is case insensitive and has an optional `v`
            prefix.
    """
    if isinstance(version, str):
        version = pact.v3.ffi.PactSpecification.from_str(version)
    pact.v3.ffi.with_specification(self._handle, version)
    return self
write_file(directory: Path | str | None = None, *, overwrite: bool = False) -> None

Write out the pact to a file.

This function should be called once all of the consumer tests have been run. It writes the Pact to a file, which can then be used to validate the provider.

PARAMETER DESCRIPTION
directory

The directory to write the pact to. If the directory does not exist, it will be created. The filename will be automatically generated from the underlying Pact.

TYPE: Path | str | None DEFAULT: None

overwrite

If set to True, the file will be overwritten if it already exists. Otherwise, the contents of the file will be merged with the existing file.

TYPE: bool DEFAULT: False

Source code in src/pact/v3/pact.py
def write_file(
    self,
    directory: Path | str | None = None,
    *,
    overwrite: bool = False,
) -> None:
    """
    Write out the pact to a file.

    This function should be called once all of the consumer tests have been
    run. It writes the Pact to a file, which can then be used to validate
    the provider.

    Args:
        directory:
            The directory to write the pact to. If the directory does not
            exist, it will be created. The filename will be
            automatically generated from the underlying Pact.

        overwrite:
            If set to True, the file will be overwritten if it already
            exists. Otherwise, the contents of the file will be merged with
            the existing file.
    """
    if directory is None:
        directory = Path.cwd()
    pact.v3.ffi.pact_handle_write_file(
        self._handle,
        directory,
        overwrite=overwrite,
    )

Verifier(name: str, host: str | None = None)

A Verifier between a consumer and a provider.

This class encapsulates the logic for verifying that a provider meets the expectations of a consumer. This is done by replaying interactions from the consumer against the provider, and ensuring that the provider's responses match the expectations set by the consumer.

PARAMETER DESCRIPTION
name

The name of the provider to verify. This is used to identify which interactions the provider is involved in, and then Pact will replay these interactions against the provider.

TYPE: str

host

The host on which the Pact verifier is running. This is used to communicate with the provider. If not specified, the default value is localhost.

TYPE: str | None DEFAULT: None

Source code in src/pact/v3/verifier.py
def __init__(self, name: str, host: str | None = None) -> None:
    """
    Create a new Verifier.

    Args:
        name:
            The name of the provider to verify. This is used to identify
            which interactions the provider is involved in, and then Pact
            will replay these interactions against the provider.

        host:
            The host on which the Pact verifier is running. This is used to
            communicate with the provider. If not specified, the default
            value is `localhost`.
    """
    self._name = name
    self._host = host or "localhost"
    self._handle = pact.v3.ffi.verifier_new_for_application()

    # In order to provide a fluent interface, we remember some options which
    # are set using the same FFI method. In particular, we remember
    # transport methods defined, and then before verification call the
    # `set_info` and `add_transport` FFI methods as needed.
    self._transports: list[_ProviderTransport] = []
    self._message_producer: MessageProducer | nullcontext[None] = nullcontext()
    self._state_handler: StateCallback | nullcontext[None] = nullcontext()
    self._disable_ssl_verification = False
    self._request_timeout = 5000

Attributes

logs: str property

Get the logs.

results: dict[str, Any] property

Get the results.

Functions

add_custom_header(name: str, value: str) -> Self

Add a customer header to the request.

These headers are added to every request made to the provider.

PARAMETER DESCRIPTION
name

The key of the header.

TYPE: str

value

The value of the header.

TYPE: str

Source code in src/pact/v3/verifier.py
def add_custom_header(self, name: str, value: str) -> Self:
    """
    Add a customer header to the request.

    These headers are added to every request made to the provider.

    Args:
        name:
            The key of the header.

        value:
            The value of the header.
    """
    pact.v3.ffi.verifier_add_custom_header(self._handle, name, value)
    return self
add_custom_headers(headers: dict[str, str] | Iterable[tuple[str, str]]) -> Self

Add multiple customer headers to the request.

These headers are added to every request made to the provider.

PARAMETER DESCRIPTION
headers

The headers to add. This can be a dictionary or an iterable of key-value pairs. The iterable is preferred as it ensures that repeated headers are not lost.

TYPE: dict[str, str] | Iterable[tuple[str, str]]

Source code in src/pact/v3/verifier.py
def add_custom_headers(
    self,
    headers: dict[str, str] | Iterable[tuple[str, str]],
) -> Self:
    """
    Add multiple customer headers to the request.

    These headers are added to every request made to the provider.

    Args:
        headers:
            The headers to add. This can be a dictionary or an iterable of
            key-value pairs. The iterable is preferred as it ensures that
            repeated headers are not lost.
    """
    if isinstance(headers, dict):
        headers = headers.items()
    for name, value in headers:
        self.add_custom_header(name, value)
    return self
add_source(source: str | Path | URL, *, username: str | None = None, password: str | None = None, token: str | None = None) -> Self
add_source(
    source: str | URL,
    *,
    username: str | None = None,
    password: str | None = None,
) -> Self
add_source(
    source: str | URL, *, token: str | None = None
) -> Self
add_source(source: str | Path) -> Self

Adds a source to the verifier.

This will use one or more Pact files as the source of interactions to verify.

PARAMETER DESCRIPTION
source

The source of the interactions. This may be either of the following:

  • A local file path to a Pact file.
  • A local file path to a directory containing Pact files.
  • A URL to a Pact file.

If using a URL, the username and password parameters can be used to provide basic HTTP authentication, or the token parameter can be used to provide bearer token authentication. The username and password parameters can also be passed as part of the URL.

TYPE: str | Path | URL

username

The username to use for basic HTTP authentication. This is only used when the source is a URL.

TYPE: str | None DEFAULT: None

password

The password to use for basic HTTP authentication. This is only used when the source is a URL.

TYPE: str | None DEFAULT: None

token

The token to use for bearer token authentication. This is only used when the source is a URL. Note that this is mutually exclusive with username and password.

TYPE: str | None DEFAULT: None

Source code in src/pact/v3/verifier.py
def add_source(
    self,
    source: str | Path | URL,
    *,
    username: str | None = None,
    password: str | None = None,
    token: str | None = None,
) -> Self:
    """
    Adds a source to the verifier.

    This will use one or more Pact files as the source of interactions to
    verify.

    Args:
        source:
            The source of the interactions. This may be either of the
            following:

            - A local file path to a Pact file.
            - A local file path to a directory containing Pact files.
            - A URL to a Pact file.

            If using a URL, the `username` and `password` parameters can be
            used to provide basic HTTP authentication, or the `token`
            parameter can be used to provide bearer token authentication.
            The `username` and `password` parameters can also be passed as
            part of the URL.

        username:
            The username to use for basic HTTP authentication. This is only
            used when the source is a URL.

        password:
            The password to use for basic HTTP authentication. This is only
            used when the source is a URL.

        token:
            The token to use for bearer token authentication. This is only
            used when the source is a URL. Note that this is mutually
            exclusive with `username` and `password`.
    """
    if isinstance(source, Path):
        return self._add_source_local(source)

    if isinstance(source, URL):
        if source.scheme == "file":
            return self._add_source_local(source.path)

        if source.scheme in ("http", "https"):
            return self._add_source_remote(
                source,
                username=username,
                password=password,
                token=token,
            )

        msg = f"Invalid source scheme: {source.scheme}"
        raise ValueError(msg)

    # Strings are ambiguous, so we need identify them as either local or
    # remote.
    if "://" in source:
        return self._add_source_remote(
            URL(source),
            username=username,
            password=password,
            token=token,
        )
    return self._add_source_local(source)
add_transport(*, url: str | URL | None = None, protocol: str | None = None, port: int | None = None, path: str | None = None, scheme: str | None = None) -> Self

Add a provider transport method.

If the provider supports multiple transport methods, or non-HTTP(S) methods, this method allows these additional transport methods to be added. It can be called multiple times to add multiple transport methods.

As some transport methods may not use ports, paths or schemes, these parameters are optional. Note that while optional, these may still be used during testing as Pact uses HTTP(S) to communicate with the provider. For example, if you are implementing your own message verification, it needs to be exposed over HTTP and the port and path arguments are used for this testing communication.

PARAMETER DESCRIPTION
url

A convenient way to set the provider transport. This option is mutually exclusive with the other options.

TYPE: str | URL | None DEFAULT: None

protocol

The protocol to use. This will typically be one of:

  • http for communications over HTTP(S)

  • message for non-plugin message-based communications

Any other protocol will be treated as a custom protocol and will be handled by a plugin.

If url is not specified, this parameter is required.

TYPE: str | None DEFAULT: None

port

The provider port.

If the protocol does not use ports, this parameter should be None. If not specified, the default port for the scheme will be used (provided the scheme is known).

TYPE: int | None DEFAULT: None

path

The provider context path.

For protocols which do not use paths, this parameter should be None.

For protocols which do use paths, this parameter should be specified to avoid any ambiguity, though if left unspecified, the root path will be used.

If a non-root path is used, the path given here will be prepended to the path in the interaction. For example, if the path is /api, and the interaction path is /users, the request will be made to /api/users.

TYPE: str | None DEFAULT: None

scheme

The provider scheme, if applicable to the protocol.

This is typically only used for the http protocol, where this value can either be http (the default) or https.

TYPE: str | None DEFAULT: None

Source code in src/pact/v3/verifier.py
def add_transport(
    self,
    *,
    url: str | URL | None = None,
    protocol: str | None = None,
    port: int | None = None,
    path: str | None = None,
    scheme: str | None = None,
) -> Self:
    """
    Add a provider transport method.

    If the provider supports multiple transport methods, or non-HTTP(S)
    methods, this method allows these additional transport methods to be
    added. It can be called multiple times to add multiple transport
    methods.

    As some transport methods may not use ports, paths or schemes, these
    parameters are optional. Note that while optional, these _may_ still be
    used during testing as Pact uses HTTP(S) to communicate with the
    provider. For example, if you are implementing your own message
    verification, it needs to be exposed over HTTP and the `port` and `path`
    arguments are used for this testing communication.

    Args:
        url:
            A convenient way to set the provider transport. This option
            is mutually exclusive with the other options.

        protocol:
            The protocol to use. This will typically be one of:

            -   `http` for communications over HTTP(S)

            -   `message` for non-plugin message-based communications

            Any other protocol will be treated as a custom protocol and will
            be handled by a plugin.

            If `url` is _not_ specified, this parameter is required.

        port:
            The provider port.

            If the protocol does not use ports, this parameter should be
            `None`. If not specified, the default port for the scheme will
            be used (provided the scheme is known).

        path:
            The provider context path.

            For protocols which do not use paths, this parameter should be
            `None`.

            For protocols which do use paths, this parameter should be
            specified to avoid any ambiguity, though if left unspecified,
            the root path will be used.

            If a non-root path is used, the path given here will be
            prepended to the path in the interaction. For example, if the
            path is `/api`, and the interaction path is `/users`, the
            request will be made to `/api/users`.

        scheme:
            The provider scheme, if applicable to the protocol.

            This is typically only used for the `http` protocol, where this
            value can either be `http` (the default) or `https`.
    """
    if url and any(x is not None for x in (protocol, port, path, scheme)):
        msg = "The `url` parameter is mutually exclusive with other parameters"
        raise ValueError(msg)

    if url:
        url = URL(url)
        if url.host != self._host:
            msg = f"Host mismatch: {url.host} != {self._host}"
            raise ValueError(msg)
        protocol = url.scheme
        if protocol == "https":
            protocol = "http"
        port = url.port
        path = url.path
        scheme = url.scheme
        return self.add_transport(
            protocol=protocol,
            port=port,
            path=path,
            scheme=scheme,
        )

    if not protocol:
        msg = "A protocol must be specified"
        raise ValueError(msg)

    if port is None and scheme:
        if scheme.lower() == "http":
            port = 80
        elif scheme.lower() == "https":
            port = 443

    logger.debug(
        "Adding transport to verifier",
        extra={
            "protocol": protocol,
            "port": port,
            "path": path,
            "scheme": scheme,
        },
    )
    self._transports.append(
        _ProviderTransport(
            transport=protocol,
            port=port,
            path=path,
            scheme=scheme,
        )
    )

    return self
broker_source(url: str | URL, *, username: str | None = None, password: str | None = None, token: str | None = None, selector: bool = False) -> BrokerSelectorBuilder | Self
broker_source(
    url: str | URL,
    *,
    username: str | None = None,
    password: str | None = None,
    selector: Literal[False] = False,
) -> Self
broker_source(
    url: str | URL,
    *,
    token: str | None = None,
    selector: Literal[False] = False,
) -> Self
broker_source(
    url: str | URL,
    *,
    username: str | None = None,
    password: str | None = None,
    selector: Literal[True],
) -> BrokerSelectorBuilder
broker_source(
    url: str | URL,
    *,
    token: str | None = None,
    selector: Literal[True],
) -> BrokerSelectorBuilder

Adds a broker source to the verifier.

PARAMETER DESCRIPTION
url

The broker URL. TThe URL may contain a username and password for basic HTTP authentication.

TYPE: str | URL

username

The username to use for basic HTTP authentication. If the source is a URL containing a username, this parameter must be None.

TYPE: str | None DEFAULT: None

password

The password to use for basic HTTP authentication. If the source is a URL containing a password, this parameter must be None.

TYPE: str | None DEFAULT: None

token

The token to use for bearer token authentication. This is mutually exclusive with username and password (whether they be specified through arguments, or embedded in the URL).

TYPE: str | None DEFAULT: None

selector

Whether to return a BrokerSelectorBuilder instance.

TYPE: bool DEFAULT: False

Source code in src/pact/v3/verifier.py
def broker_source(
    self,
    url: str | URL,
    *,
    username: str | None = None,
    password: str | None = None,
    token: str | None = None,
    selector: bool = False,
) -> BrokerSelectorBuilder | Self:
    """
    Adds a broker source to the verifier.

    Args:
        url:
            The broker URL. TThe URL may contain a username and password for
            basic HTTP authentication.

        username:
            The username to use for basic HTTP authentication. If the source
            is a URL containing a username, this parameter must be `None`.

        password:
            The password to use for basic HTTP authentication. If the source
            is a URL containing a password, this parameter must be `None`.

        token:
            The token to use for bearer token authentication. This is
            mutually exclusive with `username` and `password` (whether they
            be specified through arguments, or embedded in the URL).

        selector:
            Whether to return a BrokerSelectorBuilder instance.
    """
    url = URL(url)

    if url.user and username:
        msg = "Cannot specify both `username` and a username in the URL"
        raise ValueError(msg)
    username = url.user or username

    if url.password and password:
        msg = "Cannot specify both `password` and a password in the URL"
        raise ValueError(msg)
    password = url.password or password

    if token and (username or password):
        msg = "Cannot specify both `token` and `username`/`password`"
        raise ValueError(msg)

    if selector:
        return BrokerSelectorBuilder(
            self,
            str(url.with_user(None).with_password(None)),
            username,
            password,
            token,
        )
    pact.v3.ffi.verifier_broker_source(
        self._handle,
        str(url.with_user(None).with_password(None)),
        username,
        password,
        token,
    )
    return self
disable_ssl_verification() -> Self

Disable SSL verification.

Source code in src/pact/v3/verifier.py
def disable_ssl_verification(self) -> Self:
    """
    Disable SSL verification.
    """
    self._disable_ssl_verification = True
    pact.v3.ffi.verifier_set_verification_options(
        self._handle,
        disable_ssl_verification=self._disable_ssl_verification,
        request_timeout=self._request_timeout,
    )
    return self
filter(description: str | None = None, *, state: str | None = None, no_state: bool = False) -> Self

Set the filter for the interactions.

This method can be used to filter interactions based on their description and state. Repeated calls to this method will replace the previous filter.

PARAMETER DESCRIPTION
description

The interaction description. This should be a regular expression. If unspecified, no filtering will be done based on the description.

TYPE: str | None DEFAULT: None

state

The interaction state. This should be a regular expression. If unspecified, no filtering will be done based on the state.

TYPE: str | None DEFAULT: None

no_state

Whether to include interactions with no state.

TYPE: bool DEFAULT: False

Source code in src/pact/v3/verifier.py
def filter(
    self,
    description: str | None = None,
    *,
    state: str | None = None,
    no_state: bool = False,
) -> Self:
    """
    Set the filter for the interactions.

    This method can be used to filter interactions based on their
    description and state. Repeated calls to this method will replace the
    previous filter.

    Args:
        description:
            The interaction description. This should be a regular
            expression. If unspecified, no filtering will be done based on
            the description.

        state:
            The interaction state. This should be a regular expression. If
            unspecified, no filtering will be done based on the state.

        no_state:
            Whether to include interactions with no state.
    """
    logger.debug(
        "Setting filter for verifier",
        extra={
            "description": description,
            "state": state,
            "no_state": no_state,
        },
    )
    pact.v3.ffi.verifier_set_filter_info(
        self._handle,
        description,
        state,
        filter_no_state=no_state,
    )
    return self
filter_consumers(*filters: str) -> Self

Filter the consumers.

PARAMETER DESCRIPTION
filters

Filters to apply to the consumers.

TYPE: str DEFAULT: ()

Source code in src/pact/v3/verifier.py
def filter_consumers(self, *filters: str) -> Self:
    """
    Filter the consumers.

    Args:
        filters:
            Filters to apply to the consumers.
    """
    pact.v3.ffi.verifier_set_consumer_filters(self._handle, filters)
    return self
logs_for_provider(provider: str) -> str classmethod

Get the logs for a provider.

Source code in src/pact/v3/verifier.py
@classmethod
def logs_for_provider(cls, provider: str) -> str:
    """
    Get the logs for a provider.
    """
    return pact.v3.ffi.verifier_logs_for_provider(provider)
message_handler(handler: MessageProducerFull | dict[str, MessageProducerNoName | Message]) -> Self
message_handler(handler: MessageProducerFull) -> Self
message_handler(
    handler: dict[str, MessageProducerNoName | Message],
) -> Self

Set the message handler.

This method sets a custom message handler for the verifier. The handler can be called to produce a specific message to send to the provider.

This can be provided in one of two ways:

  1. A fully fledged function that will be called for all messages. The function must take two arguments: the name of the message (as a string), and optional parameters (as a dictionary). This then returns the message as bytes.

    This is the most powerful option as it allows for full control over the message generation.

  2. A dictionary mapping message names to producer functions, or bytes. In this case, the producer function must take optional parameters (as a dictionary) and return the message as bytes.

    If the message to be produced is static, the bytes can be provided directly.

Implementation

There are a large number of ways to send messages, and the specifics of the transport methods are not specifically relevant to Pact. As such, Pact abstracts the transport layer away and uses a lightweight HTTP server to handle messages.

Pact Python is capable of setting up this server and handling the messages internally using user-provided handlers. It is possible to use your own HTTP server to handle messages by using the add_transport method. It is not possible to use both this method and add_transport to handle messages.

PARAMETER DESCRIPTION
handler

The message handler. This should be a callable that takes no arguments: the

TYPE: MessageProducerFull | dict[str, MessageProducerNoName | Message]

Source code in src/pact/v3/verifier.py
def message_handler(
    self,
    handler: MessageProducerFull | dict[str, MessageProducerNoName | Message],
) -> Self:
    """
    Set the message handler.

    This method sets a custom message handler for the verifier. The handler
    can be called to produce a specific message to send to the provider.

    This can be provided in one of two ways:

    1.  A fully fledged function that will be called for all messages. The
        function must take two arguments: the name of the message (as a
        string), and optional parameters (as a dictionary). This then
        returns the message as bytes.

        This is the most powerful option as it allows for full control over
        the message generation.

    2.  A dictionary mapping message names to producer functions, or bytes.
        In this case, the producer function must take optional parameters
        (as a dictionary) and return the message as bytes.

        If the message to be produced is static, the bytes can be provided
        directly.

    ## Implementation

    There are a large number of ways to send messages, and the specifics of
    the transport methods are not specifically relevant to Pact. As such,
    Pact abstracts the transport layer away and uses a lightweight HTTP
    server to handle messages.

    Pact Python is capable of setting up this server and handling the
    messages internally using user-provided handlers. It is possible to use
    your own HTTP server to handle messages by using the `add_transport`
    method. It is not possible to use both this method and `add_transport`
    to handle messages.

    Args:
        handler:
            The message handler. This should be a callable that takes no
            arguments: the
    """
    logger.debug(
        "Setting message handler for verifier",
        extra={
            "path": "/_pact/message",
        },
    )

    if callable(handler):
        if len(inspect.signature(handler).parameters) != 2:  # noqa: PLR2004
            msg = "The function must take two arguments: name and parameters"
            raise TypeError(msg)

        self._message_producer = MessageProducer(handler)
        self.add_transport(
            protocol="message",
            port=self._message_producer.port,
            path=self._message_producer.path,
        )

    if isinstance(handler, dict):
        # Check that all values are either callable with one argument, or
        # bytes.
        for value in handler.values():
            if callable(value) and len(inspect.signature(value).parameters) != 1:
                msg = "All functions must take one argument: parameters"
                raise TypeError(msg)
            if not callable(value) and not isinstance(value, dict):
                msg = "All values must be callable or dictionaries"
                raise TypeError(msg)

        def _handler(name: str, parameters: dict[str, Any] | None) -> Message:
            logger.info("Internal handler called")
            val = handler[name]
            if callable(val):
                return val(parameters)
            return val

        self._message_producer = MessageProducer(_handler)
        self.add_transport(
            protocol="message",
            port=self._message_producer.port,
            path=self._message_producer.path,
        )

    return self
output(*, strip_ansi: bool = False) -> str

Get the output.

Source code in src/pact/v3/verifier.py
def output(self, *, strip_ansi: bool = False) -> str:
    """
    Get the output.
    """
    return pact.v3.ffi.verifier_output(self._handle, strip_ansi=strip_ansi)
set_coloured_output(*, enabled: bool = True) -> Self

Toggle coloured output.

Source code in src/pact/v3/verifier.py
def set_coloured_output(self, *, enabled: bool = True) -> Self:
    """
    Toggle coloured output.
    """
    pact.v3.ffi.verifier_set_coloured_output(self._handle, enabled=enabled)
    return self
set_error_on_empty_pact(*, enabled: bool = True) -> Self

Toggle error on empty pact.

If enabled, a Pact file with no interactions will cause the verifier to return an error. If disabled, a Pact file with no interactions will be ignored.

Source code in src/pact/v3/verifier.py
def set_error_on_empty_pact(self, *, enabled: bool = True) -> Self:
    """
    Toggle error on empty pact.

    If enabled, a Pact file with no interactions will cause the verifier to
    return an error. If disabled, a Pact file with no interactions will be
    ignored.
    """
    pact.v3.ffi.verifier_set_no_pacts_is_error(self._handle, enabled=enabled)
    return self
set_publish_options(version: str, url: str | None = None, branch: str | None = None, tags: list[str] | None = None) -> Self

Set options used when publishing results to the Broker.

PARAMETER DESCRIPTION
version

The provider version.

TYPE: str

url

URL to the build which ran the verification.

TYPE: str | None DEFAULT: None

tags

Collection of tags for the provider.

TYPE: list[str] | None DEFAULT: None

branch

Name of the branch used for verification.

TYPE: str | None DEFAULT: None

Source code in src/pact/v3/verifier.py
def set_publish_options(
    self,
    version: str,
    url: str | None = None,
    branch: str | None = None,
    tags: list[str] | None = None,
) -> Self:
    """
    Set options used when publishing results to the Broker.

    Args:
        version:
            The provider version.

        url:
            URL to the build which ran the verification.

        tags:
            Collection of tags for the provider.

        branch:
            Name of the branch used for verification.
    """
    pact.v3.ffi.verifier_set_publish_options(
        self._handle,
        version,
        url,
        tags or [],
        branch,
    )
    return self
set_request_timeout(timeout: int) -> Self

Set the request timeout.

PARAMETER DESCRIPTION
timeout

The request timeout in milliseconds.

TYPE: int

Source code in src/pact/v3/verifier.py
def set_request_timeout(self, timeout: int) -> Self:
    """
    Set the request timeout.

    Args:
        timeout:
            The request timeout in milliseconds.
    """
    if timeout < 0:
        msg = "Request timeout must be a positive integer"
        raise ValueError(msg)

    self._request_timeout = timeout
    pact.v3.ffi.verifier_set_verification_options(
        self._handle,
        disable_ssl_verification=self._disable_ssl_verification,
        request_timeout=self._request_timeout,
    )
    return self
state_handler(handler: StateHandlerFull | StateHandlerNoAction | dict[str, StateHandlerNoState] | dict[str, StateHandlerNoActionNoState] | StateHandlerUrl, *, teardown: bool = False, body: bool | None = None) -> Self
state_handler(
    handler: StateHandlerFull,
    *,
    teardown: Literal[True],
    body: None = None,
) -> Self
state_handler(
    handler: StateHandlerNoAction,
    *,
    teardown: Literal[False] = False,
    body: None = None,
) -> Self
state_handler(
    handler: dict[str, StateHandlerNoState],
    *,
    teardown: Literal[True],
    body: None = None,
) -> Self
state_handler(
    handler: dict[str, StateHandlerNoActionNoState],
    *,
    teardown: Literal[False] = False,
    body: None = None,
) -> Self
state_handler(
    handler: StateHandlerUrl,
    *,
    teardown: bool = False,
    body: bool,
) -> Self

Set the state handler.

In many interactions, the consumer will assume that the provider is in a certain state. For example, a consumer requesting information about a user with ID 123 will have specified given("user with ID 123 exists").

The state handler is responsible for changing the provider's internal state to match the expected state before the interaction is replayed.

This can be done in one of three ways:

  1. By providing a single function that will be called for all state changes.
  2. By providing a mapping of state names to functions.
  3. By providing the URL endpoint to which the request should be made.

The first two options are most straightforward to use.

When providing a function, the arguments should be:

  1. The state name, as a string.
  2. The action (either setup or teardown), as a string.
  3. A dictionary of parameters, or None if no parameters are provided.

Note that these arguments will change in the following ways:

  1. If a dictionary mapping is used, the state name is not provided to the function.
  2. If teardown is False thereby indicating that the function is only called for setup, the action argument is not provided.

This means that in the case of a dictionary mapping of function with teardown=False, the function should take only one argument: the dictionary of parameters (which itself may be None, albeit still an argument).

PARAMETER DESCRIPTION
handler

The handler for the state changes. This can be one of the following:

  • A single function that will be called for all state changes.
  • A dictionary mapping state names to functions.
  • A URL endpoint to which the request should be made.

See above for more information on the function signature.

TYPE: StateHandlerFull | StateHandlerNoAction | dict[str, StateHandlerNoState] | dict[str, StateHandlerNoActionNoState] | StateHandlerUrl

teardown

Whether to teardown the provider state after an interaction is validated.

TYPE: bool DEFAULT: False

body

Whether to include the state change request in the body (True) or in the query string (False). This must be left as None if providing one or more handler functions; and it must be set to a boolean if providing a URL.

TYPE: bool | None DEFAULT: None

Source code in src/pact/v3/verifier.py
def state_handler(
    self,
    handler: StateHandlerFull
    | StateHandlerNoAction
    | dict[str, StateHandlerNoState]
    | dict[str, StateHandlerNoActionNoState]
    | StateHandlerUrl,
    *,
    teardown: bool = False,
    body: bool | None = None,
) -> Self:
    """
    Set the state handler.

    In many interactions, the consumer will assume that the provider is in a
    certain state. For example, a consumer requesting information about a
    user with ID `123` will have specified `given("user with ID 123
    exists")`.

    The state handler is responsible for changing the provider's internal
    state to match the expected state before the interaction is replayed.

    This can be done in one of three ways:

    1.  By providing a single function that will be called for all state
        changes.
    2.  By providing a mapping of state names to functions.
    3.  By providing the URL endpoint to which the request should be made.

    The first two options are most straightforward to use.

    When providing a function, the arguments should be:

    1.  The state name, as a string.
    2.  The action (either `setup` or `teardown`), as a string.
    3.  A dictionary of parameters, or `None` if no parameters are provided.

    Note that these arguments will change in the following ways:

    1.  If a dictionary mapping is used, the state name is _not_ provided to
        the function.
    2.  If `teardown` is `False` thereby indicating that the function is
        only called for setup, the `action` argument is not provided.

    This means that in the case of a dictionary mapping of function with
    `teardown=False`, the function should take only one argument: the
    dictionary of parameters (which itself may be `None`, albeit still an
    argument).

    Args:
        handler:
            The handler for the state changes. This can be one of the
            following:

            -   A single function that will be called for all state changes.
            -   A dictionary mapping state names to functions.
            -   A URL endpoint to which the request should be made.

            See above for more information on the function signature.

        teardown:
            Whether to teardown the provider state after an interaction is
            validated.

        body:
            Whether to include the state change request in the body (`True`)
            or in the query string (`False`). This must be left as `None` if
            providing one or more handler functions; and it must be set to
            a boolean if providing a URL.
    """
    # A tuple is required instead of `StateHandlerUrl` for support for
    # Python 3.9. This should be changed to `StateHandlerUrl` in the future.
    if isinstance(handler, (str, URL)):
        if body is None:
            msg = "The `body` parameter must be a boolean when providing a URL"
            raise ValueError(msg)
        return self._state_handler_url(handler, teardown=teardown, body=body)

    if isinstance(handler, dict):
        if body is not None:
            msg = "The `body` parameter must be `None` when providing a dictionary"
            raise ValueError(msg)
        return self._state_handler_dict(handler, teardown=teardown)

    if callable(handler):
        if body is not None:
            msg = "The `body` parameter must be `None` when providing a function"
            raise ValueError(msg)
        return self._set_function_state_handler(handler, teardown=teardown)

    msg = "Invalid handler type"
    raise TypeError(msg)
verify() -> Self

Verify the interactions.

RETURNS DESCRIPTION
Self

Whether the interactions were verified successfully.

Source code in src/pact/v3/verifier.py
def verify(self) -> Self:
    """
    Verify the interactions.

    Returns:
        Whether the interactions were verified successfully.
    """
    if not self._transports:
        msg = "No transports have been set"
        raise RuntimeError(msg)

    first, *rest = self._transports

    pact.v3.ffi.verifier_set_provider_info(
        self._handle,
        self._name,
        first["scheme"],
        self._host,
        first["port"],
        first["path"],
    )

    for transport in rest:
        pact.v3.ffi.verifier_add_provider_transport(
            self._handle,
            transport["transport"],
            transport["port"] or 0,
            transport["path"],
            transport["scheme"],
        )

    with self._message_producer, self._state_handler:
        pact.v3.ffi.verifier_execute(self._handle)
        logger.debug("Verifier executed")

    return self