kafka.admin: Fix DESCRIBE_TOKENS ACLOperation enum; support authorized_operations=None

kafka/admin/acl_resource.py

@@ -38,7 +38,7 @@ class ACLOperation(IntEnum):
     ALTER_CONFIGS = 11,
     IDEMPOTENT_WRITE = 12,
     CREATE_TOKENS = 13,
-    DESCRIBE_TOKENS = 13
+    DESCRIBE_TOKENS = 14
 
 
 class ACLPermissionType(IntEnum):

kafka/admin/client.py

@@ -532,10 +532,10 @@ def delete_topics(self, topics, timeout_ms=None):
 
     def _process_metadata_response(self, metadata_response):
         obj = metadata_response.to_object()
-        if 'authorized_operations' in obj:
+        if obj.get('authorized_operations', None) is not None:
             obj['authorized_operations'] = list(map(lambda acl: acl.name, valid_acl_operations(obj['authorized_operations'])))
         for t in obj['topics']:
-            if 'authorized_operations' in t:
+            if t.get('authorized_operations', None) is not None:
                 t['authorized_operations'] = list(map(lambda acl: acl.name, valid_acl_operations(t['authorized_operations'])))
         return obj
 

kafka/protocol/types.py

@@ -339,6 +339,8 @@ def decode(cls, data):
 
     @classmethod
     def encode(cls, value):
+        if value is None:
+            value = {}
         ret = UnsignedVarInt32.encode(len(value))
         for k, v in value.items():
             # do we allow for other data types ?? It could get complicated really fast
@@ -388,10 +390,15 @@ def decode(self, data):
 class BitField(AbstractType):
     @classmethod
     def decode(cls, data):
-        return cls.from_32_bit_field(Int32.decode(data))
+        vals = cls.from_32_bit_field(Int32.decode(data))
+        if vals == {31}:
+            vals = None
+        return vals
 
     @classmethod
     def encode(cls, vals):
+        if vals is None:
+            vals = {31}
         # to_32_bit_field returns unsigned val, so we need to
         # encode >I to avoid crash if/when byte 31 is set
         # (note that decode as signed still works fine)

test/protocol/test_bit_field.py

@@ -6,8 +6,14 @@
 
 
 @pytest.mark.parametrize(('test_set',), [
-    (set([0, 1, 5, 10, 31]),),
-    (set(range(32)),),
+    (set([0, 1, 5, 10]),),
+    (set(range(15)),),
+    (None,),
 ])
 def test_bit_field(test_set):
     assert BitField.decode(io.BytesIO(BitField.encode(test_set))) == test_set
+
+
+def test_bit_field_null():
+    assert BitField.from_32_bit_field(-2147483648) == {31}
+    assert BitField.decode(io.BytesIO(BitField.encode({31}))) is None

test/protocol/test_parser.py

@@ -72,7 +72,7 @@
             'response': (
                 b'\x00\x00\x00E\x00\x00\x00\x03\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\tlocalhost\x00\x00\xb9}\xff\xff\x00\x1634wjNp3hRJixCRvMlK9Znw\x00\x00\x00\x00\x00\x00\x00\x00\x80\x00\x00\x00',
                 3,
-                MetadataResponse[8](throttle_time_ms=0, brokers=[(0, 'localhost', 47485, None)], cluster_id='34wjNp3hRJixCRvMlK9Znw', controller_id=0, topics=[], authorized_operations={31}),
+                MetadataResponse[8](throttle_time_ms=0, brokers=[(0, 'localhost', 47485, None)], cluster_id='34wjNp3hRJixCRvMlK9Znw', controller_id=0, topics=[], authorized_operations=None),
             ),
         },
         {
@@ -83,7 +83,7 @@
             'response': (
                 b'\x00\x00\x00E\x00\x00\x00\x04\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\tlocalhost\x00\x00\xb9}\xff\xff\x00\x1634wjNp3hRJixCRvMlK9Znw\x00\x00\x00\x00\x00\x00\x00\x00\x80\x00\x00\x00',
                 4,
-                MetadataResponse[8](throttle_time_ms=0, brokers=[(0, 'localhost', 47485, None)], cluster_id='34wjNp3hRJixCRvMlK9Znw', controller_id=0, topics=[], authorized_operations={31}),
+                MetadataResponse[8](throttle_time_ms=0, brokers=[(0, 'localhost', 47485, None)], cluster_id='34wjNp3hRJixCRvMlK9Znw', controller_id=0, topics=[], authorized_operations=None),
             ),
         }
     ),