Merge pull request #397 from dpkp/test_logging

Test logging

Author: dpkp

test/service.py

@@ -59,7 +59,8 @@ def _spawn(self):
         self.alive = True
 
     def _despawn(self):
-        self.child.terminate()
+        if self.child.poll() is None:
+            self.child.terminate()
         self.alive = False
         for _ in range(50):
             if self.child.poll() is not None:

test/test_conn.py

@@ -1,3 +1,4 @@
+import logging
 import socket
 import struct
 from threading import Thread
@@ -10,6 +11,10 @@
 
 class ConnTest(unittest.TestCase):
     def setUp(self):
+
+        # kafka.conn debug logging is verbose, so only enable in conn tests
+        logging.getLogger('kafka.conn').setLevel(logging.DEBUG)
+
         self.config = {
             'host': 'localhost',
             'port': 9090,

test/test_consumer_integration.py

@@ -170,11 +170,11 @@ def test_simple_consumer__seek(self):
     def test_simple_consumer_blocking(self):
         consumer = self.consumer()
 
-        # Ask for 5 messages, nothing in queue, block 5 seconds
+        # Ask for 5 messages, nothing in queue, block 1 second
         with Timer() as t:
-            messages = consumer.get_messages(block=True, timeout=5)
+            messages = consumer.get_messages(block=True, timeout=1)
             self.assert_message_count(messages, 0)
-        self.assertGreaterEqual(t.interval, 5)
+        self.assertGreaterEqual(t.interval, 1)
 
         self.send_messages(0, range(0, 10))
 
@@ -184,11 +184,11 @@ def test_simple_consumer_blocking(self):
             self.assert_message_count(messages, 5)
         self.assertLessEqual(t.interval, 1)
 
-        # Ask for 10 messages, get 5 back, block 5 seconds
+        # Ask for 10 messages, get 5 back, block 1 second
         with Timer() as t:
-            messages = consumer.get_messages(count=10, block=True, timeout=5)
+            messages = consumer.get_messages(count=10, block=True, timeout=1)
             self.assert_message_count(messages, 5)
-        self.assertGreaterEqual(t.interval, 5)
+        self.assertGreaterEqual(t.interval, 1)
 
         consumer.stop()
 
@@ -236,12 +236,12 @@ def test_multi_process_consumer(self):
     def test_multi_process_consumer_blocking(self):
         consumer = self.consumer(consumer = MultiProcessConsumer)
 
-        # Ask for 5 messages, No messages in queue, block 5 seconds
+        # Ask for 5 messages, No messages in queue, block 1 second
         with Timer() as t:
-            messages = consumer.get_messages(block=True, timeout=5)
+            messages = consumer.get_messages(block=True, timeout=1)
             self.assert_message_count(messages, 0)
 
-        self.assertGreaterEqual(t.interval, 5)
+        self.assertGreaterEqual(t.interval, 1)
 
         # Send 10 messages
         self.send_messages(0, range(0, 10))
@@ -252,11 +252,11 @@ def test_multi_process_consumer_blocking(self):
             self.assert_message_count(messages, 5)
         self.assertLessEqual(t.interval, 1)
 
-        # Ask for 10 messages, 5 in queue, block 5 seconds
+        # Ask for 10 messages, 5 in queue, block 1 second
         with Timer() as t:
-            messages = consumer.get_messages(count=10, block=True, timeout=5)
+            messages = consumer.get_messages(count=10, block=True, timeout=1)
             self.assert_message_count(messages, 5)
-        self.assertGreaterEqual(t.interval, 4.95)
+        self.assertGreaterEqual(t.interval, 1)
 
         consumer.stop()
 
@@ -450,7 +450,7 @@ def test_kafka_consumer__blocking(self):
         consumer = self.kafka_consumer(auto_offset_reset='smallest',
                                        consumer_timeout_ms=TIMEOUT_MS)
 
-        # Ask for 5 messages, nothing in queue, block 5 seconds
+        # Ask for 5 messages, nothing in queue, block 500ms
         with Timer() as t:
             with self.assertRaises(ConsumerTimeout):
                 msg = consumer.next()
@@ -467,7 +467,7 @@ def test_kafka_consumer__blocking(self):
         self.assertEqual(len(messages), 5)
         self.assertLess(t.interval, TIMEOUT_MS / 1000.0 )
 
-        # Ask for 10 messages, get 5 back, block 5 seconds
+        # Ask for 10 messages, get 5 back, block 500ms
         messages = set()
         with Timer() as t:
             with self.assertRaises(ConsumerTimeout):

test/test_failover_integration.py

@@ -98,10 +98,14 @@ def test_switch_leader_async(self):
 
         # Test the base class Producer -- send_messages to a specific partition
         producer = Producer(self.client, async=True,
-                            req_acks=Producer.ACK_AFTER_CLUSTER_COMMIT)
+                            batch_send_every_n=15,
+                            batch_send_every_t=3,
+                            req_acks=Producer.ACK_AFTER_CLUSTER_COMMIT,
+                            async_log_messages_on_error=False)
 
         # Send 10 random messages
         self._send_random_messages(producer, topic, partition, 10)
+        self._send_random_messages(producer, topic, partition + 1, 10)
 
         # kill leader for partition
         self._kill_leader(topic, partition)
@@ -110,9 +114,11 @@ def test_switch_leader_async(self):
 
         # in async mode, this should return immediately
         producer.send_messages(topic, partition, b'success')
+        producer.send_messages(topic, partition + 1, b'success')
 
         # send to new leader
         self._send_random_messages(producer, topic, partition, 10)
+        self._send_random_messages(producer, topic, partition + 1, 10)
 
         # Stop the producer and wait for it to shutdown
         producer.stop()
@@ -129,6 +135,8 @@ def test_switch_leader_async(self):
         # Should be equal to 10 before + 1 recovery + 10 after
         self.assert_message_count(topic, 21, partitions=(partition,),
                                   at_least=True)
+        self.assert_message_count(topic, 21, partitions=(partition + 1,),
+                                  at_least=True)
 
     @kafka_versions("all")
     def test_switch_leader_keyed_producer(self):

test/testutil.py

@@ -113,3 +113,8 @@ def __exit__(self, *args):
         self.interval = self.end - self.start
 
 logging.basicConfig(level=logging.DEBUG)
+logging.getLogger('test.fixtures').setLevel(logging.ERROR)
+logging.getLogger('test.service').setLevel(logging.ERROR)
+
+# kafka.conn debug logging is verbose, disable in tests by default
+logging.getLogger('kafka.conn').setLevel(logging.INFO)

tox.ini

@@ -11,7 +11,7 @@ deps =
     mock
     python-snappy
 commands =
-    nosetests {posargs:-v --with-id --id-file={envdir}/.noseids --with-timer --timer-top-n 10 --with-coverage --cover-erase --cover-package kafka}
+    nosetests {posargs:-v -x --with-id --id-file={envdir}/.noseids --with-timer --timer-top-n 10 --with-coverage --cover-erase --cover-package kafka}
 setenv =
     PROJECT_ROOT = {toxinidir}
 passenv = KAFKA_VERSION