openstack/deb-python-kafka
Code Issues Proposed changes

Add pylint to tox.ini; test both kafka and test; default to error-checking only; fixup errors; skip kafka/queue.py

Browse Source
This commit is contained in:
Dana Powers
2014-08-25 10:15:16 -07:00
parent d73d1690b1
commit e151529078
10 changed files with 54 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.travis.yml
View File

@@ -6,7 +6,7 @@ python:
- pypy - pypy
env: env:
- - UNIT_AND_LINT_ONLY=true
- KAFKA_VERSION=0.8.0 - KAFKA_VERSION=0.8.0
- KAFKA_VERSION=0.8.1 - KAFKA_VERSION=0.8.1
- KAFKA_VERSION=0.8.1.1 - KAFKA_VERSION=0.8.1.1
@@ -35,4 +35,4 @@ deploy:
# branch: master # branch: master
script: script:
- tox -e `./travis_selector.sh $TRAVIS_PYTHON_VERSION` - if [ -n "$UNIT_AND_LINT_ONLY" ]; then tox -e lint,`./travis_selector.sh $TRAVIS_PYTHON_VERSION`; else tox -e `./travis_selector.sh $TRAVIS_PYTHON_VERSION`; fi

2
kafka/client.py
View File

@@ -187,7 +187,7 @@ class KafkaClient(object):
def _raise_on_response_error(self, resp): def _raise_on_response_error(self, resp):
try: try:
kafka.common.check_error(resp) kafka.common.check_error(resp)
except (UnknownTopicOrPartitionError, NotLeaderForPartitionError) as e: except (UnknownTopicOrPartitionError, NotLeaderForPartitionError):
self.reset_topic_metadata(resp.topic) self.reset_topic_metadata(resp.topic)
raise raise

2
kafka/conn.py
View File

@@ -93,7 +93,7 @@ class KafkaConnection(local):
# that the socket is in error. we will never get # that the socket is in error. we will never get
# more data from this socket # more data from this socket
if data == '': if data == '':
raise socket.error('Not enough data to read message -- did server kill socket?') raise socket.error("Not enough data to read message -- did server kill socket?")
except socket.error: except socket.error:
log.exception('Unable to receive data from Kafka') log.exception('Unable to receive data from Kafka')

49
test/fixtures.py
View File

@@ -1,5 +1,4 @@
import logging import logging
import glob
import os import os
import os.path import os.path
import shutil import shutil
@@ -9,8 +8,8 @@ import urllib2
import uuid import uuid
from urlparse import urlparse from urlparse import urlparse
from service import ExternalService, SpawnedService from test.service import ExternalService, SpawnedService
from testutil import get_open_port from test.testutil import get_open_port
class Fixture(object): class Fixture(object):
kafka_version = os.environ.get('KAFKA_VERSION', '0.8.0') kafka_version = os.environ.get('KAFKA_VERSION', '0.8.0')
@@ -36,23 +35,23 @@ class Fixture(object):
output_file = os.path.join(output_dir, distfile + '.tgz') output_file = os.path.join(output_dir, distfile + '.tgz')
if os.path.isfile(output_file): if os.path.isfile(output_file):
logging.info("Found file already on disk: %s" % output_file) logging.info("Found file already on disk: %s", output_file)
return output_file return output_file
# New tarballs are .tgz, older ones are sometimes .tar.gz # New tarballs are .tgz, older ones are sometimes .tar.gz
try: try:
url = url_base + distfile + '.tgz' url = url_base + distfile + '.tgz'
logging.info("Attempting to download %s" % (url,)) logging.info("Attempting to download %s", url)
response = urllib2.urlopen(url) response = urllib2.urlopen(url)
except urllib2.HTTPError: except urllib2.HTTPError:
logging.exception("HTTP Error") logging.exception("HTTP Error")
url = url_base + distfile + '.tar.gz' url = url_base + distfile + '.tar.gz'
logging.info("Attempting to download %s" % (url,)) logging.info("Attempting to download %s", url)
response = urllib2.urlopen(url) response = urllib2.urlopen(url)
logging.info("Saving distribution file to %s" % (output_file,)) logging.info("Saving distribution file to %s", output_file)
with open(os.path.join(output_dir, distfile + '.tgz'), 'w') as f: with open(output_file, 'w') as output_file_fd:
f.write(response.read()) output_file_fd.write(response.read())
return output_file return output_file
@@ -117,11 +116,9 @@ class ZookeeperFixture(Fixture):
self.render_template(template, properties, vars(self)) self.render_template(template, properties, vars(self))
# Configure Zookeeper child process # Configure Zookeeper child process
self.child = SpawnedService(args=self.kafka_run_class_args( args = self.kafka_run_class_args("org.apache.zookeeper.server.quorum.QuorumPeerMain", properties)
"org.apache.zookeeper.server.quorum.QuorumPeerMain", env = self.kafka_run_class_env()
properties), self.child = SpawnedService(args, env)
env=self.kafka_run_class_env()
)
# Party! # Party!
self.out("Starting...") self.out("Starting...")
@@ -162,7 +159,7 @@ class KafkaFixture(Fixture):
self.zk_port = zk_port self.zk_port = zk_port
self.zk_chroot = zk_chroot self.zk_chroot = zk_chroot
self.replicas = replicas self.replicas = replicas
self.partitions = partitions self.partitions = partitions
self.tmp_dir = None self.tmp_dir = None
@@ -199,21 +196,19 @@ class KafkaFixture(Fixture):
self.render_template(template, properties, vars(self)) self.render_template(template, properties, vars(self))
# Configure Kafka child process # Configure Kafka child process
self.child = SpawnedService(args=self.kafka_run_class_args( args = self.kafka_run_class_args("kafka.Kafka", properties)
"kafka.Kafka", properties), env = self.kafka_run_class_env()
env=self.kafka_run_class_env() self.child = SpawnedService(args, env)
)
# Party! # Party!
self.out("Creating Zookeeper chroot node...") self.out("Creating Zookeeper chroot node...")
proc = subprocess.Popen(self.kafka_run_class_args( args = self.kafka_run_class_args("org.apache.zookeeper.ZooKeeperMain",
"org.apache.zookeeper.ZooKeeperMain", "-server", "%s:%d" % (self.zk_host, self.zk_port),
"-server", "%s:%d" % (self.zk_host, self.zk_port), "create",
"create", "/%s" % self.zk_chroot, "kafka-python" "/%s" % self.zk_chroot,
), "kafka-python")
env=self.kafka_run_class_env(), env = self.kafka_run_class_env()
stdout=subprocess.PIPE, proc = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stderr=subprocess.PIPE)
if proc.wait() != 0: if proc.wait() != 0:
self.out("Failed to create Zookeeper chroot node") self.out("Failed to create Zookeeper chroot node")

9
test/service.py
View File

@@ -2,7 +2,6 @@ import logging
import re import re
import select import select
import subprocess import subprocess
import sys
import threading import threading
import time import time
@@ -14,7 +13,7 @@ __all__ = [
class ExternalService(object): class ExternalService(object):
def __init__(self, host, port): def __init__(self, host, port):
print("Using already running service at %s:%d" % (host, port)) logging.info("Using already running service at %s:%d", host, port)
self.host = host self.host = host
self.port = port self.port = port
@@ -26,9 +25,11 @@ class ExternalService(object):
class SpawnedService(threading.Thread): class SpawnedService(threading.Thread):
def __init__(self, args=[], env=None): def __init__(self, args=None, env=None):
threading.Thread.__init__(self) threading.Thread.__init__(self)
if args is None:
raise TypeError("args parameter is required")
self.args = args self.args = args
self.env = env self.env = env
self.captured_stdout = [] self.captured_stdout = []
@@ -49,7 +50,7 @@ class SpawnedService(threading.Thread):
alive = True alive = True
while True: while True:
(rds, wds, xds) = select.select([self.child.stdout, self.child.stderr], [], [], 1) (rds, _, _) = select.select([self.child.stdout, self.child.stderr], [], [], 1)
if self.child.stdout in rds: if self.child.stdout in rds:
line = self.child.stdout.readline() line = self.child.stdout.readline()

7
test/test_client.py
View File

@@ -1,6 +1,3 @@
import os
import random
import struct
import unittest2 import unittest2
from mock import MagicMock, patch from mock import MagicMock, patch
@@ -11,9 +8,7 @@ from kafka.common import (
TopicAndPartition, KafkaUnavailableError, TopicAndPartition, KafkaUnavailableError,
LeaderUnavailableError, PartitionUnavailableError LeaderUnavailableError, PartitionUnavailableError
) )
from kafka.protocol import ( from kafka.protocol import create_message
create_message, KafkaProtocol
)
class TestKafkaClient(unittest2.TestCase): class TestKafkaClient(unittest2.TestCase):
def test_init_with_list(self): def test_init_with_list(self):

8
test/test_client_integration.py
View File

@@ -1,13 +1,11 @@
import os import os
import random
import socket import socket
import time
import unittest2 import unittest2
import kafka import kafka
from kafka.common import * from kafka.common import *
from fixtures import ZookeeperFixture, KafkaFixture from test.fixtures import ZookeeperFixture, KafkaFixture
from testutil import * from test.testutil import *
class TestKafkaClientIntegration(KafkaIntegrationTestCase): class TestKafkaClientIntegration(KafkaIntegrationTestCase):
@classmethod @classmethod
@@ -34,7 +32,7 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
with Timer() as t: with Timer() as t:
with self.assertRaises((socket.timeout, socket.error)): with self.assertRaises((socket.timeout, socket.error)):
conn = kafka.conn.KafkaConnection("localhost", server_port, 1.0) kafka.conn.KafkaConnection("localhost", server_port, 1.0)
self.assertGreaterEqual(t.interval, 1.0) self.assertGreaterEqual(t.interval, 1.0)
@kafka_versions("all") @kafka_versions("all")

18
test/test_conn.py
View File

@@ -24,13 +24,13 @@ class ConnTest(unittest2.TestCase):
self.addCleanup(patcher.stop) self.addCleanup(patcher.stop)
# Also mock socket.sendall() to appear successful # Also mock socket.sendall() to appear successful
socket.create_connection().sendall.return_value = None self.MockCreateConn().sendall.return_value = None
# And mock socket.recv() to return two payloads, then '', then raise # And mock socket.recv() to return two payloads, then '', then raise
# Note that this currently ignores the num_bytes parameter to sock.recv() # Note that this currently ignores the num_bytes parameter to sock.recv()
payload_size = len(self.config['payload']) payload_size = len(self.config['payload'])
payload2_size = len(self.config['payload2']) payload2_size = len(self.config['payload2'])
socket.create_connection().recv.side_effect = [ self.MockCreateConn().recv.side_effect = [
struct.pack('>i', payload_size), struct.pack('>i', payload_size),
struct.pack('>%ds' % payload_size, self.config['payload']), struct.pack('>%ds' % payload_size, self.config['payload']),
struct.pack('>i', payload2_size), struct.pack('>i', payload2_size),
@@ -42,7 +42,7 @@ class ConnTest(unittest2.TestCase):
self.conn = KafkaConnection(self.config['host'], self.config['port']) self.conn = KafkaConnection(self.config['host'], self.config['port'])
# Reset any mock counts caused by __init__ # Reset any mock counts caused by __init__
socket.create_connection.reset_mock() self.MockCreateConn.reset_mock()
def test_collect_hosts__happy_path(self): def test_collect_hosts__happy_path(self):
hosts = "localhost:1234,localhost" hosts = "localhost:1234,localhost"
@@ -81,7 +81,7 @@ class ConnTest(unittest2.TestCase):
def test_init_creates_socket_connection(self): def test_init_creates_socket_connection(self):
KafkaConnection(self.config['host'], self.config['port']) KafkaConnection(self.config['host'], self.config['port'])
socket.create_connection.assert_called_with((self.config['host'], self.config['port']), DEFAULT_SOCKET_TIMEOUT_SECONDS) self.MockCreateConn.assert_called_with((self.config['host'], self.config['port']), DEFAULT_SOCKET_TIMEOUT_SECONDS)
def test_init_failure_raises_connection_error(self): def test_init_failure_raises_connection_error(self):
@@ -102,9 +102,9 @@ class ConnTest(unittest2.TestCase):
pass pass
# Now test that sending attempts to reconnect # Now test that sending attempts to reconnect
self.assertEqual(socket.create_connection.call_count, 0) self.assertEqual(self.MockCreateConn.call_count, 0)
self.conn.send(self.config['request_id'], self.config['payload']) self.conn.send(self.config['request_id'], self.config['payload'])
self.assertEqual(socket.create_connection.call_count, 1) self.assertEqual(self.MockCreateConn.call_count, 1)
def test_send__failure_sets_dirty_connection(self): def test_send__failure_sets_dirty_connection(self):
@@ -131,9 +131,9 @@ class ConnTest(unittest2.TestCase):
pass pass
# Now test that recv'ing attempts to reconnect # Now test that recv'ing attempts to reconnect
self.assertEqual(socket.create_connection.call_count, 0) self.assertEqual(self.MockCreateConn.call_count, 0)
self.conn.recv(self.config['request_id']) self.conn.recv(self.config['request_id'])
self.assertEqual(socket.create_connection.call_count, 1) self.assertEqual(self.MockCreateConn.call_count, 1)
def test_recv__failure_sets_dirty_connection(self): def test_recv__failure_sets_dirty_connection(self):
@@ -160,5 +160,5 @@ class ConnTest(unittest2.TestCase):
# will re-connect and send data to the socket # will re-connect and send data to the socket
self.conn.close() self.conn.close()
self.conn.send(self.config['request_id'], self.config['payload']) self.conn.send(self.config['request_id'], self.config['payload'])
self.assertEqual(socket.create_connection.call_count, 1) self.assertEqual(self.MockCreateConn.call_count, 1)
self.conn._sock.sendall.assert_called_with(self.config['payload']) self.conn._sock.sendall.assert_called_with(self.config['payload'])

1
test/testutil.py
View File

@@ -48,6 +48,7 @@ def get_open_port():
class KafkaIntegrationTestCase(unittest2.TestCase): class KafkaIntegrationTestCase(unittest2.TestCase):
create_client = True create_client = True
topic = None topic = None
server = None
def setUp(self): def setUp(self):
super(KafkaIntegrationTestCase, self).setUp() super(KafkaIntegrationTestCase, self).setUp()

8
tox.ini
View File

@@ -1,5 +1,5 @@
[tox] [tox]
envlist = py26, py27, pypy envlist = lint, py26, py27, pypy
[testenv] [testenv]
deps = deps =
unittest2 unittest2
@@ -11,3 +11,9 @@ commands =
nosetests {posargs:-v --with-id --with-coverage --cover-erase --cover-package kafka} nosetests {posargs:-v --with-id --with-coverage --cover-erase --cover-package kafka}
setenv = setenv =
PROJECT_ROOT = {toxinidir} PROJECT_ROOT = {toxinidir}
[testenv:lint]
deps =
unittest2
mock
pylint
commands = pylint {posargs: -E --ignore=queue.py kafka test}