<s> from contextlib import contextmanager <EOL> import unittest2 <EOL> from pykafka . exceptions import RdKafkaStoppedException , RdKafkaException <EOL> from pykafka . rdkafka import _rd_kafka <EOL> from pykafka . test . utils import get_cluster , stop_cluster <EOL> from pykafka . utils . compat import get_bytes <EOL> class TestRdKafkaConsumer ( unittest2 . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . kafka = get_cluster ( ) <EOL> cls . topic_name = b'<STR_LIT>' <EOL> cls . n_partitions = <NUM_LIT:3> <EOL> cls . kafka . create_topic ( cls . topic_name , cls . n_partitions , <NUM_LIT:2> ) <EOL> cls . partition_ids = list ( range ( cls . n_partitions ) ) <EOL> cls . start_offsets = cls . n_partitions * [ <NUM_LIT:0> ] <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> stop_cluster ( cls . kafka ) <EOL> @ contextmanager <EOL> def assert_thread_cnt_non_increasing ( self ) : <EOL> """<STR_LIT>""" <EOL> preexisting_threads = _rd_kafka . _thread_cnt ( ) <EOL> yield <EOL> try : <EOL> _rd_kafka . _wait_destroyed ( <NUM_LIT:1000> ) <EOL> except : <EOL> pass <EOL> self . assertLessEqual ( _rd_kafka . _thread_cnt ( ) , preexisting_threads ) <EOL> def test_start_fail ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assert_thread_cnt_non_increasing ( ) : <EOL> consumer = _rd_kafka . Consumer ( ) <EOL> with self . assertRaises ( RdKafkaException ) : <EOL> consumer . start ( brokers = b"<STR_LIT>" , <EOL> topic_name = self . topic_name , <EOL> partition_ids = self . partition_ids , <EOL> start_offsets = self . start_offsets ) <EOL> def test_stop ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assert_thread_cnt_non_increasing ( ) : <EOL> consumer = _rd_kafka . Consumer ( ) <EOL> consumer . configure ( conf = [ ] ) <EOL> consumer . configure ( topic_conf = [ ] ) <EOL> consumer . start ( brokers = get_bytes ( self . kafka . brokers ) , <EOL> topic_name = self . topic_name , <EOL> partition_ids = self . partition_ids , <EOL> start_offsets = self . start_offsets ) <EOL> consumer . consume ( <NUM_LIT:100> ) <EOL> consumer . stop ( ) <EOL> def test_stopped_exception ( self ) : <EOL> """<STR_LIT>""" <EOL> consumer = _rd_kafka . Consumer ( brokers = get_bytes ( self . kafka . brokers ) , <EOL> topic_name = self . topic_name , <EOL> partition_ids = self . partition_ids , <EOL> start_offsets = self . start_offsets ) <EOL> consumer . stop ( ) <EOL> with self . assertRaises ( RdKafkaStoppedException ) : <EOL> consumer . consume ( <NUM_LIT:1> ) </s>
