docs(examples): wip on a kafka example

Author: elliottmurray

examples/kafka/pacts/animalconsumer_message-dogproducer_message.json

@@ -0,0 +1,32 @@
+{
+  "consumer": {
+    "name": "AnimalConsumer_message"
+  },
+  "provider": {
+    "name": "DogProducer_message"
+  },
+  "messages": [
+    {
+      "description": "Spot the poodle",
+      "providerStates": [
+        {
+          "name": "A dog created",
+          "params": null
+        }
+      ],
+      "contents": "{\"type\": \"dog\", \"name\": \"spot\", \"breed\": \"poodle\"}",
+      "matchingRules": {
+        "body": {
+        }
+      },
+      "metaData": {
+        "Content-Type": "application/json"
+      }
+    }
+  ],
+  "metadata": {
+    "pactSpecification": {
+      "version": "3.0.0"
+    }
+  }
+}

examples/kafka/requirements.txt

@@ -8,3 +8,4 @@ psutil==5.8.0
 requests==2.25.1
 six==1.15.0
 urllib3==1.26.4
+pytest==5.4.1

examples/kafka/src/__init__.py

@@ -18,14 +18,31 @@ def __init__(self, type, name, breed):
     def __str__(self):
         return f'Animal {self.type}, Name {self.name}, Breed {self.breed}'
 
-def object_decoder(obj):
-    if '__type__' in obj and obj['__type__'] == 'dog':
+class CustomError(Exception):
+    def __init__(self, *args):
+        if args:
+            self.topic = args[0]
+        else:
+            self.topic = None
+
+    def __str__(self):
+        if self.topic:
+            return 'Custom Error:, {0}'.format(self.topic)
+
+
+def message_decoder(obj):
+    if 'type' in obj and obj['type'] == 'dog':
         return Animal('dog', obj['name'], obj['breed'])
-    return obj
+    raise CustomError('type unknown')
+
+
+def send_dog_event_foo(foo, payload):
+    # ignoring foo here
+    return send_dog_event(payload)
 
 
 def send_dog_event(payload):
-    dog = json.loads(payload, object_hook=object_decoder)
+    dog = json.loads(payload, object_hook=message_decoder)
     print(dog)
     return dog
 

examples/kafka/src/kafka_consumer.py

@@ -6,6 +6,8 @@
 NAMES = ['Spot', 'Wolf', 'Winston', 'Tobey']
 BREEDS = ['poodle', 'bulldog', 'Great Dane', 'Greyhound']
 
+PRODUCER = Producer({'bootstrap.servers': 'localhost:9092'})
+
 def acked(err, msg):
     if err is not None:
         print("Failed to deliver dog: {0}: {1}"
@@ -24,20 +26,19 @@ def __init__(self):
     def __str__(self):
         return f'Name {self.name}, Breed {self.breed}'
 
-def send_message(p, dog):
+def send_message(dog):
     print(dog)
 
-    p.produce('mytopic', json.dumps(dog.__dict__), callback=acked)
-    p.poll(0.5)
+    PRODUCER.produce('mytopic', json.dumps(dog.__dict__), callback=acked)
+    PRODUCER.poll(0.5)
 
 def main():
     number_of_dogs = int(sys.argv[1])
-    p = Producer({'bootstrap.servers': 'localhost:9092'})
 
     try:
         for i in range(0, number_of_dogs):
-            send_message(p, Dog())
-        p.flush(30)
+            send_message(Dog())
+        PRODUCER.flush(30)
 
     except KeyboardInterrupt:
         pass

examples/kafka/src/kafka_producer.py

@@ -0,0 +1,6 @@
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--publish-pact", type=str, action="store",
+        help="Upload generated pact file to pact broker with version"
+    )

examples/kafka/tests/__init__.py

@@ -0,0 +1,97 @@
+"""pact test for a message consumer"""
+
+import logging
+import pytest
+import json
+
+from pact import MessageConsumer, Provider
+from src.kafka_consumer import send_dog_event, send_dog_event_foo, CustomError
+
+log = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+PACT_BROKER_URL = "http://localhost"
+PACT_BROKER_USERNAME = "pactbroker"
+PACT_BROKER_PASSWORD = "pactbroker"
+PACT_DIR = 'pacts'
+
+CONSUMER_NAME = 'AnimalConsumer'
+PROVIDER_NAME = 'DogProducer'
+
+PACT_FILE = (f"{PACT_DIR}/{CONSUMER_NAME.lower().replace(' ', '_')}_message-"
+             + f"{PROVIDER_NAME.lower().replace(' ', '_')}_message.json")
+
+EXPECTED_DOG = {'type': 'dog', 'name': 'spot', 'breed': 'poodle'}
+
+@pytest.fixture(scope='session')
+def pact(request):
+    version = request.config.getoption('--publish-pact')
+    publish = True if version else False
+
+    pact = MessageConsumer(CONSUMER_NAME, version=version).has_pact_with(
+        Provider(PROVIDER_NAME),
+        publish_to_broker=publish, broker_base_url=PACT_BROKER_URL,
+        broker_username=PACT_BROKER_USERNAME, broker_password=PACT_BROKER_PASSWORD, pact_dir=PACT_DIR)
+
+    yield pact
+
+
+def test_assert_verify_message(pact):
+
+    (pact
+     .given('A dog created')
+     .expects_to_receive('Spot the poodle')
+     .with_content(str(json.dumps(EXPECTED_DOG)))
+     .with_metadata({
+         'Content-Type': 'application/json'
+     })
+     .verify(send_dog_event)
+     )
+
+
+def test_throw_exception_handler(pact):
+
+    with pytest.raises(CustomError):
+        (pact
+            .given('A dog created')
+            .expects_to_receive('Spot the poodle')
+            .with_content(str(json.dumps({'type': 'cat', 'name': 'spot', 'breed': 'poodle'})))
+            .with_metadata({
+                'Content-Type': 'application/json'
+            })
+            .verify(send_dog_event))
+
+
+def test_assert_dog_returned(pact):
+
+    (pact
+     .given('A dog created')
+     .expects_to_receive('Spot the poodle')
+     .with_content(str(json.dumps(EXPECTED_DOG)))
+     .with_metadata({
+         'Content-Type': 'application/json'
+     })
+     .verify(send_dog_event)
+     )
+
+    with pact:
+        dog = pact.result
+        assert dog.name == 'spot'
+        assert dog.breed == 'poodle'
+
+
+def test_assert_calling_dog(pact):
+
+    (pact
+     .given('A dog created')
+     .expects_to_receive('Spot the poodle')
+     .with_content(str(json.dumps(EXPECTED_DOG)))
+     .with_metadata({
+         'Content-Type': 'application/json'
+     })
+     )
+
+    with pact:
+        dog = send_dog_event_foo({}, str(json.dumps(EXPECTED_DOG)))
+        assert dog.name == 'spot'
+        assert dog.breed == 'poodle'

examples/kafka/tests/conftest.py

@@ -171,6 +171,10 @@ def write_to_pact_file(self):
 
         self._message_process = Popen(command)
 
+    def verify(self, func_to_call):
+        """Verify our target method with our content."""
+        self.result = func_to_call(self._messages[0]['contents'])
+
     def _insert_message_if_complete(self):
         """
         Insert a new message if current message is complete.
@@ -185,8 +189,13 @@ def _insert_message_if_complete(self):
             self._messages.insert(0, {})
 
     def __enter__(self):
-        """Enter a Python context. This function is required for context manager to work."""
-        pass
+        """
+        Enter a Python context. This function is required for context manager to work.
+
+        Calls our handler function with our payload
+        """
+        if hasattr(self, 'func_to_call') and callable(self.func_to_call):
+            self.result = self.func_to_call(self._messages[0]['contents'])
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         """