Skip to content

Message Provider

Contract Message Provider.

Classes

MessageProvider(message_providers, provider, consumer, pact_dir=os.getcwd(), version='3.0.0', proxy_host='localhost', proxy_port='1234')

Bases: object

A Pact message provider.

provider = MessageProvider( message_providers = { "a document created successfully": handler }, provider='DocumentService', pact_dir='pacts', version='3.0.0' )

Source code in src/pact/message_provider.py
def __init__(
    self,
    message_providers,
    provider,
    consumer,
    pact_dir=os.getcwd(),
    version="3.0.0",
    proxy_host='localhost',
    proxy_port='1234'
):
    """Create a Message Provider instance."""
    warnings.warn(
        "This class will be deprecated Pact Python v3 "
        "(see pact-foundation/pact-python#396)",
        PendingDeprecationWarning,
        stacklevel=2,
    )
    self.message_providers = message_providers
    self.provider = provider
    self.consumer = consumer
    self.version = version
    self.pact_dir = pact_dir
    self.proxy_host = proxy_host
    self.proxy_port = proxy_port
    self._process = None

Attributes

consumer = consumer instance-attribute
message_providers = message_providers instance-attribute
pact_dir = pact_dir instance-attribute
provider = provider instance-attribute
proxy_host = proxy_host instance-attribute
proxy_port = proxy_port instance-attribute
version = version instance-attribute

Functions

verify()

Verify pact files with executable verifier.

Source code in src/pact/message_provider.py
def verify(self):
    """Verify pact files with executable verifier."""
    pact_files = f'{self.pact_dir}/{self._pact_file()}'
    verifier = Verifier(provider=self.provider,
                        provider_base_url=self._proxy_url())
    return_code, _ = verifier.verify_pacts(pact_files, verbose=False)
    assert (return_code == 0), f'Expected returned_code = 0, actual = {return_code}'
verify_with_broker(enable_pending=False, include_wip_pacts_since=None, **kwargs)

Use Broker to verify.

PARAMETER DESCRIPTION
broker_username

broker username

TYPE: [String]

broker_password

broker password

TYPE: [String]

broker_url

url of broker

TYPE: [String]

Source code in src/pact/message_provider.py
def verify_with_broker(self, enable_pending=False, include_wip_pacts_since=None, **kwargs):
    """Use Broker to verify.

    Args:
        broker_username ([String]): broker username
        broker_password ([String]): broker password
        broker_url ([String]): url of broker
        enable_pending ([Boolean])
        include_wip_pacts_since ([String])
        publish_version ([String])

    """
    verifier = Verifier(provider=self.provider,
                        provider_base_url=self._proxy_url())

    return_code, _ = verifier.verify_with_broker(enable_pending, include_wip_pacts_since, **kwargs)

    assert (return_code == 0), f'Expected returned_code = 0, actual = {return_code}'

Functions