diff options
| author | Dana Powers <dana.powers@rd.io> | 2014-09-17 13:53:04 -0700 |
|---|---|---|
| committer | Dana Powers <dana.powers@rd.io> | 2014-12-15 12:43:45 -0800 |
| commit | 1a06f79cb8ca0d68f06e517e2aad7f0b30c0278e (patch) | |
| tree | 341d850d0a1d8bd31933e22b9d1129a003c6ceea /test/test_consumer_integration.py | |
| parent | 9e1b20306919eaa14774befead190ff52df19ba4 (diff) | |
| download | kafka-python-1a06f79cb8ca0d68f06e517e2aad7f0b30c0278e.tar.gz | |
Make TIMEOUT_MS configurable in test_kafka_consumer__blocking
Diffstat (limited to 'test/test_consumer_integration.py')
| -rw-r--r-- | test/test_consumer_integration.py | 9 |
1 files changed, 5 insertions, 4 deletions
diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py index 007f788..ea32318 100644 --- a/test/test_consumer_integration.py +++ b/test/test_consumer_integration.py @@ -326,14 +326,15 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): @kafka_versions("all") def test_kafka_consumer__blocking(self): + TIMEOUT_MS = 500 consumer = self.kafka_consumer(auto_offset_reset='smallest', - consumer_timeout_ms=1000) + consumer_timeout_ms=TIMEOUT_MS) # Ask for 5 messages, nothing in queue, block 5 seconds with Timer() as t: with self.assertRaises(ConsumerTimeout): msg = consumer.next() - self.assertGreaterEqual(t.interval, 1) + self.assertGreaterEqual(t.interval, TIMEOUT_MS / 1000.0 ) self.send_messages(0, range(0, 10)) @@ -344,7 +345,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): msg = consumer.next() messages.add((msg.partition, msg.offset)) self.assertEqual(len(messages), 5) - self.assertLess(t.interval, 1) + self.assertLess(t.interval, TIMEOUT_MS / 1000.0 ) # Ask for 10 messages, get 5 back, block 5 seconds messages = set() @@ -354,7 +355,7 @@ class TestConsumerIntegration(KafkaIntegrationTestCase): msg = consumer.next() messages.add((msg.partition, msg.offset)) self.assertEqual(len(messages), 5) - self.assertGreaterEqual(t.interval, 1) + self.assertGreaterEqual(t.interval, TIMEOUT_MS / 1000.0 ) @kafka_versions("0.8.1", "0.8.1.1") def test_kafka_consumer__offset_commit_resume(self): |