Fix tests for version below 0.11
-# import asyncio
from ._testutil import (
KafkaIntegrationTestCase, run_until_complete, kafka_versions
)
@@ -137,6 +136,27 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
consumer = await self.consumer_factory()
await consumer.getone()
+ @kafka_versions('>=0.10.0')
+ @run_until_complete
+ async def test_sasl_deny_group_read(self):
+ self.acl_manager.add_acl(
+ allow_principal="test", operation="All", group=self.group_id)
+ self.acl_manager.add_acl(
+ deny_principal="test", operation="READ", group=self.group_id)
+
+ # Consumer will require READ to perform JoinGroup
+ with self.assertRaises(GroupAuthorizationFailedError):
+ consumer = await self.consumer_factory()
+ await consumer.getone()
+
+ @kafka_versions('>=0.11.0')
+ @run_until_complete
+ async def test_sasl_deny_transaction_group_describe(self):
+ self.acl_manager.add_acl(
+ allow_principal="test", operation="All", group=self.group_id)
+ self.acl_manager.add_acl(
+ deny_principal="test", operation="DESCRIBE", group=self.group_id)
+
# Transactional producers will also have the same problem to commit
producer = await self.producer_factory(transactional_id="test_id")
with self.assertRaises(GroupAuthorizationFailedError):
@@ -145,19 +165,14 @@ class TestKafkaProducerIntegration(KafkaIntegrationTestCase):
{TopicPartition(self.topic, 0): 0},
group_id=self.group_id)
- @kafka_versions('>=0.10.0')
+ @kafka_versions('>=0.11.0')
@run_until_complete
- async def test_sasl_deny_group_read(self):
+ async def test_sasl_deny_transaction_group_read(self):
self.acl_manager.add_acl(
allow_principal="test", operation="All", group=self.group_id)
self.acl_manager.add_acl(
deny_principal="test", operation="READ", group=self.group_id)
- # Consumer will require READ to perform JoinGroup
- with self.assertRaises(GroupAuthorizationFailedError):
- consumer = await self.consumer_factory()
- await consumer.getone()
-
# Transactional producers will also have the same problem to commit
producer = await self.producer_factory(transactional_id="test_id")
with self.assertRaises(GroupAuthorizationFailedError):
