<s> import sys <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> import unittest2 as unittest <EOL> else : <EOL> import unittest <EOL> import mock <EOL> import tempfile <EOL> import logging <EOL> from kafka import KafkaClient , MultiProcessConsumer <EOL> import beaver <EOL> from beaver . config import BeaverConfig <EOL> from beaver . transports import create_transport <EOL> from beaver . unicode_dammit import unicode_dammit <EOL> from fixtures import Fixture , ZookeeperFixture , KafkaFixture <EOL> try : <EOL> from beaver . transports . kafka_transport import KafkaTransport <EOL> skip = False <EOL> except ImportError , e : <EOL> if e . message == '<STR_LIT>' : <EOL> skip = True <EOL> else : <EOL> raise <EOL> @ unittest . skipIf ( skip , '<STR_LIT>' ) <EOL> class KafkaTests ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . logger = logging . getLogger ( __name__ ) <EOL> empty_conf = tempfile . NamedTemporaryFile ( delete = True ) <EOL> cls . beaver_config = BeaverConfig ( mock . Mock ( config = empty_conf . name ) ) <EOL> output_file = Fixture . download_official_distribution ( ) <EOL> Fixture . extract_distribution ( output_file ) <EOL> cls . zk = ZookeeperFixture . instance ( ) <EOL> cls . server = KafkaFixture . instance ( <NUM_LIT:0> , cls . zk . host , cls . zk . port ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . server . close ( ) <EOL> cls . zk . close ( ) <EOL> def test_builtin_kafka ( cls ) : <EOL> cls . beaver_config . set ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> cls . beaver_config . set ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> cls . beaver_config . set ( '<STR_LIT>' , cls . server . host + "<STR_LIT::>" + str ( cls . server . port ) ) <EOL> transport = create_transport ( cls . beaver_config , logger = cls . logger ) <EOL> cls . assertIsInstance ( transport , beaver . transports . kafka_transport . KafkaTransport ) <EOL> data = { } <EOL> lines = [ ] <EOL> n = <NUM_LIT:100> <EOL> for i in range ( n ) : <EOL> lines . append ( '<STR_LIT>' + str ( i ) + '<STR_LIT:\n>' ) <EOL> new_lines = [ ] <EOL> for line in lines : <EOL> message = unicode_dammit ( line ) <EOL> if len ( message ) == <NUM_LIT:0> : <EOL> continue <EOL> new_lines . append ( message ) <EOL> data [ '<STR_LIT>' ] = new_lines <EOL> data [ '<STR_LIT>' ] = [ ] <EOL> transport . callback ( "<STR_LIT>" , ** data ) <EOL> messages = cls . _consume_messages ( cls . server . host , cls . server . port ) <EOL> cls . assertEqual ( n , messages . __len__ ( ) ) <EOL> for message in messages : <EOL> cls . assertIn ( '<STR_LIT>' , message . message . value ) ; <EOL> print ( message ) <EOL> print ( '<STR_LIT:\n>' ) <EOL> transport . interrupt ( ) <EOL> def _consume_messages ( cls , host , port ) : <EOL> kafka = KafkaClient ( cls . server . host + "<STR_LIT::>" + str ( cls . server . port ) ) <EOL> consumer = MultiProcessConsumer ( kafka , None , cls . beaver_config . get ( '<STR_LIT>' ) , num_procs = <NUM_LIT:5> ) <EOL> return consumer . get_messages ( count = <NUM_LIT:100> , block = True , timeout = <NUM_LIT:5> ) </s>
