Added missing _send_upstream mock patch and converted the if statements to asserts

Author: rogaha

kafka/producer/base.py

@@ -121,15 +121,9 @@ def __init__(self, client, async=False,
 
         if batch_send:
             async = True
-            if batch_send_every_n <= 0:
-                log.exception('Batch send message count lower than zero.')
-                raise ValueError
-            if batch_send_every_t <= 0:
-                log.exception('Batch send message interval lower than zero.')
-                raise ValueError
-            if maxsize < 0:
-                log.exception('Queue size upper bound lower than zero.')
-                raise ValueError
+            assert batch_send_every_n > 0
+            assert batch_send_every_t > 0
+            assert maxsize >= 0
         else:
             batch_send_every_n = 1
             batch_send_every_t = 3600

test/test_producer.py

@@ -61,7 +61,8 @@ def test_producer_async_queue_overfilled_batch_send(self, mock):
         for _ in xrange(producer.queue.qsize()):
             producer.queue.get()
 
-    def test_producer_async_queue_overfilled(self):
+    @patch('kafka.producer.base._send_upstream')
+    def test_producer_async_queue_overfilled(self, mock):
         queue_size = 2
         producer = Producer(MagicMock(), async=True, maxsize=queue_size)
 
@@ -73,6 +74,8 @@ def test_producer_async_queue_overfilled(self):
             message_list = [message] * (queue_size + 1)
             producer.send_messages(topic, partition, *message_list)
         self.assertEqual(producer.queue.qsize(), queue_size)
+        for _ in xrange(producer.queue.qsize()):
+            producer.queue.get()
 
     def test_producer_async_queue_normal(self):
         queue_size = 4