CODEWITHSUNDEEP
pythonapache-kafkadatacontractpact
Zeruno
Zeruno

Reputation: 1639

Contract testing with Kafka in Python environment?

I am working with multiple applications that communicate asynchronously using Kafka. These applications are managed by several departments and contract testing is appropriate to ensure that the messages used during communication follow the expected schema and will evolve according to the contract specification.

It sounded like the pact library for python is a good fit because it helps creating contract tests for HTTP and message integrations.

What I wanted to do is to send an HTTP request and to listen from the appropriate and dedicated Kafka topic immediately after. But it seems that the test is forcing me specify an HTTP code even if what I am expecting is a message from a queue without an HTTP status code. Furthermore, it seems that the HTTP request is being sent before the consumer is listening. Here is some sample code.

from pact.consumer import Consumer as p_Consumer
from pact.provider import Provider as p_Provider
from confluent_kafka import Consumer as k_Consumer


pact = p_Consumer('Consumer').has_pact_with(p_Provider('Provider'))
pact.start_service()
atexit.register(pact.stop_service)

config = {'bootstrap.servers':'server', 'group.id':0, 'auto.offset.reset':'latest'}
consumer = k_consumer(config)
consumer.subscribe(['usertopic'])

def user():   
  while True:
    msg = consumer.poll(timeout=1)
    if msg is None:
      continue
    else:
      return msg.value().decode()

class ConstractTesting(unittest.TestCase):
  expected = {
  'username': 'UserA',
  'id':123,
  'groups':['Editors']
  }  

  pact.given('UserA exists and is not an administrator')
  .upon_receiving('a request for UserA')
  .with_request(method='GET',path='/user/')
  .will_respond_with(200, body=expected)
  
  with pact:
    result = user()
  
  self.assertEqual(result,expected)

How would I carry out contract testing in Python using Kafka? It feels like I am going through a lot of hoops to carry out this test.

Upvotes: 2

Views: 837

Answers (1)

Matthew Fellows
Matthew Fellows

Reputation: 4065

With Pact message it's a different API you write tests against. You don't use the standard HTTP one, in fact the transport itself is ignored altogether and it's just the payload - the message - we're interested in capturing and verifying. This allows us to test any queue without having to build specific interfaces for each

See this example: https://github.com/pact-foundation/pact-python/blob/02643d4fb89ff7baad63e6436f6a929256c6bf12/examples/message/tests/consumer/test_message_consumer.py#L65

You can read more about message pact testing here: https://docs.pact.io/getting_started/how_pact_works#non-http-testing-message-pact

And finally here are some Kafka examples for other languages that may be helpful: https://docs.pactflow.io/docs/examples/kafka/js/consumer

Upvotes: 3

Related Questions