Require resource type objects when creating an ACLResource

Author: Ulrik Johansson

kafka/admin/acl_resource.py

@@ -1,4 +1,5 @@
 from __future__ import absolute_import
+from kafka.errors import IllegalArgumentError
 
 # enum in stdlib as of py3.4
 try:
@@ -69,17 +70,17 @@ def __init__(
             pattern_type=ACLResourcePatternType.LITERAL
     ):
         if not isinstance(resource_type, ACLResourceType):
-            resource_type = ACLResourceType[str(resource_type).upper()]  # pylint: disable-msg=unsubscriptable-object
+            raise IllegalArgumentError("resource_param must be of type ACLResourceType")
         self.resource_type = resource_type
         if not isinstance(operation, ACLOperation):
-            operation = ACLOperation[str(operation).upper()]  # pylint: disable-msg:unsubscriptable-object
+            raise IllegalArgumentError("operation must be of type ACLOperation")
         self.operation = operation
         if not isinstance(permission_type, ACLPermissionType):
-            permission_type = ACLPermissionType[str(permission_type).upper()]  # pylint: disable-msg=unsubscriptable-object
+            raise IllegalArgumentError("permission_type must be of type ACLPermissionType")
         self.permission_type = permission_type
         self.name = name
         self.principal = principal
         self.host = host
         if not isinstance(pattern_type, ACLResourcePatternType):
-            pattern_type = ACLResourcePatternType[str(pattern_type).upper()]  # pylint: disable-msg=unsubscriptable-object
+            raise IllegalArgumentError("pattern_type must be of type ACLResourcePatternType")
         self.pattern_type = pattern_type

test/test_admin.py

@@ -27,13 +27,23 @@ def test_new_partitions():
 
 
 def test_acl_resource():
-    good_resource = kafka.admin.ACLResource("TOPIC", "ALL", "ALLOW", "foo",
-                                            "User:bar", "*", "LITERAL")
+    good_resource = kafka.admin.ACLResource(
+        kafka.admin.ACLResourceType.TOPIC,
+        kafka.admin.ACLOperation.ALL,
+        kafka.admin.ACLPermissionType.ALLOW,
+        "foo",
+        "User:bar",
+        "*",
+        kafka.admin.ACLResourcePatternType.LITERAL
+    )
     assert(good_resource.resource_type == kafka.admin.ACLResourceType.TOPIC)
     assert(good_resource.operation == kafka.admin.ACLOperation.ALL)
     assert(good_resource.permission_type == kafka.admin.ACLPermissionType.ALLOW)
     assert(good_resource.pattern_type == kafka.admin.ACLResourcePatternType.LITERAL)
 
+    with pytest.raises(IllegalArgumentError):
+        bad_resource = kafka.admin.ACLResource("TOPIC", "ALL", "ALLOW", "foo", "User:bar", "*", "LITERAL")
+
 def test_new_topic():
     with pytest.raises(IllegalArgumentError):
         bad_topic = kafka.admin.NewTopic('foo', -1, -1)