summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/test_client_integration.py12
-rw-r--r--test/test_codec.py5
-rw-r--r--test/test_conn.py4
-rw-r--r--test/test_consumer.py16
-rw-r--r--test/test_consumer_integration.py16
-rw-r--r--test/test_failover_integration.py53
-rw-r--r--test/test_producer.py8
-rw-r--r--test/test_producer_integration.py24
-rw-r--r--test/test_protocol.py25
-rw-r--r--test/test_util.py2
10 files changed, 79 insertions, 86 deletions
diff --git a/test/test_client_integration.py b/test/test_client_integration.py
index b33d17c..b433146 100644
--- a/test/test_client_integration.py
+++ b/test/test_client_integration.py
@@ -2,10 +2,16 @@ import os
import socket
import unittest2
-import kafka
-from kafka.common import *
+from kafka.conn import KafkaConnection
+from kafka.common import (
+ FetchRequest, OffsetCommitRequest, OffsetFetchRequest,
+ KafkaTimeoutError
+)
+
from test.fixtures import ZookeeperFixture, KafkaFixture
-from test.testutil import *
+from test.testutil import (
+ KafkaIntegrationTestCase, get_open_port, kafka_versions, Timer
+)
class TestKafkaClientIntegration(KafkaIntegrationTestCase):
@classmethod
diff --git a/test/test_codec.py b/test/test_codec.py
index 2e6f67e..0ee7ce0 100644
--- a/test/test_codec.py
+++ b/test/test_codec.py
@@ -5,10 +5,7 @@ from kafka.codec import (
has_snappy, gzip_encode, gzip_decode,
snappy_encode, snappy_decode
)
-from kafka.protocol import (
- create_gzip_message, create_message, create_snappy_message, KafkaProtocol
-)
-from testutil import *
+from testutil import random_string
class TestCodec(unittest2.TestCase):
def test_gzip(self):
diff --git a/test/test_conn.py b/test/test_conn.py
index 5451398..931ace7 100644
--- a/test/test_conn.py
+++ b/test/test_conn.py
@@ -4,8 +4,8 @@ import struct
import mock
import unittest2
-from kafka.common import *
-from kafka.conn import *
+from kafka.common import ConnectionError
+from kafka.conn import KafkaConnection, collect_hosts, DEFAULT_SOCKET_TIMEOUT_SECONDS
class ConnTest(unittest2.TestCase):
def setUp(self):
diff --git a/test/test_consumer.py b/test/test_consumer.py
index 778d76a..f70e292 100644
--- a/test/test_consumer.py
+++ b/test/test_consumer.py
@@ -1,22 +1,10 @@
-import os
-import random
-import struct
import unittest2
-from mock import MagicMock, patch
+from mock import MagicMock
-from kafka import KafkaClient
from kafka.consumer import SimpleConsumer
-from kafka.common import (
- ProduceRequest, BrokerMetadata, PartitionMetadata,
- TopicAndPartition, KafkaUnavailableError,
- LeaderUnavailableError, PartitionUnavailableError
-)
-from kafka.protocol import (
- create_message, KafkaProtocol
-)
class TestKafkaConsumer(unittest2.TestCase):
def test_non_integer_partitions(self):
with self.assertRaises(AssertionError):
- consumer = SimpleConsumer(MagicMock(), 'group', 'topic', partitions = [ '0' ])
+ SimpleConsumer(MagicMock(), 'group', 'topic', partitions = [ '0' ])
diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py
index 44dafe4..6576a32 100644
--- a/test/test_consumer_integration.py
+++ b/test/test_consumer_integration.py
@@ -1,11 +1,13 @@
import os
-from datetime import datetime
-from kafka import * # noqa
-from kafka.common import * # noqa
+from kafka import SimpleConsumer, MultiProcessConsumer, create_message
+from kafka.common import ProduceRequest, ConsumerFetchSizeTooSmall
from kafka.consumer import MAX_FETCH_BUFFER_SIZE_BYTES
-from fixtures import ZookeeperFixture, KafkaFixture
-from testutil import *
+
+from test.fixtures import ZookeeperFixture, KafkaFixture
+from test.testutil import (
+ KafkaIntegrationTestCase, kafka_versions, random_string, Timer
+)
class TestConsumerIntegration(KafkaIntegrationTestCase):
@classmethod
@@ -215,8 +217,8 @@ class TestConsumerIntegration(KafkaIntegrationTestCase):
@kafka_versions("0.8.1", "0.8.1.1")
def test_offset_behavior__resuming_behavior(self):
- msgs1 = self.send_messages(0, range(0, 100))
- msgs2 = self.send_messages(1, range(100, 200))
+ self.send_messages(0, range(0, 100))
+ self.send_messages(1, range(100, 200))
# Start a consumer
consumer1 = self.consumer(
diff --git a/test/test_failover_integration.py b/test/test_failover_integration.py
index 6c0e662..5e737b0 100644
--- a/test/test_failover_integration.py
+++ b/test/test_failover_integration.py
@@ -3,11 +3,14 @@ import os
import time
import unittest2
-from kafka import * # noqa
-from kafka.common import * # noqa
+from kafka import KafkaClient, SimpleConsumer
+from kafka.common import TopicAndPartition, FailedPayloadsError, ConnectionError
from kafka.producer import Producer
-from fixtures import ZookeeperFixture, KafkaFixture
-from testutil import *
+
+from test.fixtures import ZookeeperFixture, KafkaFixture
+from test.testutil import (
+ KafkaIntegrationTestCase, kafka_versions, random_string
+)
class TestFailover(KafkaIntegrationTestCase):
@@ -42,16 +45,18 @@ class TestFailover(KafkaIntegrationTestCase):
@kafka_versions("all")
def test_switch_leader(self):
- key, topic, partition = random_string(5), self.topic, 0
+ topic = self.topic
+ partition = 0
# Test the base class Producer -- send_messages to a specific partition
- producer = Producer(self.client, async=False)
+ producer = Producer(self.client, async=False,
+ req_acks=Producer.ACK_AFTER_CLUSTER_COMMIT)
# Send 10 random messages
- self._send_random_messages(producer, topic, partition, 10)
+ self._send_random_messages(producer, topic, partition, 100)
# kill leader for partition
- broker = self._kill_leader(topic, partition)
+ self._kill_leader(topic, partition)
# expect failure, but dont wait more than 60 secs to recover
recovered = False
@@ -71,20 +76,18 @@ class TestFailover(KafkaIntegrationTestCase):
self.assertTrue(recovered)
# send some more messages to new leader
- self._send_random_messages(producer, topic, partition, 10)
+ self._send_random_messages(producer, topic, partition, 100)
# count number of messages
- count = self._count_messages('test_switch_leader group', topic,
- partitions=(partition,))
-
# Should be equal to 10 before + 1 recovery + 10 after
- self.assertEquals(count, 21)
+ self.assert_message_count(topic, 201, partitions=(partition,))
#@kafka_versions("all")
@unittest2.skip("async producer does not support reliable failover yet")
def test_switch_leader_async(self):
- key, topic, partition = random_string(5), self.topic, 0
+ topic = self.topic
+ partition = 0
# Test the base class Producer -- send_messages to a specific partition
producer = Producer(self.client, async=True)
@@ -93,7 +96,7 @@ class TestFailover(KafkaIntegrationTestCase):
self._send_random_messages(producer, topic, partition, 10)
# kill leader for partition
- broker = self._kill_leader(topic, partition)
+ self._kill_leader(topic, partition)
logging.debug("attempting to send 'success' message after leader killed")
@@ -109,12 +112,8 @@ class TestFailover(KafkaIntegrationTestCase):
producer.stop()
# count number of messages
- count = self._count_messages('test_switch_leader_async group', topic,
- partitions=(partition,))
-
# Should be equal to 10 before + 1 recovery + 10 after
- self.assertEquals(count, 21)
-
+ self.assert_message_count(topic, 21, partitions=(partition,))
def _send_random_messages(self, producer, topic, partition, n):
for j in range(n):
@@ -130,17 +129,25 @@ class TestFailover(KafkaIntegrationTestCase):
broker.close()
return broker
- def _count_messages(self, group, topic, timeout=1, partitions=None):
+ def assert_message_count(self, topic, check_count, timeout=10, partitions=None):
hosts = ','.join(['%s:%d' % (broker.host, broker.port)
for broker in self.brokers])
client = KafkaClient(hosts)
+ group = random_string(10)
consumer = SimpleConsumer(client, group, topic,
partitions=partitions,
auto_commit=False,
iter_timeout=timeout)
- count = consumer.pending(partitions)
+ started_at = time.time()
+ pending = consumer.pending(partitions)
+
+ # Keep checking if it isn't immediately correct, subject to timeout
+ while pending != check_count and (time.time() - started_at < timeout):
+ pending = consumer.pending(partitions)
+
consumer.stop()
client.close()
- return count
+
+ self.assertEqual(pending, check_count)
diff --git a/test/test_producer.py b/test/test_producer.py
index a84e20f..e00f9af 100644
--- a/test/test_producer.py
+++ b/test/test_producer.py
@@ -1,14 +1,10 @@
# -*- coding: utf-8 -*-
import logging
-import os
-import random
-import struct
-import unittest2
-from mock import MagicMock, patch
+import unittest2
+from mock import MagicMock
-from kafka import KafkaClient
from kafka.producer import Producer
class TestKafkaProducer(unittest2.TestCase):
diff --git a/test/test_producer_integration.py b/test/test_producer_integration.py
index 7d3a180..19d3a6d 100644
--- a/test/test_producer_integration.py
+++ b/test/test_producer_integration.py
@@ -2,11 +2,18 @@ import os
import time
import uuid
-from kafka import * # noqa
-from kafka.common import * # noqa
-from kafka.codec import has_gzip, has_snappy
-from fixtures import ZookeeperFixture, KafkaFixture
-from testutil import *
+from kafka import (
+ SimpleProducer, KeyedProducer,
+ create_message, create_gzip_message, create_snappy_message,
+ RoundRobinPartitioner, HashedPartitioner
+)
+from kafka.common import (
+ FetchRequest, ProduceRequest, UnknownTopicOrPartitionError
+)
+from kafka.codec import has_snappy
+
+from test.fixtures import ZookeeperFixture, KafkaFixture
+from test.testutil import KafkaIntegrationTestCase, kafka_versions
class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
topic = 'produce_topic'
@@ -149,7 +156,7 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
# At first it doesn't exist
with self.assertRaises(UnknownTopicOrPartitionError):
- resp = producer.send_messages(new_topic, self.msg("one"))
+ producer.send_messages(new_topic, self.msg("one"))
@kafka_versions("all")
def test_producer_random_order(self):
@@ -219,7 +226,6 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
@kafka_versions("all")
def test_acks_none(self):
start_offset0 = self.current_offset(self.topic, 0)
- start_offset1 = self.current_offset(self.topic, 1)
producer = SimpleProducer(self.client, req_acks=SimpleProducer.ACK_NOT_REQUIRED)
resp = producer.send_messages(self.topic, self.msg("one"))
@@ -231,7 +237,6 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
@kafka_versions("all")
def test_acks_local_write(self):
start_offset0 = self.current_offset(self.topic, 0)
- start_offset1 = self.current_offset(self.topic, 1)
producer = SimpleProducer(self.client, req_acks=SimpleProducer.ACK_AFTER_LOCAL_WRITE)
resp = producer.send_messages(self.topic, self.msg("one"))
@@ -244,7 +249,6 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
@kafka_versions("all")
def test_acks_cluster_commit(self):
start_offset0 = self.current_offset(self.topic, 0)
- start_offset1 = self.current_offset(self.topic, 1)
producer = SimpleProducer(
self.client,
@@ -360,7 +364,6 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
@kafka_versions("all")
def test_async_simple_producer(self):
start_offset0 = self.current_offset(self.topic, 0)
- start_offset1 = self.current_offset(self.topic, 1)
producer = SimpleProducer(self.client, async=True)
resp = producer.send_messages(self.topic, self.msg("one"))
@@ -373,7 +376,6 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
@kafka_versions("all")
def test_async_keyed_producer(self):
start_offset0 = self.current_offset(self.topic, 0)
- start_offset1 = self.current_offset(self.topic, 1)
producer = KeyedProducer(self.client, partitioner = RoundRobinPartitioner, async=True)
diff --git a/test/test_protocol.py b/test/test_protocol.py
index 2089f48..f6e3c96 100644
--- a/test/test_protocol.py
+++ b/test/test_protocol.py
@@ -1,25 +1,18 @@
-import contextlib
-from contextlib import contextmanager
+from contextlib import contextmanager, nested
import struct
-import unittest2
-import mock
-from mock import sentinel
+import unittest2
+from mock import patch, sentinel
-from kafka import KafkaClient
from kafka.common import (
OffsetRequest, OffsetCommitRequest, OffsetFetchRequest,
OffsetResponse, OffsetCommitResponse, OffsetFetchResponse,
ProduceRequest, FetchRequest, Message, ChecksumError,
ConsumerFetchSizeTooSmall, ProduceResponse, FetchResponse, OffsetAndMessage,
- BrokerMetadata, PartitionMetadata, TopicAndPartition, KafkaUnavailableError,
- ProtocolError, LeaderUnavailableError, PartitionUnavailableError,
+ BrokerMetadata, PartitionMetadata, ProtocolError,
UnsupportedCodecError
)
-from kafka.codec import (
- has_snappy, gzip_encode, gzip_decode,
- snappy_encode, snappy_decode
-)
+from kafka.codec import has_snappy, gzip_decode, snappy_decode
import kafka.protocol
from kafka.protocol import (
ATTRIBUTE_CODEC_MASK, CODEC_NONE, CODEC_GZIP, CODEC_SNAPPY, KafkaProtocol,
@@ -701,12 +694,12 @@ class TestProtocol(unittest2.TestCase):
@contextmanager
def mock_create_message_fns(self):
- patches = contextlib.nested(
- mock.patch.object(kafka.protocol, "create_message",
+ patches = nested(
+ patch.object(kafka.protocol, "create_message",
return_value=sentinel.message),
- mock.patch.object(kafka.protocol, "create_gzip_message",
+ patch.object(kafka.protocol, "create_gzip_message",
return_value=sentinel.gzip_message),
- mock.patch.object(kafka.protocol, "create_snappy_message",
+ patch.object(kafka.protocol, "create_snappy_message",
return_value=sentinel.snappy_message),
)
diff --git a/test/test_util.py b/test/test_util.py
index 7b5f294..dbc3fe6 100644
--- a/test/test_util.py
+++ b/test/test_util.py
@@ -1,6 +1,8 @@
# -*- coding: utf-8 -*-
import struct
+
import unittest2
+
import kafka.util
import kafka.common
View Lorry Controller Status