Fix TaggedFields value encoding; add test coverage (#2725)

Author: dpkp

kafka/protocol/types.py

@@ -335,8 +335,9 @@ def encode(cls, value):
         for k, v in value.items():
             # do we allow for other data types ?? It could get complicated really fast
             assert isinstance(v, bytes), 'Value {} is not a byte array'.format(v)
-            assert isinstance(k, int) and k > 0, 'Key {} is not a positive integer'.format(k)
+            assert isinstance(k, int) and k >= 0, 'Key {} is not a non-negative integer'.format(k)
             ret += UnsignedVarInt32.encode(k)
+            ret += UnsignedVarInt32.encode(len(v))
             ret += v
         return ret
 

test/protocol/test_tagged_fields.py

@@ -0,0 +1,17 @@
+import io
+
+import pytest
+
+from kafka.protocol.types import TaggedFields, UnsignedVarInt32
+
+
+def test_tagged_fields():
+    val = {0: b'fizz', 1: b'foobar'}
+    encoded = TaggedFields.encode(val)
+    # length(2), tag(0), size(4), 'fizz', tag(2), size(6), 'foobar'
+    expected = (UnsignedVarInt32.encode(2) +
+                UnsignedVarInt32.encode(0) + UnsignedVarInt32.encode(4) + b'fizz' +
+                UnsignedVarInt32.encode(1) + UnsignedVarInt32.encode(6) + b'foobar')
+    assert encoded == expected
+    decoded = TaggedFields.decode(io.BytesIO(encoded))
+    assert decoded == val