improve tests

Author: Swen Wenzel

test/test_admin_integration.py

@@ -7,7 +7,7 @@
 
 from kafka.admin import (
     ACLFilter, ACLOperation, ACLPermissionType, ResourcePattern, ResourceType, ACL, ConfigResource, ConfigResourceType)
-from kafka.errors import (NoError, GroupCoordinatorNotAvailableError)
+from kafka.errors import (NoError, GroupCoordinatorNotAvailableError, NonEmptyGroupError, GroupIdNotFoundError)
 
 
 @pytest.mark.skipif(env_kafka_version() < (0, 11), reason="ACL features require broker >=0.11")
@@ -241,9 +241,8 @@ def consumer_thread(i, group_id):
 
 
 @pytest.mark.skipif(env_kafka_version() < (1, 1), reason="Delete consumer groups requires broker >=1.1")
-def test_delete_consumergroups_inactive_group(kafka_admin_client, kafka_consumer_factory, send_messages):
+def test_delete_consumergroups(kafka_admin_client, kafka_consumer_factory, send_messages):
     send_messages(range(0, 100), partition=0)
-    send_messages(range(0, 100), partition=1)
     consumer1 = kafka_consumer_factory(group_id="group1")
     next(consumer1)
     consumer1.close()
@@ -261,7 +260,13 @@ def test_delete_consumergroups_inactive_group(kafka_admin_client, kafka_consumer
     assert "group2" in consumergroups
     assert "group3" in consumergroups
 
-    kafka_admin_client.delete_consumer_groups(["group1", "group2"])
+    delete_results = {
+        group_id: error
+        for group_id, error in kafka_admin_client.delete_consumer_groups(["group1", "group2"])
+    }
+    assert delete_results["group1"] == NoError
+    assert delete_results["group2"] == NoError
+    assert "group3" not in delete_results
 
     consumergroups = {group_id for group_id, _ in kafka_admin_client.list_consumer_groups()}
     assert "group1" not in consumergroups
@@ -270,30 +275,30 @@ def test_delete_consumergroups_inactive_group(kafka_admin_client, kafka_consumer
 
 
 @pytest.mark.skipif(env_kafka_version() < (1, 1), reason="Delete consumer groups requires broker >=1.1")
-def test_delete_consumergroups_active_group(kafka_admin_client, kafka_consumer_factory, send_messages):
+def test_delete_consumergroups_with_errors(kafka_admin_client, kafka_consumer_factory, send_messages):
         send_messages(range(0, 100), partition=0)
-        send_messages(range(0, 100), partition=1)
         consumer1 = kafka_consumer_factory(group_id="group1")
         next(consumer1)
         consumer1.close()
 
         consumer2 = kafka_consumer_factory(group_id="group2")
         next(consumer2)
 
-        consumer3 = kafka_consumer_factory(group_id="group3")
-        next(consumer3)
-        consumer3.close()
-
         consumergroups = {group_id for group_id, _ in kafka_admin_client.list_consumer_groups()}
         assert "group1" in consumergroups
         assert "group2" in consumergroups
-        assert "group3" in consumergroups
+        assert "group3" not in consumergroups
+
+        delete_results = {
+            group_id: error
+            for group_id, error in kafka_admin_client.delete_consumer_groups(["group1", "group2", "group3"])
+        }
 
-        # TODO use more specific exception
-        with pytest.raises(Exception):
-            kafka_admin_client.delete_consumer_groups(["group1", "group2"])
+        assert delete_results["group1"] == NoError
+        assert delete_results["group2"] == NonEmptyGroupError
+        assert delete_results["group3"] == GroupIdNotFoundError
 
         consumergroups = {group_id for group_id, _ in kafka_admin_client.list_consumer_groups()}
         assert "group1" not in consumergroups
         assert "group2" in consumergroups
-        assert "group3" in consumergroups
+        assert "group3" not in consumergroups