Kalimaha

class GrillChefPactHelper(PactHelper):
    test_port = 8080
    process = None

    def setup(self):
        cmd = 'gunicorn start:app -w 3 -b :8080 --log-level error'
        self.process = subprocess.Popen(cmd, shell=True)

    def tear_down(self):
        self.process.kill()

Kalimaha

@honours_pact_with('Sous Chef')
@pact_uri('http://localhost:9292/pacts/provider/Grill%20Chef/consumer/Sous%20Chef/latest.json')
class SousChef(ServiceConsumerTest):

    @state('steaks are available')
    def test_steak(self):
        DB.save(Steak(cooked=False))

Kalimaha

@given('steaks are available')
@upon_receiving('a request for a well-done steak')
@with_request({'method': 'POST', 'path': '/steaks/', 'body': {'cooking': 'well-done'}})
@will_respond_with({'status': 422, 'body': {'reason': 'we do not serve well-done steaks'}})
def test_steak(self):
    response = steak('well-done')
    assert response.status_code == 422

Kalimaha

@service_consumer('Sous Chef')
@has_pact_with('Grill Chef')
class GrillChefTest(ServiceProviderTest):
    pass

Kalimaha

@service_consumer('Sous Chef')
@has_pact_with('Grill Chef')
class GrillChefTest(ServiceProviderTest):

    @given('steaks are available')
    @upon_receiving('a request for a steak')
    @with_request({'method': 'POST', 'path': '/steaks/', 'body': {'cooking': 'medium-rare'}})
    @will_respond_with({'status': 201, 'body': {'waitingTime': 10}})
    def test_steak(self):
        response = GrillChefRepository.steak('medium-rare')

Kalimaha

{
  "provider": { "name": "Grill Chef" },
  "consumer": { "name": "Sous Chef" },
  "interactions": [
    {
      "status": "PASSED",
      "providerState": "steaks are available",
      "description": "a request for a steak",
      "request": {
        "body": { "cooking": "medium-rare" },

Kalimaha

headers = [{'Content-Type': 'application/json'}, {'Accept', '*/*'}]

for key, value in sum(list(map(list, map(dict.items, headers))), []):
    print(str(key) + ': ' + str(value))

Kalimaha

import time
from threading import Thread
try:
    import socketserver as SocketServer
    import http.server as SimpleHTTPServer
except ImportError:
    import SocketServer
    import SimpleHTTPServer

Kalimaha

print('\033[95m' + '[ ] - ' + s.state + '\033[0m')
print('\033[92m' + '[✓] - ' + s.state + '\033[0m')
print('\033[93m' + '[!] - ' + s.state + '\033[0m')
print('\033[91m' + '=======================\033[0m')
print('\033[91m' + '[x] - ' + s.state + '\033[0m')
print('\033[91m' + '-----------------------\033[0m')
print('\033[91m' + 'Expected:\n\n  ' + str({'spam': 'eggs'}) + '\n\nActual:\n\n  ' + str({'eggs': 'bacon'}) + '\033[0m')
print('\033[91m' + '=======================\033[0m')

Kalimaha

def state(state_value):
    def decorate(function):
        function.state = state_value
        return function
    return decorate


class ConsumerTest(object):
    @property
    def states(self):