Contract Testing Microservices with Pact using Python: A Beginner’s Guide
Consumer-driven Contract testing made easy :)
What is contract testing?
Contract testing is a software testing technique that verifies the expected behavior of an API. It does this by creating a contract, which is a formal agreement between the Consumer and the Provider of an API. The contract specifies the expected request and response messages for each API endpoint.
In consumer-driven contract testing, the consumer is responsible for creating the contracts and the provider is responsible for abiding by those contracts.
This actually makes a lot of sense. A consumer needs to make announce the contracts it needs from all the producers, and a producer needs to make sure that all contracts associated with it are passed.
Why contract testing?
- Improved quality: Contract testing can help to improve the quality of APIs and microservices by ensuring that they are interoperable and that they meet the needs of the users.
- Reduced risk: Contract testing can help to reduce the risk of regressions when changes are made to APIs and microservices.
- Increased confidence: Contract testing can help to increase confidence in the quality of APIs and microservices.
- Automated testing: Contract testing can be automated, which can help to save time and resources.
What is Pact? And its components
Pact is a framework for testing APIs and microservices. It involves having the service consumer and service provider agree on the expected behavior of the API.
This is done by creating a pact file, which is a contract between the two applications. The pact file specifies the expected requests and responses for each API endpoint. The pact file is uploaded to the pact broker, which is a centralized repository for storing contracts.
The flow is as follows:
- The consumer creates a mock provider and writes the integrations against the mock provider.
- These interactions are recorded in a file (contract).
- The contracts are published to the broker.
- The provider on it’s end would validate run the mock consumer and replay all the interactions. It would check and make sure that all integrations are passed.
Example
To keep it simple, we have a consumer who makes an API call to the provider. It looks something like this:
import requests
def user():
uri = 'http://localhost:8001/' # provider
res = requests.get(uri)
return res.json()
if __name__ == '__main__':
print(user())
The contract tests for the consumer would be as follows:
import atexit
import unittest
from pact import Consumer, Provider
from consumer import user
pact = Consumer('IP-test-AndroidApp').has_pact_with(Provider('IP-test-BEService'), host_name='localhost', port=8001)
pact.start_service()
atexit.register(pact.stop_service)
class GetUserInfoContract(unittest.TestCase):
def test_get_user(self):
expected = {
"hello": "world"
}
(pact
.upon_receiving('a request')
.with_request('get', '/')
.will_respond_with(200, body=expected))
with pact:
result = user()
pact.verify()
self.assertEqual(result, expected)
In the above consumer-contract test,
- We have started by defining a Consumer and the relation it has with a Provider.
- Start a mock server on localhost:8001.
- Define the contract and validate the code against the specified contract.
- This would generate a json file in the directory, which is the contract of the app.
Now we need to publish this contract to a broker. I’ll use the docker image to achieve this (you can use the cli as well):
docker run --rm \
-w ${PWD} \
-v ${PWD}:${PWD} \
-e PACT_BROKER_BASE_URL \
-e PACT_BROKER_USERNAME \
-e PACT_BROKER_PASSWORD \
--network=host \
pactfoundation/pact-cli:latest \
publish \
pacts/*.json \
--consumer-app-version dev-test
The pact would show up like this on the broker:
If you check the matrix, it would show that contract has been published but the provider hasn’t verified it yet.
Now, let’s have a provider ready. We’ll use a Flask server with one endpoint:
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return {"hello": "world"}
if __name__ == "__main__":
app.run("0.0.0.0", port=8001)
Let’s write the provider contract test:
from pact import Verifier
import unittest
broker_opts = {
"broker_username": "pactbroker",
"broker_password": "pactbroker",
"broker_url": "http://localhost/",
"publish_version": "3",
"publish_verification_results": True,
}
verifier = Verifier(provider="IP-test-BEService", provider_base_url="http://localhost:8001")
class GetUserInfoContract(unittest.TestCase):
def test_get_user(self):
success, logs = verifier.verify_with_broker(
**broker_opts,
verbose=True,
# provider_states_setup_url=f"{PROVIDER_URL}/_pact/provider_states",
enable_pending=False,
)
assert success==0
Here we’ve created a Verifier, which fetches the contract from Broker (hosted on localhost) and runs all the contracts against the provider, which is running locally.
After executing the above test, the matrix is updated:
This sets up your basic consumer-driven contract tests using Pact.
Further steps
This is just a small setup, we can extend this setup into a full fledge CICD set up by the following:
- Integrating multiple consumers and providers.
- Integrating webhooks to alert for events.
- Integrating into CICD workflow — contracts to be published as part of the CICD of the consumer and to be verified as part of the CICD of the provider.
Conclusion
Contract testing is a powerful technique that can help to improve the quality of microservices applications. By using Pact, you can ensure that your microservices are compatible with each other and that your deployments are safe.
I hope this blog post has helped you to understand contract testing and how to use Pact. If you have any questions, please feel free to leave a comment.
Resources:
- https://docs.pact.io/ — official website
- https://openai.com/dall-e-2 — for the cover image