osm_mon/test/core/kafka_test.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 # Copyright 2017 Intel Research and Development Ireland Limited
   5 # *************************************************************
   6
   7 # This file is part of OSM Monitoring module
   8 # All Rights Reserved to Intel Corporation
   9
  10 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  11 # not use this file except in compliance with the License. You may obtain
  12 # a copy of the License at
  13
  14 #         http://www.apache.org/licenses/LICENSE-2.0
  15
  16 # Unless required by applicable law or agreed to in writing, software
  17 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  18 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  19 # License for the specific language governing permissions and limitations
  20 # under the License.
  21
  22 # For those usages not covered by the Apache License, Version 2.0 please
  23 # contact: prithiv.mohan@intel.com or adrian.hoban@intel.com
  24
  25 #__author__ = "Prithiv Mohan"
  26 #__date__   = "25/Sep/2017"
  27
  28 import sys
  29 import threading
  30 import pytest
  31 from kafka import KafkaConsumer, KafkaProducer
  32
  33 def test_end_to_end():
  34     producer = KafkaProducer(bootstrap_servers='localhost:9092',
  35                              retries=5,
  36                              max_block_ms=10000,
  37                              value_serializer=str.encode)
  38     consumer = KafkaConsumer(bootstrap_servers='localhost:9092',
  39                              group_id=None,
  40                              consumer_timeout_ms=10000,
  41                              auto_offset_reset='earliest',
  42                              value_deserializer=bytes.decode)
  43
  44     topic = 'TutorialTopic'
  45
  46     messages = 100
  47     futures = []
  48     for i in range(messages):
  49         futures.append(producer.send(topic, 'msg %d' % i))
  50     ret = [f.get(timeout=30) for f in futures]
  51     assert len(ret) == messages
  52
  53     producer.close()
  54
  55     consumer.subscribe([topic])
  56     msgs = set()
  57     for i in range(messages):
  58         try:
  59             msgs.add(next(consumer).value)
  60         except StopIteration:
  61             break
  62
  63     assert msgs == set(['msg %d' % i for i in range(messages)])