Introduce new fixtures to prepare for migration to pytest.

This commits adds new pytest fixtures in prepation for the
migration of unittest.TestCases to pytest test cases. The handling
of temporary dir creation was also changed so that we can use
the pytest tmpdir fixture after the migration.

Author: asdaraujo

pylint.rc

@@ -1,5 +1,6 @@
 [TYPECHECK]
 ignored-classes=SyncManager,_socketobject
+generated-members=py.*
 
 [MESSAGES CONTROL]
 disable=E1129

test/conftest.py

@@ -1,35 +1,138 @@
 from __future__ import absolute_import
 
-import os
+import inspect
 
 import pytest
+from decorator import decorate
 
 from test.fixtures import KafkaFixture, ZookeeperFixture
-
+from test.testutil import kafka_version, random_string
 
 @pytest.fixture(scope="module")
 def version():
-    if 'KAFKA_VERSION' not in os.environ:
-        return ()
-    return tuple(map(int, os.environ['KAFKA_VERSION'].split('.')))
-
+    return kafka_version()
 
 @pytest.fixture(scope="module")
-def zookeeper(version, request):
-    assert version
+def zookeeper():
     zk = ZookeeperFixture.instance()
     yield zk
     zk.close()
 
+_params = {}
+def _set_params(func, *args, **kw):
+    """Private helper method for set_params
+    """
+    ret = func(*args, **kw)
+    return ret
+
+def set_params(**params):
+    """Allows specific parameters to be injected into the instatiation of a fixture.
+
+    The parameters are specified as shown in the example below:
+
+    @set_params(simple_consumer={'auto_offset_reset':'earliest'})
+    def test_consumer_with_earliest(simple_consumer):
+        ...
+
+    @set_params(simple_consumer={'auto_offset_reset':'latest'})
+    def test_consumer_with_latest(simple_consumer):
+        ...
+
+    The parameters set this way are exposed to the fixture method as an injected parameter:
+
+    @pytest.fixture
+    def simple_consumer(...some_params..., simple_consumer_params):
+        ...
+
+    This avoids having to explicitly handle instantiation and destruction on each test method.
+    This works only for fixtures with "function" scope.
+
+    Note that this is different from the behaviour of @pytest.mark.parametrize, since
+    @pytest.mark.parametrize will force multiple executions of the same test for the different
+    parameter values, which is different from the intended use above.
+    """
+    global _params
+
+    def real_set_params(func, *args, **kwargs):
+        _params[func.__name__] = params
+        ret = decorate(func, _set_params)
+        return ret
+
+    return real_set_params
+
+# this must be called from the module where the broker is used
+def set_broker_params(**params):
+    global _params
+    caller = inspect.stack()[1]
+    if isinstance(caller, tuple):
+        filename = caller[1]
+    else:
+        filename = inspect.stack()[1].filename
+    _params[filename] = params
+
+@pytest.fixture(scope='module')
+def broker_params(request):
+    global _params
+    return _params[request.node.fspath] if request.node.fspath in _params else None
+
+@pytest.fixture
+def kafka_consumer_params(request):
+    return _get_params(request)
+
+@pytest.fixture
+def kafka_producer_params(request):
+    return _get_params(request)
+
+def _get_params(request):
+    global _params
+    if request._pyfuncitem.name in _params and \
+       request.fixturename.replace('_params', '') in _params[request._pyfuncitem.name]:
+        return _params[request._pyfuncitem.name][request.fixturename.replace('_params', '')]
+    return None
+
+@pytest.fixture(scope="module")
+def kafka_broker(kafka_brokers):
+    return kafka_brokers[0]
 
 @pytest.fixture(scope="module")
-def kafka_broker(version, zookeeper, request):
-    assert version
-    k = KafkaFixture.instance(0, zookeeper.host, zookeeper.port,
-                              partitions=4)
-    yield k
-    k.close()
+def kafka_brokers(version, zookeeper, broker_params):
+    assert version, 'KAFKA_VERSION must be specified to run integration tests'
+    params = {} if broker_params is None else broker_params.copy()
+    params.setdefault('partitions', 4)
+    num_brokers = params.pop('num_brokers', 1)
+    brokers = tuple(KafkaFixture.instance(x, zookeeper.host, zookeeper.port, **params)
+                    for x in range(num_brokers))
+    yield brokers
+    for broker in brokers:
+        broker.close()
 
+@pytest.fixture
+def kafka_client(kafka_broker, request):
+    (client,) = kafka_broker.get_clients(cnt=1, client_id='%s_client' % (request.node.name,))
+    yield client
+    client.close()
+
+@pytest.fixture
+def kafka_consumer(kafka_broker, topic, request, kafka_consumer_params):
+    params = {} if kafka_consumer_params is None else kafka_consumer_params.copy()
+    params.setdefault('client_id', 'consumer_%s' % (request.node.name,))
+    (consumer,) = kafka_broker.get_consumers(cnt=1, topics=[topic], **params)
+    yield consumer
+    consumer.close()
+
+@pytest.fixture
+def kafka_producer(kafka_broker, request, kafka_producer_params):
+    params = {} if kafka_producer_params is None else kafka_producer_params.copy()
+    params.setdefault('client_id', 'producer_%s' % (request.node.name,))
+    (producer,) = kafka_broker.get_producers(cnt=1, **params)
+    yield producer
+    producer.close()
+
+@pytest.fixture
+def topic(kafka_broker, request):
+    topic_name = '%s_%s' % (request.node.name, random_string(10))
+    kafka_broker.create_topics([topic_name])
+    return topic_name
 
 @pytest.fixture
 def conn(mocker):