Writing Contract test for API endpoint using Pact

This post shows you a short example of how to use the Python module Pact to write Consumer Contract tests against Provider Contract. We will use Trello API endpoints for writing sample test. Note that this blog will only cover one sample test. Testers can write many more for a different response such as unauthorized access, server not found.

What is Contract testing?

Contract testing is a testing technique used for testing endpoints, for applications which are isolated. Messages sent or received confirms that they are adhering to a written Pact or Contract. This helps the tester to confirm all calls made to the Mock Provider, returns the same response as actual service. This reference link will give more information about Contract testing.

Background

Python package of Pact helps in writing consumer-driven python tests. As shown in the below diagram, while running Pact Test, Consumer does not contact actual Provider, but it will contact Mock Provider. The mock provider will validate the defined contract and send the appropriate response. The advantage over here is you don’t have to spin up actual Provider setup.

API Endpoint under Test

We will be using /1/boards/{id} endpoint for GET request of Trello API’s. This request will be used to fetch a single Trello board based on the ID. More information can be found at Trello API Documentation

Using pact-python

The Python module pact makes it really easy for us to define pact between REST API endpoint Provider and Consumer of those endpoints. The next steps will help to write the first test:

1.Define the Consumer and Provider objects that describe API endpoint and expected payload
2.Define the setup criteria for the Provider
3.Define the Consumer request using Pact
4.Define how the provider is expected to respond using Pact
5.Assert the response to validate the contract

1. Define the Consumer and Provider objects that describe API endpoint and expected payload

Create a new file called test_trello_get_board.py and add the following lines. Note that you will require to import Unittest, atexit and Pytest to run the test.

"contract test for Trello board"
import os
import json
import atexit
import unittest
import pytest
import requests
from pact import Consumer, Provider
 
# Setting up pact
pact = Consumer('Consumer').has_pact_with(Provider('Provider'))
pact.start_service()
atexit.register(pact.stop_service)
 
#setting up path for pact file
CURR_FILE_PATH = os.path.dirname(os.path.abspath(__file__))
PACT_DIR = os.path.join(CURR_FILE_PATH, '')
PACT_FILE = os.path.join(PACT_DIR, 'pact.json')
 
# Defining Class
class GetBoard(unittest.TestCase):
  def test_get_board(self):
   # Defining test method
     with open(os.path.join(PACT_DIR, PACT_FILE), 'rb') as pact_file:
        pact_file_json = json.load(pact_file)
        expected = pact_file_json
2. Define the setup criteria for the Provider

Using .given, setup criteria for the Provider will be defined as shown below.

(pact
  .given('Response Payload will be received as expected')
  .upon_receiving('a request for get trello board')
3. Define the Consumer request using Pact

Using .with_request, you can define the request type and API endpoint to which the request needs to be made.

(pact
  .given('Request for get trello board')
  .upon_receiving('a request for get trello board')
  .with_request('GET','/1/boards/1000')
4. Define how the Provider is expected to response using Pact

You can define the Provider response using .will_respond_with which will include status code and expected contract payload.

(pact
  .given('Request for get trello board')
  .upon_receiving('a request for get trello board')
  .with_request('GET','/1/boards/1000')
  .will_respond_with(200, body=expected))
5. Assert the response with expected contract

The below steps will help you to assert the response with a defined expected contract.

with pact:
    result = requests.get(pact.uri + '/1/boards/1000')
 
    self.assertEqual(result.json(), expected)
    pact.verify()

Putting it all together

Here is how our test looks:

"contract test for trello board"
import os
import json
import logging
import atexit
import unittest
import pytest
import requests
from pact import Consumer, Provider, Format
 
# Declaring logger
log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
print(Format().__dict__)
 
# Setting up pact
pact = Consumer('Consumer').has_pact_with(Provider('Provider'))
pact.start_service()
atexit.register(pact.stop_service)
 
#setting up path for pact file
CURR_FILE_PATH = os.path.dirname(os.path.abspath(__file__))
PACT_DIR = os.path.join(CURR_FILE_PATH, '')
PACT_FILE = os.path.join(PACT_DIR, 'pact.json')
 
# Defining Class
class GetBoard(unittest.TestCase):
  def test_get_board(self):
     """
     Defining test method
     """
     with open(os.path.join(PACT_DIR, PACT_FILE), 'rb') as pact_file:
        pact_file_json = json.load(pact_file)
        expected = pact_file_json
 
        (pact
         .given('Request to send message')
         .upon_receiving('a request for response for send message')
         .with_request('GET','/1/boards/1000')
         .will_respond_with(200, body=expected))
 
        with pact:
           result = requests.get(pact.uri + '/1/boards/1000')
 
        self.assertEqual(result.json(), expected)
        pact.verify()

A sample pact.json file would look like below.

# pact.json
{
    "id":"1000",
 "desc":"Track changes to Trello's Platform on this board.",
 "descData":"test",
 "closed":"false",
 "idMemberCreator":"1000",
 "idOrganization":"1000",
 "pinned":"false",
 "url":"https://trello.com/b/dQHqCohZ/trello-platform-changelog",
 "shortUrl":"https://trello.com/b/dQHqCohZ",
 "prefs":{
    "permissionLevel":"org",
    "hideVotes":"true",
    "voting":"disabled",
    "comments":"test",
    "selfJoin":"true",
    "cardCovers":"true",
    "isTemplate":"true",
    "cardAging":"pirate",
    "calendarFeedEnabled":"true",
    "background":"1000",
    "backgroundImage":"test",
    "backgroundImageScaled":"test"
 },
 "labelNames":{
    "green":"Addition",
    "yellow":"Update",
    "orange":"Deprecation",
    "red":"Deletion",
    "purple":"Power-Ups",
    "blue":"News",
    "sky":"Announcement",
    "lime":"Delight",
    "pink":"REST API",
    "black":"Capabilties"
 },
 "limits":{
    "attachments":{
       "perBoard":{
          "status":"ok",
          "disableAt":36000,
          "warnAt":32400
       }
    }
 },
 "starred":"true",
 "memberships":"test",
 "shortLink":"test",
 "subscribed":"true",
 "powerUps":"test",
 "dateLastActivity":"test",
 "dateLastView":"test",
 "idTags":"test",
 "datePluginDisable":"test",
 "creationMethod":"test",
 "ixUpdate":2154,
 "templateGallery":"test",
 "enterpriseOwned":"true"
  }

After putting up everything you can run the test using python -m pytest test_trello_get_board.py.

Note: I have run the test on WSL(Windows Subsystem for Linux).

After the test runs successfully, you will find consumer-provider.json created. Here, you can verify response status as 200.The consumer-provide.json will look like as below:

# consumer-provide.json
{
  "consumer": {
    "name": "Consumer"
  },
  "provider": {
    "name": "Provider"
  },
  "interactions": [
    {
      "description": "a request for get trello board",
      "providerState": "Response Payload will be received as expected",
      "request": {
        "method": "GET",
        "path": "/1/boards/1000"
      },
      "response": {
        "status": 200,
        "headers": {
        },
        "body": {
          "id": "1000",
          "desc": "Track changes to Trello's Platform on this board.",
          "descData": "test",
          "closed": "false",
          "idMemberCreator": "1000",
          "idOrganization": "1000",
          "pinned": "false",
          "url": "https://trello.com/b/dQHqCohZ/trello-platform-changelog",
          "shortUrl": "https://trello.com/b/dQHqCohZ",
          "prefs": {
            "permissionLevel": "org",
            "hideVotes": "true",
            "voting": "disabled",
            "comments": "test",
            "selfJoin": "true",
            "cardCovers": "true",
            "isTemplate": "true",
            "cardAging": "pirate",
            "calendarFeedEnabled": "true",
            "background": "1000",
            "backgroundImage": "test",
            "backgroundImageScaled": "test"
          },
          "labelNames": {
            "green": "Addition",
            "yellow": "Update",
            "orange": "Deprecation",
            "red": "Deletion",
            "purple": "Power-Ups",
            "blue": "News",
            "sky": "Announcement",
            "lime": "Delight",
            "pink": "REST API",
            "black": "Capabilties"
          },
          "limits": {
            "attachments": {
              "perBoard": {
                "status": "ok",
                "disableAt": 36000,
                "warnAt": 32400
              }
            }
          },
          "starred": "true",
          "memberships": "test",
          "shortLink": "test",
          "subscribed": "true",
          "powerUps": "test",
          "dateLastActivity": "test",
          "dateLastView": "test",
          "idTags": "test",
          "datePluginDisable": "test",
          "creationMethod": "test",
          "ixUpdate": 2154,
          "templateGallery": "test",
          "enterpriseOwned": "true"
        }
      }
    }
  ],
  "metadata": {
    "pactSpecification": {
      "version": "2.0.0"
    }
  }
}

If you are a tester working on testing microservices then the above example will help you to add contract tests to your test strategy.

Related posts:

Default ThumbnailHow to stop Pact Mock Service manually? Default ThumbnailAutomate creating new boards on Trello using Python Default ThumbnailPython Trello API utility script
Rahul Bhave

12 thoughts on “Writing Contract test for API endpoint using Pact”

  1. Thanks for the post, however I’m running into some issues and was hoping you could help? I have the following:
    (pact
    .given(‘Short Form Transfer for FDR’)
    .upon_receiving(‘An offer_id from XPS Evaluate’)
    .with_request(‘POST’, transPost)
    .will_respond_with(200))
    …
    And then:
    with pact:
    result = requests.post(transPost, headers=headers, data=json.dumps(transEx1))

    Where the ‘transEx1’ is the post request, and I know this works fine but with Pact I keep getting the error:
    Verifying – actual interactions do not match expected interactions.
    Missing requests:
    POST ‘transPost’

    But the ‘transPost’ variable IS in the pact-mock-service.log? I’m keeping the URL private but just wondering what I might be doing wrong?

    Thanks,
    Rahul

    Reply

    1. Hi Rahul,

      Log message shared by you mentions, `Missing requests: Post transPost`, I think pact is not getting proper post request, you may have to re-check following steps:

      I.
      .with_request(‘POST’,’transPost’)- I think in this step the path should come after ‘/’ like ‘/transPost’- You can check this based on your API

      II.
      In the below step should include pact.uri before path e.g. result= resquests.post(pact.uri + ‘/transPost’)
      result = requests.post(transPost, headers=headers, data=json.dumps(transEx1))

      III.
      Please re look into expected response as well which you are passing in the test:

      I think below sample with POST request may help above 3 points(Please note that, this may not be the similar to your test but this may help you to check above 3 points. You can also refer examples given at https://pypi.org/project/pact-python/)

      “`
      “sample post”
      import logging
      import atexit
      import unittest
      import pytest
      import requests
      from pact import Consumer, Provider, Format

      # Declaring logger
      log = logging.getLogger(__name__)
      logging.basicConfig(level=logging.INFO)
      print(Format().__dict__)

      # Setting up pact
      pact = Consumer(‘Consumer’).has_pact_with(Provider(‘Provider’))
      pact.start_service()
      atexit.register(pact.stop_service)

      # Defining Class
      class PostReaction(unittest.TestCase):
      def test_post_reactions(self):
      “””
      Defining test method
      “””
      expected_response = {“status”: 200,”headers”: {‘Content-Type’: ‘application/json’},”body”: {‘msg’: ‘Success’}}
      (pact
      .given(‘Request to post reaction’)
      .upon_receiving(‘a request for response after post reaction’)
      .with_request(‘POST’,’/1/actions/id/reactions/’)
      .will_respond_with(200, body=expected_response))
      with pact:
      result = requests.post(pact.uri + ‘/1/actions/id/reactions/’)

      self.assertEqual(result.json(), expected_response)

      pact.verify()

      “`
      This is how my consumer-provider.json look like:

      “`

      {
      “consumer”: {
      “name”: “Consumer”
      },
      “provider”: {
      “name”: “Provider”
      },
      “interactions”: [
      {
      “description”: “a request for response after post reaction”,
      “providerState”: “Request to post reaction”,
      “request”: {
      “method”: “POST”,
      “path”: “/1/actions/id/reactions/”
      },
      “response”: {
      “status”: 200,
      “headers”: {
      },
      “body”: {
      “status”: 200,
      “headers”: {
      “Content-Type”: “application/json”
      },
      “body”: {
      “msg”: “Success”
      }
      }
      }
      }
      ],
      “metadata”: {
      “pactSpecification”: {
      “version”: “2.0.0”
      }
      }
      }
      “`

      Regards,
      Rahul

      Reply

      1. Thanks for the response, but that leads me to some different questions:
        1. What is “pact.uri” defined as, and where is it defined in your example above?
        2. In your reply you have in your example: result = requests.post(pact.uri + ‘/1/actions/id/reactions/’)
        But where is the actual POST data being passed in? And is that POST data supposed to be the consumer-provider.json, or is the consumer-provider.json the result of running the test?

      2. Hi Rahul,

        Below are my comments on your queries:

        1. What is “pact.uri” defined as, and where is it defined in your example above?

        You do not need to define pact.uri. Please refer above diagram in the blog. Generally default pact uri is http://localhost:1234/
        Some of the good references around pact are as below:

        https://docs.pact.io/implementation_guides/python/readme
        https://stackoverflow.com/questions/54805941/actual-interactions-do-not-match-expected-interactions-for-mock-mockservice

        2. In your reply you have in your example: result = requests.post(pact.uri + ‘/1/actions/id/reactions/’)
        But where is the actual POST data being passed in?

        Note that example taken in below neither the exact API under test nor the ideal candidate POST method. I have modified the test to demonstrate some usage of posting data. You may have to look at how to post data based on your API under test.

        “`
        class PostReaction(unittest.TestCase):
        def test_post_reactions(self):
        “””
        Defining test method
        “””
        expected_response = {“status”: 200}

        (pact
        .given(‘Request to post reaction’)
        .upon_receiving(‘a request for response after post reaction’)
        .with_request(method=’POST’,path=’/1/actions/id/reactions/’)
        .will_respond_with(200, body=expected_response))
        with pact:
        result = requests.post(pact.uri +’/1/actions/id/reactions/’,data={“body”:{“status”:200}})

        self.assertEqual(result.json(), expected_response)

        pact.verify()

        “`
        consumer-provider.json will look like as below:

        “`
        {
        “consumer”: {
        “name”: “Consumer”
        },
        “provider”: {
        “name”: “Provider”
        },
        “interactions”: [
        {
        “description”: “a request for response after post reaction”,
        “providerState”: “Request to post reaction”,
        “request”: {
        “method”: “POST”,
        “path”: “/1/actions/id/reactions/”
        },
        “response”: {
        “status”: 200,
        “headers”: {
        },
        “body”: {
        “status”: 200
        }
        }
        }
        ],
        “metadata”: {
        “pactSpecification”: {
        “version”: “2.0.0”
        }
        }

        “`
        And is that POST data supposed to be the consumer-provider.json, or is the consumer-provider.json the result of running the test?
        consumer-provider.json is the result of running test.

        Regards,
        Rahul

  2. Hi Rahul,
    So I just realized the disconnect here is I’m trying to hit an endpoint off “https://xyz.com”, and finally realized that the “pact.uri” you’re referring to defaults to “http://localhost:1234”.

    Is there anyway to SET the “hostname” to refer to an “https://xyz.com” and not have the port set at all or MUST the service being tested reside on some localhost/server?

    Thanks,
    Rahul

    Reply

    1. Hey, you may try to set URL as provider_base_url and try out the test but that will defeat the purpose of using pact . We are using pact as a mock server so, we don’t have to actually communicate the application. Without actually communicating the application, pact helps consumer and provider to validate contracts.

      Reply

    2. Hi Rahul,
      I’ve made some progress on this but still questions:
      1. I start up the localhost on port 1234, but when I go to http://localhost:1234, I keep seeing:
      {“message”:”No interaction found for GET /”,”interaction_diffs”:[]}
      2. We actually already have our service running on “https”, and I set the “ssl=True” flag but it still complains in the setup that it can’t find the “interactions”. So do we NEED the mock service for Pact to work, and it’s simply not possible to run off our “https”, and if so must the endpoint that we already have be replicated on the localhost?

      Thanks for your time!

      Reply

      1. Hi,
        I am not exactly sure the reasons for your error, but here are some suggestions:
        Point 1 – Can you verify if you have setup your interactions correctly in the pact for your test-interactions? For instance, you can set it up in the following way:
        (pact
        .given(‘Request to post reaction’)
        .upon_receiving(‘a request for response after post reaction’)
        .with_request(‘POST’,’/1/actions/id/reactions/’)
        .will_respond_with(200, body=expected_response))
        You can see exact error in the pact-mock-service.log file. Also, one more reason could be if the expected_response is not correctly specified for this particular end point.

        Point 2 – Pact probably is not a good option. Can you try out requests module or any native testclient like fastapi. Testclient can be used for testing application built using FastAPI, think most of these test clients use requests module, so you may want to try with that. We have a blog on how to use fastapi.testclient – https://qxf2.com/blog/testing-fastapi-endpoints-using-fastapi-testclient/

  3. Hi Rahul,
    Please help if possible.
    I’m facing issue when pact.start_service() executes, getting following error:

    requests.exceptions.ConnectionError: HTTPConnectionPool(host=’localhost’, port=1234): Max retries exceeded with url: / (Caused by NewConnectionError(‘: Failed to establish a new connection: [Errno 111] Connection refused’))

    I’m trying to run it inside docker container from here: https://github.com/pact-foundation/pact-python/tree/master/docker
    I just have following code in my python file:
    pact = Consumer(‘Consumer’).has_pact_with(Provider(‘Provider’))
    print(‘start service’)
    pact.start_service()
    atexit.register(pact.stop_service)

    Thanks!

    Reply

Leave a Reply

Your email address will not be published. Required fields are marked *