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,111 @@
 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):
+    ret = func(*args, **kw)
+    return ret
+
+def set_params(**params):
+    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):