Add api_version config to KafkaClient, deprecate str in favor of tuplesclient_api_version

1 files changed, 14 insertions, 13 deletions

diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index bb62b7b..3435292 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -24,24 +24,24 @@ from kafka.util import WeakMethod
 
 
 @pytest.fixture
-def coordinator(conn):
-    return ConsumerCoordinator(KafkaClient(), SubscriptionState(), Metrics(),
-                               'consumer')
+def client(conn):
+    return KafkaClient(api_version=(0, 9))
 
+@pytest.fixture
+def coordinator(client):
+    return ConsumerCoordinator(client, SubscriptionState(), Metrics(),
+                               'consumer')
 
-def test_init(conn):
-    cli = KafkaClient()
-    coordinator = ConsumerCoordinator(cli, SubscriptionState(), Metrics(),
-                                      'consumer')
 
+def test_init(client, coordinator):
     # metadata update on init 
-    assert cli.cluster._need_update is True
-    assert WeakMethod(coordinator._handle_metadata_update) in cli.cluster._listeners
+    assert client.cluster._need_update is True
+    assert WeakMethod(coordinator._handle_metadata_update) in client.cluster._listeners
 
 
 @pytest.mark.parametrize("api_version", [(0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9)])
-def test_autocommit_enable_api_version(conn, api_version):
-    coordinator = ConsumerCoordinator(KafkaClient(), SubscriptionState(),
+def test_autocommit_enable_api_version(client, api_version):
+    coordinator = ConsumerCoordinator(client, SubscriptionState(),
                                       Metrics(), 'consumer',
                                       enable_auto_commit=True,
                                       group_id='foobar',
@@ -80,7 +80,7 @@ def test_group_protocols(coordinator):
     ]
 
 
-@pytest.mark.parametrize('api_version', [(0, 8), (0, 8, 1), (0, 8, 2), (0, 9)])
+@pytest.mark.parametrize('api_version', [(0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9)])
 def test_pattern_subscription(coordinator, api_version):
     coordinator.config['api_version'] = api_version
     coordinator._subscription.subscribe(pattern='foo')
@@ -360,7 +360,8 @@ def test_maybe_auto_commit_offsets_sync(mocker, api_version, group_id, enable,
                                         warn, exc):
     mock_warn = mocker.patch('kafka.coordinator.consumer.log.warning')
     mock_exc = mocker.patch('kafka.coordinator.consumer.log.exception')
-    coordinator = ConsumerCoordinator(KafkaClient(), SubscriptionState(),
+    client = KafkaClient(api_version=api_version)
+    coordinator = ConsumerCoordinator(client, SubscriptionState(),
                                       Metrics(), 'consumer',
                                       api_version=api_version,
                                       enable_auto_commit=enable,