Skip to content

Commit

Permalink
Merge pull request #796 from dpkp/fixture_auto_ports
Browse files Browse the repository at this point in the history
Fixture auto ports
  • Loading branch information
dpkp committed Aug 5, 2016
2 parents bccac7b + 8d19cae commit b24a5c2
Show file tree
Hide file tree
Showing 3 changed files with 27 additions and 32 deletions.
3 changes: 3 additions & 0 deletions .covrc
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
[run]
omit =
kafka/vendor/*
50 changes: 21 additions & 29 deletions test/fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ def instance(cls):
(host, port) = (parse.hostname, parse.port)
fixture = ExternalService(host, port)
else:
(host, port) = ("127.0.0.1", get_open_port())
(host, port) = ("127.0.0.1", None)
fixture = cls(host, port)

fixture.open()
Expand All @@ -124,21 +124,18 @@ def kafka_run_class_env(self):
return env

def out(self, message):
log.info("*** Zookeeper [%s:%d]: %s", self.host, self.port, message)
log.info("*** Zookeeper [%s:%s]: %s", self.host, self.port or '(auto)', message)

def open(self):
self.tmp_dir = tempfile.mkdtemp()
self.out("Running local instance...")
log.info(" host = %s", self.host)
log.info(" port = %s", self.port)
log.info(" port = %s", self.port or '(auto)')
log.info(" tmp_dir = %s", self.tmp_dir)

# Generate configs
# Configure Zookeeper child process
template = self.test_resource("zookeeper.properties")
properties = os.path.join(self.tmp_dir, "zookeeper.properties")
self.render_template(template, properties, vars(self))

# Configure Zookeeper child process
args = self.kafka_run_class_args("org.apache.zookeeper.server.quorum.QuorumPeerMain", properties)
env = self.kafka_run_class_env()

Expand All @@ -148,13 +145,12 @@ def open(self):
backoff = 1
end_at = time.time() + max_timeout
tries = 1
auto_port = (self.port is None)
while time.time() < end_at:
self.out('Attempting to start (try #%d)' % tries)
try:
os.stat(properties)
except:
log.warning('Config %s not found -- re-rendering', properties)
self.render_template(template, properties, vars(self))
if auto_port:
self.port = get_open_port()
self.out('Attempting to start on port %d (try #%d)' % (self.port, tries))
self.render_template(template, properties, vars(self))
self.child = SpawnedService(args, env)
self.child.start()
timeout = min(timeout, max(end_at - time.time(), 0))
Expand Down Expand Up @@ -194,8 +190,6 @@ def instance(cls, broker_id, zk_host, zk_port, zk_chroot=None,
(host, port) = (parse.hostname, parse.port)
fixture = ExternalService(host, port)
else:
if port is None:
port = get_open_port()
# force IPv6 here because of a confusing point:
#
# - if the string "localhost" is passed, Kafka will *only* bind to the IPv4 address of localhost
Expand Down Expand Up @@ -245,7 +239,7 @@ def kafka_run_class_env(self):
return env

def out(self, message):
log.info("*** Kafka [%s:%d]: %s", self.host, self.port, message)
log.info("*** Kafka [%s:%s]: %s", self.host, self.port or '(auto)', message)

def open(self):
if self.running:
Expand All @@ -255,7 +249,7 @@ def open(self):
self.tmp_dir = tempfile.mkdtemp()
self.out("Running local instance...")
log.info(" host = %s", self.host)
log.info(" port = %s", self.port)
log.info(" port = %s", self.port or '(auto)')
log.info(" transport = %s", self.transport)
log.info(" broker_id = %s", self.broker_id)
log.info(" zk_host = %s", self.zk_host)
Expand All @@ -269,12 +263,6 @@ def open(self):
os.mkdir(os.path.join(self.tmp_dir, "logs"))
os.mkdir(os.path.join(self.tmp_dir, "data"))

# Generate configs
template = self.test_resource("kafka.properties")
properties = os.path.join(self.tmp_dir, "kafka.properties")
self.render_template(template, properties, vars(self))

# Party!
self.out("Creating Zookeeper chroot node...")
args = self.kafka_run_class_args("org.apache.zookeeper.ZooKeeperMain",
"-server", "%s:%d" % (self.zk_host, self.zk_port),
Expand All @@ -292,6 +280,8 @@ def open(self):
self.out("Done!")

# Configure Kafka child process
properties = os.path.join(self.tmp_dir, "kafka.properties")
template = self.test_resource("kafka.properties")
args = self.kafka_run_class_args("kafka.Kafka", properties)
env = self.kafka_run_class_env()

Expand All @@ -300,13 +290,15 @@ def open(self):
backoff = 1
end_at = time.time() + max_timeout
tries = 1
auto_port = (self.port is None)
while time.time() < end_at:
self.out('Attempting to start (try #%d)' % tries)
try:
os.stat(properties)
except:
log.warning('Config %s not found -- re-rendering', properties)
self.render_template(template, properties, vars(self))
# We have had problems with port conflicts on travis
# so we will try a different port on each retry
# unless the fixture was passed a specific port
if auto_port:
self.port = get_open_port()
self.out('Attempting to start on port %d (try #%d)' % (self.port, tries))
self.render_template(template, properties, vars(self))
self.child = SpawnedService(args, env)
self.child.start()
timeout = min(timeout, max(end_at - time.time(), 0))
Expand Down
6 changes: 3 additions & 3 deletions tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -21,18 +21,18 @@ deps =
xxhash
py26: unittest2
commands =
py.test {posargs:--pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka}
py.test {posargs:--pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka --cov-config=.covrc}
setenv =
PROJECT_ROOT = {toxinidir}
passenv = KAFKA_VERSION

[testenv:py26]
# pylint doesn't support python2.6
commands = py.test {posargs:--cov=kafka}
commands = py.test {posargs:--cov=kafka --cov-config=.covrc}

[testenv:pypy]
# pylint is super slow on pypy...
commands = py.test {posargs:--cov=kafka}
commands = py.test {posargs:--cov=kafka --cov-config=.covrc}

[testenv:docs]
deps =
Expand Down

0 comments on commit b24a5c2

Please sign in to comment.