Skip to content

Message Pact

API for creating a contract and configuring the mock service.

Attributes

Classes

MessagePact(consumer, provider, publish_to_broker=False, broker_base_url=None, broker_username=None, broker_password=None, broker_token=None, pact_dir=None, version='3.0.0', file_write_mode='merge')

Bases: Broker

Represents a contract between a consumer and provider using messages.

Provides Python context handler to perform tests on a Python consumer. For example:

from pact import MessageConsumer, Provider pact = MessageConsumer('MyMessageConsumer').has_pact_with(Provider('provider')) (pact ... .given({"name": "Test provider"}]) ... .expects_to_receive('Test description') ... .with_content({'name': 'John', 'document_name': 'sample_document.doc'}) ... .with_metadata({'contentType': 'application/json'})) with pact: ... handler(event, context)

:param consumer: A consumer for this contract that uses messages. :type consumer: pact.MessageConsumer :param provider: The generic provider for this contract. :type provider: pact.Provider :param publish_to_broker: Flag to control automatic publishing of pacts to a pact broker. Defaults to False. :type publish_to_broker: bool :param broker_base_url: URL of the pact broker that pacts will be published to. Can also be supplied through the PACT_BROKER_BASE_URL environment variable. Defaults to None. :type broker_base_url: str :param broker_username: Username to use when connecting to the pact broker if authentication is required. Can also be supplied through the PACT_BROKER_USERNAME environment variable. Defaults to None. :type broker_username: str :param broker_password: Password to use when connecting to the pact broker if authentication is required. Strongly recommend supplying this value through the PACT_BROKER_PASSWORD environment variable instead. Defaults to None. :type broker_password: str :param broker_token: Authentication token to use when connecting to the pact broker. Strongly recommend supplying this value through the PACT_BROKER_TOKEN environment variable instead. Defaults to None. :type broker_token: str :param pact_dir: Directory where the resulting pact files will be written. Defaults to the current directory. :type pact_dir: str :param version: The Pact Specification version to use, defaults to '3.0.0'. :type version: str :param file_write_mode: overwrite or merge. Use merge when running multiple mock service instances in parallel for the same consumer/provider pair. Ensure the pact file is deleted before running tests when using this option so that interactions deleted from the code are not maintained in the file. Defaults to merge. :type file_write_mode: str

Source code in src/pact/message_pact.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def __init__(
    self,
    consumer,
    provider,
    publish_to_broker=False,
    broker_base_url=None,
    broker_username=None,
    broker_password=None,
    broker_token=None,
    pact_dir=None,
    version='3.0.0',
    file_write_mode="merge",
):
    """
    Create a Pact instance using messages.

    :param consumer: A consumer for this contract that uses messages.
    :type consumer: pact.MessageConsumer
    :param provider: The generic provider for this contract.
    :type provider: pact.Provider
    :param publish_to_broker: Flag to control automatic publishing of
        pacts to a pact broker. Defaults to False.
    :type publish_to_broker: bool
    :param broker_base_url: URL of the pact broker that pacts will be
        published to. Can also be supplied through the PACT_BROKER_BASE_URL
        environment variable. Defaults to None.
    :type broker_base_url: str
    :param broker_username: Username to use when connecting to the pact
        broker if authentication is required. Can also be supplied through
        the PACT_BROKER_USERNAME environment variable. Defaults to None.
    :type broker_username: str
    :param broker_password: Password to use when connecting to the pact
        broker if authentication is required. Strongly recommend supplying
        this value through the PACT_BROKER_PASSWORD environment variable
        instead. Defaults to None.
    :type broker_password: str
    :param broker_token: Authentication token to use when connecting to
        the pact broker. Strongly recommend supplying this value through
        the PACT_BROKER_TOKEN environment variable instead.
        Defaults to None.
    :type broker_token: str
    :param pact_dir: Directory where the resulting pact files will be
        written. Defaults to the current directory.
    :type pact_dir: str
    :param version: The Pact Specification version to use, defaults to
        '3.0.0'.
    :type version: str
    :param file_write_mode: `overwrite` or `merge`. Use `merge` when
        running multiple mock service instances in parallel for the same
        consumer/provider pair. Ensure the pact file is deleted before
        running tests when using this option so that interactions deleted
        from the code are not maintained in the file. Defaults to
        `merge`.
    :type file_write_mode: str
    """
    warnings.warn(
        "This class will be deprecated Pact Python v3 "
        "(see pact-foundation/pact-python#396)",
        PendingDeprecationWarning,
        stacklevel=2,
    )
    super().__init__(
        broker_base_url, broker_username, broker_password, broker_token
    )

    self.consumer = consumer
    self.file_write_mode = file_write_mode
    self.pact_dir = pact_dir or os.getcwd()
    self.provider = provider
    self.publish_to_broker = publish_to_broker
    self.version = version
    self._process = None
    self._messages = []
    self._message_process = None

Attributes

MANDATORY_FIELDS = {'providerStates', 'description', 'contents', 'metaData'} class-attribute instance-attribute
consumer = consumer instance-attribute
file_write_mode = file_write_mode instance-attribute
pact_dir = pact_dir or os.getcwd() instance-attribute
provider = provider instance-attribute
publish_to_broker = publish_to_broker instance-attribute
version = version instance-attribute

Functions

expects_to_receive(description)

Define the name of this contract (Same as upon_receiving in http pact implementation).

:param scenario: A unique name for this contract. :type scenario: basestring :rtype: Pact

Source code in src/pact/message_pact.py 
159
160
161
162
163
164
165
166
167
168
169
def expects_to_receive(self, description):
    """
    Define the name of this contract (Same as upon_receiving in http pact implementation).

    :param scenario: A unique name for this contract.
    :type scenario: basestring
    :rtype: Pact
    """
    self._insert_message_if_complete()
    self._messages[0]['description'] = description
    return self
given(name, params=None)

Define the provider state for this pact.

When the provider verifies this contract, they will use this field to setup pre-defined data that will satisfy the response expectations.

:param name: The short sentence that is unique to describe the provider state for this contract. :type name: basestring :param params: Additional arguments necessary to set the provider state :type params: dict :rtype: Pact

Source code in src/pact/message_pact.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def given(self, name, params=None):
    """
    Define the provider state for this pact.

    When the provider verifies this contract, they will use this field to
    setup pre-defined data that will satisfy the response expectations.

    :param name: The short sentence that is unique to describe the provider
    state for this contract.
    :type name: basestring
    :param params: Additional arguments necessary to set the provider state
    :type params: dict
    :rtype: Pact
    """
    self._insert_message_if_complete()

    provider_state = {'name': "{}".format(name)}
    if params:
        provider_state['params'] = params

    if 'providerStates' not in self._messages[0]:
        self._messages[0]['providerStates'] = [provider_state]
    else:
        self._messages[0]['providerStates'].append(provider_state)

    return self
with_content(contents)

Define message content (event) that will be use in the message.

:param contents: dictionary of dictionary used in the message. :type metadata: dict :rtype: Pact

Source code in src/pact/message_pact.py 
147
148
149
150
151
152
153
154
155
156
157
def with_content(self, contents):
    """
    Define message content (event) that will be use in the message.

    :param contents: dictionary of dictionary used in the message.
    :type metadata: dict
    :rtype: Pact
    """
    self._insert_message_if_complete()
    self._messages[0]['contents'] = from_term(contents)
    return self
with_metadata(metadata)

Define metadata attached to the message.

:param metadata: dictionary of metadata attached to the message. :type metadata: dict or None :rtype: Pact

Source code in src/pact/message_pact.py 
135
136
137
138
139
140
141
142
143
144
145
def with_metadata(self, metadata):
    """
    Define metadata attached to the message.

    :param metadata: dictionary of metadata attached to the message.
    :type metadata: dict or None
    :rtype: Pact
    """
    self._insert_message_if_complete()
    self._messages[0]['metaData'] = from_term(metadata)
    return self
write_to_pact_file()

Create a pact file based on provided attributes in DSL.

Return 0 if success, 1 otherwise.

:rtype: int

Source code in src/pact/message_pact.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def write_to_pact_file(self):
    """
    Create a pact file based on provided attributes in DSL.

    Return 0 if success, 1 otherwise.

    :rtype: int
    """
    command = [
        MESSAGE_PATH,
        "update",
        json.dumps(self._messages[0]),
        "--pact-dir", self.pact_dir,
        f"--pact-specification-version={self.version}",
        "--consumer", f"{self.consumer.name}",
        "--provider", f"{self.provider.name}",
    ]

    self._message_process = Popen(command)
    self._message_process.wait()

Functions