
from mock import MagicMock, patch
from . import unittest

from kafka import SimpleConsumer, KafkaConsumer, MultiProcessConsumer
from kafka.common import KafkaConfigurationError

class TestKafkaConsumer(unittest.TestCase):
    def test_non_integer_partitions(self):
        with self.assertRaises(AssertionError):
            SimpleConsumer(MagicMock(), 'group', 'topic', partitions = [ '0' ])

    def test_broker_list_required(self):
        with self.assertRaises(KafkaConfigurationError):
            KafkaConsumer()

class TestMultiProcessConsumer(unittest.TestCase):
    def test_partition_list(self):
        client = MagicMock()
        partitions = (0,)
        with patch.object(MultiProcessConsumer, 'fetch_last_known_offsets') as fetch_last_known_offsets:
            consumer = MultiProcessConsumer(client, 'testing-group', 'testing-topic', partitions=partitions)
            self.assertEqual(fetch_last_known_offsets.call_args[0], (partitions,) )
        self.assertEqual(client.get_partition_ids_for_topic.call_count, 0) # pylint: disable=no-member

@@ -147,16 +147,22 @@
  import 
+(%0A    
 KafkaCon
@@ -176,16 +176,139 @@
 ionError
+, FetchResponse,%0A    FailedPayloadsError, OffsetAndMessage,%0A    NotLeaderForPartitionError, UnknownTopicOrPartitionError%0A)%0A
 %0A%0Aclass 
@@ -644,16 +644,17 @@
 umer()%0A%0A
+%0A
 class Te
@@ -1084,16 +1084,16 @@
 ons,) )%0A
-
         
@@ -1187,8 +1187,2848 @@
 -member%0A
+%0A    def test_simple_consumer_failed_payloads(self):%0A        client = MagicMock()%0A        consumer = SimpleConsumer(client, group=None,%0A                                  topic='topic', partitions=%5B0, 1%5D,%0A                                  auto_commit=False)%0A%0A        def failed_payloads(payload):%0A            return FailedPayloadsError(payload)%0A%0A        client.send_fetch_request.side_effect = self.fail_requests_factory(failed_payloads)%0A%0A        # This should not raise an exception%0A        consumer.get_messages(5)%0A%0A    def test_simple_consumer_leader_change(self):%0A        client = MagicMock()%0A        consumer = SimpleConsumer(client, group=None,%0A                                  topic='topic', partitions=%5B0, 1%5D,%0A                                  auto_commit=False)%0A%0A        # Mock so that only the first request gets a valid response%0A        def not_leader(request):%0A            return FetchResponse(request.topic, request.partition,%0A                                 NotLeaderForPartitionError.errno, -1, ())%0A%0A        client.send_fetch_request.side_effect = self.fail_requests_factory(not_leader)%0A%0A        # This should not raise an exception%0A        consumer.get_messages(20)%0A%0A        # client should have updated metadata%0A        self.assertGreaterEqual(client.reset_topic_metadata.call_count, 1)%0A        self.assertGreaterEqual(client.load_metadata_for_topics.call_count, 1)%0A%0A    def test_simple_consumer_unknown_topic_partition(self):%0A        client = MagicMock()%0A        consumer = SimpleConsumer(client, group=None,%0A                                  topic='topic', partitions=%5B0, 1%5D,%0A                                  auto_commit=False)%0A%0A        # Mock so that only the first request gets a valid response%0A        def unknown_topic_partition(request):%0A            return FetchResponse(request.topic, request.partition,%0A                                 UnknownTopicOrPartitionError.errno, -1, ())%0A%0A        client.send_fetch_request.side_effect = self.fail_requests_factory(unknown_topic_partition)%0A%0A        # This should not raise an exception%0A        with self.assertRaises(UnknownTopicOrPartitionError):%0A            consumer.get_messages(20)%0A%0A    @staticmethod%0A    def fail_requests_factory(error_factory):%0A        # Mock so that only the first request gets a valid response%0A        def fail_requests(payloads, **kwargs):%0A            responses = %5B%0A                FetchResponse(payloads%5B0%5D.topic, payloads%5B0%5D.partition, 0, 0,%0A                              (OffsetAndMessage(%0A                                  payloads%5B0%5D.offset + i,%0A                                  %22msg %25d%22 %25 (payloads%5B0%5D.offset + i))%0A                               for i in range(10))),%0A            %5D%0A            for failure in payloads%5B1:%5D:%0A                responses.append(error_factory(failure))%0A            return responses%0A        return fail_requests%0A

Add unit tests for SimpleConsumer error handling