diff options
| author | Brian Sang <sang.bri@gmail.com> | 2019-05-31 19:05:41 -0700 |
|---|---|---|
| committer | Jeff Widman <jeff@jeffwidman.com> | 2019-06-04 22:53:45 -0700 |
| commit | f126e5bfcc8f41ee5ea29b41ec6eabbc3f441647 (patch) | |
| tree | 83b33425304bdba8dcacb13872b908452ad60761 | |
| parent | 79dd508b14fd2d66a8b6d32353e8e64989c4ff84 (diff) | |
| download | kafka-python-f126e5bfcc8f41ee5ea29b41ec6eabbc3f441647.tar.gz | |
Sanity test for consumer.topics() and consumer.partitions_for_topic()
| -rw-r--r-- | test/test_consumer_group.py | 9 |
1 files changed, 9 insertions, 0 deletions
diff --git a/test/test_consumer_group.py b/test/test_consumer_group.py index d7aaa88..ec26857 100644 --- a/test/test_consumer_group.py +++ b/test/test_consumer_group.py @@ -29,6 +29,15 @@ def test_consumer(kafka_broker, topic, version): assert consumer._client._conns[node_id].state is ConnectionStates.CONNECTED consumer.close() +@pytest.mark.skipif(not version(), reason="No KAFKA_VERSION set") +def test_consumer_topics(kafka_broker, topic, version): + consumer = KafkaConsumer(bootstrap_servers=get_connect_str(kafka_broker)) + # Necessary to drive the IO + consumer.poll(500) + consumer_topics = consumer.topics() + assert topic in consumer_topics + assert len(consumer.partitions_for_topic(topic)) > 0 + consumer.close() @pytest.mark.skipif(version() < (0, 9), reason='Unsupported Kafka Version') @pytest.mark.skipif(not version(), reason="No KAFKA_VERSION set") |