From cf0b7f0530e765f2cf710bd35daf53bb4ea205d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bruno=20Reni=C3=A9?= Date: Thu, 28 Aug 2014 20:50:20 +0200 Subject: Make all unit tests pass on py3.3/3.4 --- test/test_protocol.py | 357 +++++++++++++++++++++++++------------------------- 1 file changed, 179 insertions(+), 178 deletions(-) (limited to 'test/test_protocol.py') diff --git a/test/test_protocol.py b/test/test_protocol.py index e586228..2107f82 100644 --- a/test/test_protocol.py +++ b/test/test_protocol.py @@ -1,9 +1,9 @@ -import contextlib from contextlib import contextmanager import struct from . import unittest import mock +import six from mock import sentinel from kafka import KafkaClient @@ -38,21 +38,21 @@ class TestProtocol(unittest.TestCase): self.assertEqual(msg.value, payload) def test_create_gzip(self): - payloads = ["v1", "v2"] + payloads = [b"v1", b"v2"] msg = create_gzip_message(payloads) self.assertEqual(msg.magic, 0) self.assertEqual(msg.attributes, ATTRIBUTE_CODEC_MASK & CODEC_GZIP) self.assertEqual(msg.key, None) # Need to decode to check since gzipped payload is non-deterministic decoded = gzip_decode(msg.value) - expect = "".join([ + expect = b"".join([ struct.pack(">q", 0), # MsgSet offset struct.pack(">i", 16), # MsgSet size struct.pack(">i", 1285512130), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", -1), # -1 indicates a null key struct.pack(">i", 2), # Msg length (bytes) - "v1", # Message contents + b"v1", # Message contents struct.pack(">q", 0), # MsgSet offset struct.pack(">i", 16), # MsgSet size @@ -60,27 +60,27 @@ class TestProtocol(unittest.TestCase): struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", -1), # -1 indicates a null key struct.pack(">i", 2), # Msg length (bytes) - "v2", # Message contents + b"v2", # Message contents ]) self.assertEqual(decoded, expect) @unittest.skipUnless(has_snappy(), "Snappy not available") def test_create_snappy(self): - payloads = ["v1", "v2"] + payloads = [b"v1", b"v2"] msg = create_snappy_message(payloads) self.assertEqual(msg.magic, 0) self.assertEqual(msg.attributes, ATTRIBUTE_CODEC_MASK & CODEC_SNAPPY) self.assertEqual(msg.key, None) decoded = snappy_decode(msg.value) - expect = "".join([ + expect = b"".join([ struct.pack(">q", 0), # MsgSet offset struct.pack(">i", 16), # MsgSet size struct.pack(">i", 1285512130), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", -1), # -1 indicates a null key struct.pack(">i", 2), # Msg length (bytes) - "v1", # Message contents + b"v1", # Message contents struct.pack(">q", 0), # MsgSet offset struct.pack(">i", 16), # MsgSet size @@ -88,52 +88,53 @@ class TestProtocol(unittest.TestCase): struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", -1), # -1 indicates a null key struct.pack(">i", 2), # Msg length (bytes) - "v2", # Message contents + b"v2", # Message contents ]) self.assertEqual(decoded, expect) def test_encode_message_header(self): - expect = "".join([ + expect = b"".join([ struct.pack(">h", 10), # API Key struct.pack(">h", 0), # API Version struct.pack(">i", 4), # Correlation Id struct.pack(">h", len("client3")), # Length of clientId - "client3", # ClientId + b"client3", # ClientId ]) - encoded = KafkaProtocol._encode_message_header("client3", 4, 10) + encoded = KafkaProtocol._encode_message_header(b"client3", 4, 10) self.assertEqual(encoded, expect) def test_encode_message(self): - message = create_message("test", "key") + message = create_message(b"test", b"key") encoded = KafkaProtocol._encode_message(message) - expect = "".join([ + print("CRC", -1427009701) + expect = b"".join([ struct.pack(">i", -1427009701), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 3), # Length of key - "key", # key + b"key", # key struct.pack(">i", 4), # Length of value - "test", # value + b"test", # value ]) self.assertEqual(encoded, expect) def test_decode_message(self): - encoded = "".join([ + encoded = b"".join([ struct.pack(">i", -1427009701), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 3), # Length of key - "key", # key + b"key", # key struct.pack(">i", 4), # Length of value - "test", # value + b"test", # value ]) offset = 10 (returned_offset, decoded_message) = list(KafkaProtocol._decode_message(encoded, offset))[0] self.assertEqual(returned_offset, offset) - self.assertEqual(decoded_message, create_message("test", "key")) + self.assertEqual(decoded_message, create_message(b"test", b"key")) def test_encode_message_failure(self): with self.assertRaises(ProtocolError): @@ -141,52 +142,52 @@ class TestProtocol(unittest.TestCase): def test_encode_message_set(self): message_set = [ - create_message("v1", "k1"), - create_message("v2", "k2") + create_message(b"v1", b"k1"), + create_message(b"v2", b"k2") ] encoded = KafkaProtocol._encode_message_set(message_set) - expect = "".join([ + expect = b"".join([ struct.pack(">q", 0), # MsgSet Offset struct.pack(">i", 18), # Msg Size struct.pack(">i", 1474775406), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 2), # Length of key - "k1", # Key + b"k1", # Key struct.pack(">i", 2), # Length of value - "v1", # Value + b"v1", # Value struct.pack(">q", 0), # MsgSet Offset struct.pack(">i", 18), # Msg Size struct.pack(">i", -16383415), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 2), # Length of key - "k2", # Key + b"k2", # Key struct.pack(">i", 2), # Length of value - "v2", # Value + b"v2", # Value ]) self.assertEqual(encoded, expect) def test_decode_message_set(self): - encoded = "".join([ + encoded = b"".join([ struct.pack(">q", 0), # MsgSet Offset struct.pack(">i", 18), # Msg Size struct.pack(">i", 1474775406), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 2), # Length of key - "k1", # Key + b"k1", # Key struct.pack(">i", 2), # Length of value - "v1", # Value + b"v1", # Value struct.pack(">q", 1), # MsgSet Offset struct.pack(">i", 18), # Msg Size struct.pack(">i", -16383415), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 2), # Length of key - "k2", # Key + b"k2", # Key struct.pack(">i", 2), # Length of value - "v2", # Value + b"v2", # Value ]) msgs = list(KafkaProtocol._decode_message_set_iter(encoded)) @@ -197,17 +198,17 @@ class TestProtocol(unittest.TestCase): returned_offset2, decoded_message2 = msg2 self.assertEqual(returned_offset1, 0) - self.assertEqual(decoded_message1, create_message("v1", "k1")) + self.assertEqual(decoded_message1, create_message(b"v1", b"k1")) self.assertEqual(returned_offset2, 1) - self.assertEqual(decoded_message2, create_message("v2", "k2")) + self.assertEqual(decoded_message2, create_message(b"v2", b"k2")) def test_decode_message_gzip(self): - gzip_encoded = ('\xc0\x11\xb2\xf0\x00\x01\xff\xff\xff\xff\x00\x00\x000' - '\x1f\x8b\x08\x00\xa1\xc1\xc5R\x02\xffc`\x80\x03\x01' - '\x9f\xf9\xd1\x87\x18\x18\xfe\x03\x01\x90\xc7Tf\xc8' - '\x80$wu\x1aW\x05\x92\x9c\x11\x00z\xc0h\x888\x00\x00' - '\x00') + gzip_encoded = (b'\xc0\x11\xb2\xf0\x00\x01\xff\xff\xff\xff\x00\x00\x000' + b'\x1f\x8b\x08\x00\xa1\xc1\xc5R\x02\xffc`\x80\x03\x01' + b'\x9f\xf9\xd1\x87\x18\x18\xfe\x03\x01\x90\xc7Tf\xc8' + b'\x80$wu\x1aW\x05\x92\x9c\x11\x00z\xc0h\x888\x00\x00' + b'\x00') offset = 11 messages = list(KafkaProtocol._decode_message(gzip_encoded, offset)) @@ -216,18 +217,18 @@ class TestProtocol(unittest.TestCase): returned_offset1, decoded_message1 = msg1 self.assertEqual(returned_offset1, 0) - self.assertEqual(decoded_message1, create_message("v1")) + self.assertEqual(decoded_message1, create_message(b"v1")) returned_offset2, decoded_message2 = msg2 self.assertEqual(returned_offset2, 0) - self.assertEqual(decoded_message2, create_message("v2")) + self.assertEqual(decoded_message2, create_message(b"v2")) @unittest.skipUnless(has_snappy(), "Snappy not available") def test_decode_message_snappy(self): - snappy_encoded = ('\xec\x80\xa1\x95\x00\x02\xff\xff\xff\xff\x00\x00' - '\x00,8\x00\x00\x19\x01@\x10L\x9f[\xc2\x00\x00\xff' - '\xff\xff\xff\x00\x00\x00\x02v1\x19\x1bD\x00\x10\xd5' - '\x96\nx\x00\x00\xff\xff\xff\xff\x00\x00\x00\x02v2') + snappy_encoded = (b'\xec\x80\xa1\x95\x00\x02\xff\xff\xff\xff\x00\x00' + b'\x00,8\x00\x00\x19\x01@\x10L\x9f[\xc2\x00\x00\xff' + b'\xff\xff\xff\x00\x00\x00\x02v1\x19\x1bD\x00\x10\xd5' + b'\x96\nx\x00\x00\xff\xff\xff\xff\x00\x00\x00\x02v2') offset = 11 messages = list(KafkaProtocol._decode_message(snappy_encoded, offset)) self.assertEqual(len(messages), 2) @@ -236,14 +237,14 @@ class TestProtocol(unittest.TestCase): returned_offset1, decoded_message1 = msg1 self.assertEqual(returned_offset1, 0) - self.assertEqual(decoded_message1, create_message("v1")) + self.assertEqual(decoded_message1, create_message(b"v1")) returned_offset2, decoded_message2 = msg2 self.assertEqual(returned_offset2, 0) - self.assertEqual(decoded_message2, create_message("v2")) + self.assertEqual(decoded_message2, create_message(b"v2")) def test_decode_message_checksum_error(self): - invalid_encoded_message = "This is not a valid encoded message" + invalid_encoded_message = b"This is not a valid encoded message" iter = KafkaProtocol._decode_message(invalid_encoded_message, 0) self.assertRaises(ChecksumError, list, iter) @@ -254,25 +255,25 @@ class TestProtocol(unittest.TestCase): list(KafkaProtocol._decode_message_set_iter('a')) def test_decode_message_set_stop_iteration(self): - encoded = "".join([ + encoded = b"".join([ struct.pack(">q", 0), # MsgSet Offset struct.pack(">i", 18), # Msg Size struct.pack(">i", 1474775406), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 2), # Length of key - "k1", # Key + b"k1", # Key struct.pack(">i", 2), # Length of value - "v1", # Value + b"v1", # Value struct.pack(">q", 1), # MsgSet Offset struct.pack(">i", 18), # Msg Size struct.pack(">i", -16383415), # CRC struct.pack(">bb", 0, 0), # Magic, flags struct.pack(">i", 2), # Length of key - "k2", # Key + b"k2", # Key struct.pack(">i", 2), # Length of value - "v2", # Value - "@1$%(Y!", # Random padding + b"v2", # Value + b"@1$%(Y!", # Random padding ]) msgs = list(KafkaProtocol._decode_message_set_iter(encoded)) @@ -283,40 +284,40 @@ class TestProtocol(unittest.TestCase): returned_offset2, decoded_message2 = msg2 self.assertEqual(returned_offset1, 0) - self.assertEqual(decoded_message1, create_message("v1", "k1")) + self.assertEqual(decoded_message1, create_message(b"v1", b"k1")) self.assertEqual(returned_offset2, 1) - self.assertEqual(decoded_message2, create_message("v2", "k2")) + self.assertEqual(decoded_message2, create_message(b"v2", b"k2")) def test_encode_produce_request(self): requests = [ - ProduceRequest("topic1", 0, [ - create_message("a"), - create_message("b") + ProduceRequest(b"topic1", 0, [ + create_message(b"a"), + create_message(b"b") ]), - ProduceRequest("topic2", 1, [ - create_message("c") + ProduceRequest(b"topic2", 1, [ + create_message(b"c") ]) ] - msg_a_binary = KafkaProtocol._encode_message(create_message("a")) - msg_b_binary = KafkaProtocol._encode_message(create_message("b")) - msg_c_binary = KafkaProtocol._encode_message(create_message("c")) + msg_a_binary = KafkaProtocol._encode_message(create_message(b"a")) + msg_b_binary = KafkaProtocol._encode_message(create_message(b"b")) + msg_c_binary = KafkaProtocol._encode_message(create_message(b"c")) - header = "".join([ + header = b"".join([ struct.pack('>i', 0x94), # The length of the message overall struct.pack('>h', 0), # Msg Header, Message type = Produce struct.pack('>h', 0), # Msg Header, API version struct.pack('>i', 2), # Msg Header, Correlation ID - struct.pack('>h7s', 7, "client1"), # Msg Header, The client ID + struct.pack('>h7s', 7, b"client1"), # Msg Header, The client ID struct.pack('>h', 2), # Num acks required struct.pack('>i', 100), # Request Timeout struct.pack('>i', 2), # The number of requests ]) total_len = len(msg_a_binary) + len(msg_b_binary) - topic1 = "".join([ - struct.pack('>h6s', 6, 'topic1'), # The topic1 + topic1 = b"".join([ + struct.pack('>h6s', 6, b'topic1'), # The topic1 struct.pack('>i', 1), # One message set struct.pack('>i', 0), # Partition 0 struct.pack('>i', total_len + 24), # Size of the incoming message set @@ -328,8 +329,8 @@ class TestProtocol(unittest.TestCase): msg_b_binary, # Actual message ]) - topic2 = "".join([ - struct.pack('>h6s', 6, 'topic2'), # The topic1 + topic2 = b"".join([ + struct.pack('>h6s', 6, b'topic2'), # The topic1 struct.pack('>i', 1), # One message set struct.pack('>i', 1), # Partition 1 struct.pack('>i', len(msg_c_binary) + 12), # Size of the incoming message set @@ -338,68 +339,72 @@ class TestProtocol(unittest.TestCase): msg_c_binary, # Actual message ]) - expected1 = "".join([ header, topic1, topic2 ]) - expected2 = "".join([ header, topic2, topic1 ]) + expected1 = b"".join([ header, topic1, topic2 ]) + expected2 = b"".join([ header, topic2, topic1 ]) - encoded = KafkaProtocol.encode_produce_request("client1", 2, requests, 2, 100) + encoded = KafkaProtocol.encode_produce_request(b"client1", 2, requests, 2, 100) self.assertIn(encoded, [ expected1, expected2 ]) def test_decode_produce_response(self): - t1 = "topic1" - t2 = "topic2" + t1 = b"topic1" + t2 = b"topic2" + _long = int + if six.PY2: + _long = long encoded = struct.pack('>iih%dsiihqihqh%dsiihq' % (len(t1), len(t2)), - 2, 2, len(t1), t1, 2, 0, 0, 10L, 1, 1, 20L, - len(t2), t2, 1, 0, 0, 30L) + 2, 2, len(t1), t1, 2, 0, 0, _long(10), 1, 1, _long(20), + len(t2), t2, 1, 0, 0, _long(30)) responses = list(KafkaProtocol.decode_produce_response(encoded)) self.assertEqual(responses, - [ProduceResponse(t1, 0, 0, 10L), - ProduceResponse(t1, 1, 1, 20L), - ProduceResponse(t2, 0, 0, 30L)]) + [ProduceResponse(t1, 0, 0, _long(10)), + ProduceResponse(t1, 1, 1, _long(20)), + ProduceResponse(t2, 0, 0, _long(30))]) def test_encode_fetch_request(self): requests = [ - FetchRequest("topic1", 0, 10, 1024), - FetchRequest("topic2", 1, 20, 100), + FetchRequest(b"topic1", 0, 10, 1024), + FetchRequest(b"topic2", 1, 20, 100), ] - header = "".join([ + header = b"".join([ struct.pack('>i', 89), # The length of the message overall struct.pack('>h', 1), # Msg Header, Message type = Fetch struct.pack('>h', 0), # Msg Header, API version struct.pack('>i', 3), # Msg Header, Correlation ID - struct.pack('>h7s', 7, "client1"), # Msg Header, The client ID + struct.pack('>h7s', 7, b"client1"),# Msg Header, The client ID struct.pack('>i', -1), # Replica Id struct.pack('>i', 2), # Max wait time struct.pack('>i', 100), # Min bytes struct.pack('>i', 2), # Num requests ]) - topic1 = "".join([ - struct.pack('>h6s', 6, 'topic1'), # Topic + topic1 = b"".join([ + struct.pack('>h6s', 6, b'topic1'),# Topic struct.pack('>i', 1), # Num Payloads struct.pack('>i', 0), # Partition 0 struct.pack('>q', 10), # Offset struct.pack('>i', 1024), # Max Bytes ]) - topic2 = "".join([ - struct.pack('>h6s', 6, 'topic2'), # Topic + topic2 = b"".join([ + struct.pack('>h6s', 6, b'topic2'),# Topic struct.pack('>i', 1), # Num Payloads struct.pack('>i', 1), # Partition 0 struct.pack('>q', 20), # Offset struct.pack('>i', 100), # Max Bytes ]) - expected1 = "".join([ header, topic1, topic2 ]) - expected2 = "".join([ header, topic2, topic1 ]) + expected1 = b"".join([ header, topic1, topic2 ]) + expected2 = b"".join([ header, topic2, topic1 ]) - encoded = KafkaProtocol.encode_fetch_request("client1", 3, requests, 2, 100) + encoded = KafkaProtocol.encode_fetch_request(b"client1", 3, requests, 2, 100) self.assertIn(encoded, [ expected1, expected2 ]) def test_decode_fetch_response(self): - t1 = "topic1" - t2 = "topic2" - msgs = map(create_message, ["message1", "hi", "boo", "foo", "so fun!"]) + t1 = b"topic1" + t2 = b"topic2" + msgs = [create_message(msg) + for msg in [b"message1", b"hi", b"boo", b"foo", b"so fun!"]] ms1 = KafkaProtocol._encode_message_set([msgs[0], msgs[1]]) ms2 = KafkaProtocol._encode_message_set([msgs[2]]) ms3 = KafkaProtocol._encode_message_set([msgs[3], msgs[4]]) @@ -416,7 +421,7 @@ class TestProtocol(unittest.TestCase): response.error, response.highwaterMark, list(response.messages)) - expanded_responses = map(expand_messages, responses) + expanded_responses = list(map(expand_messages, responses)) expect = [FetchResponse(t1, 0, 0, 10, [OffsetAndMessage(0, msgs[0]), OffsetAndMessage(0, msgs[1])]), FetchResponse(t1, 1, 1, 20, [OffsetAndMessage(0, msgs[2])]), @@ -425,47 +430,47 @@ class TestProtocol(unittest.TestCase): self.assertEqual(expanded_responses, expect) def test_encode_metadata_request_no_topics(self): - expected = "".join([ + expected = b"".join([ struct.pack(">i", 17), # Total length of the request struct.pack('>h', 3), # API key metadata fetch struct.pack('>h', 0), # API version struct.pack('>i', 4), # Correlation ID - struct.pack('>h3s', 3, "cid"), # The client ID + struct.pack('>h3s', 3, b"cid"),# The client ID struct.pack('>i', 0), # No topics, give all the data! ]) - encoded = KafkaProtocol.encode_metadata_request("cid", 4) + encoded = KafkaProtocol.encode_metadata_request(b"cid", 4) self.assertEqual(encoded, expected) def test_encode_metadata_request_with_topics(self): - expected = "".join([ + expected = b"".join([ struct.pack(">i", 25), # Total length of the request struct.pack('>h', 3), # API key metadata fetch struct.pack('>h', 0), # API version struct.pack('>i', 4), # Correlation ID - struct.pack('>h3s', 3, "cid"), # The client ID + struct.pack('>h3s', 3, b"cid"),# The client ID struct.pack('>i', 2), # Number of topics in the request - struct.pack('>h2s', 2, "t1"), # Topic "t1" - struct.pack('>h2s', 2, "t2"), # Topic "t2" + struct.pack('>h2s', 2, b"t1"), # Topic "t1" + struct.pack('>h2s', 2, b"t2"), # Topic "t2" ]) - encoded = KafkaProtocol.encode_metadata_request("cid", 4, ["t1", "t2"]) + encoded = KafkaProtocol.encode_metadata_request(b"cid", 4, [b"t1", b"t2"]) self.assertEqual(encoded, expected) def _create_encoded_metadata_response(self, broker_data, topic_data, topic_errors, partition_errors): encoded = struct.pack('>ii', 3, len(broker_data)) - for node_id, broker in broker_data.iteritems(): + for node_id, broker in six.iteritems(broker_data): encoded += struct.pack('>ih%dsi' % len(broker.host), node_id, len(broker.host), broker.host, broker.port) encoded += struct.pack('>i', len(topic_data)) - for topic, partitions in topic_data.iteritems(): + for topic, partitions in six.iteritems(topic_data): encoded += struct.pack('>hh%dsi' % len(topic), topic_errors[topic], len(topic), topic, len(partitions)) - for partition, metadata in partitions.iteritems(): + for partition, metadata in six.iteritems(partitions): encoded += struct.pack('>hiii', partition_errors[(topic, partition)], partition, metadata.leader, @@ -483,25 +488,25 @@ class TestProtocol(unittest.TestCase): def test_decode_metadata_response(self): node_brokers = { - 0: BrokerMetadata(0, "brokers1.kafka.rdio.com", 1000), - 1: BrokerMetadata(1, "brokers1.kafka.rdio.com", 1001), - 3: BrokerMetadata(3, "brokers2.kafka.rdio.com", 1000) + 0: BrokerMetadata(0, b"brokers1.kafka.rdio.com", 1000), + 1: BrokerMetadata(1, b"brokers1.kafka.rdio.com", 1001), + 3: BrokerMetadata(3, b"brokers2.kafka.rdio.com", 1000) } topic_partitions = { - "topic1": { - 0: PartitionMetadata("topic1", 0, 1, (0, 2), (2,)), - 1: PartitionMetadata("topic1", 1, 3, (0, 1), (0, 1)) + b"topic1": { + 0: PartitionMetadata(b"topic1", 0, 1, (0, 2), (2,)), + 1: PartitionMetadata(b"topic1", 1, 3, (0, 1), (0, 1)) }, - "topic2": { - 0: PartitionMetadata("topic2", 0, 0, (), ()) + b"topic2": { + 0: PartitionMetadata(b"topic2", 0, 0, (), ()) } } - topic_errors = {"topic1": 0, "topic2": 1} + topic_errors = {b"topic1": 0, b"topic2": 1} partition_errors = { - ("topic1", 0): 0, - ("topic1", 1): 1, - ("topic2", 0): 0 + (b"topic1", 0): 0, + (b"topic1", 1): 1, + (b"topic2", 0): 0 } encoded = self._create_encoded_metadata_response(node_brokers, topic_partitions, @@ -511,31 +516,31 @@ class TestProtocol(unittest.TestCase): self.assertEqual(decoded, (node_brokers, topic_partitions)) def test_encode_offset_request(self): - expected = "".join([ + expected = b"".join([ struct.pack(">i", 21), # Total length of the request struct.pack('>h', 2), # Message type = offset fetch struct.pack('>h', 0), # API version struct.pack('>i', 4), # Correlation ID - struct.pack('>h3s', 3, "cid"), # The client ID + struct.pack('>h3s', 3, b"cid"), # The client ID struct.pack('>i', -1), # Replica Id struct.pack('>i', 0), # No topic/partitions ]) - encoded = KafkaProtocol.encode_offset_request("cid", 4) + encoded = KafkaProtocol.encode_offset_request(b"cid", 4) self.assertEqual(encoded, expected) def test_encode_offset_request__no_payload(self): - expected = "".join([ + expected = b"".join([ struct.pack(">i", 65), # Total length of the request struct.pack('>h', 2), # Message type = offset fetch struct.pack('>h', 0), # API version struct.pack('>i', 4), # Correlation ID - struct.pack('>h3s', 3, "cid"), # The client ID + struct.pack('>h3s', 3, b"cid"), # The client ID struct.pack('>i', -1), # Replica Id struct.pack('>i', 1), # Num topics - struct.pack(">h6s", 6, "topic1"), # Topic for the request + struct.pack(">h6s", 6, b"topic1"),# Topic for the request struct.pack(">i", 2), # Two partitions struct.pack(">i", 3), # Partition 3 @@ -547,18 +552,18 @@ class TestProtocol(unittest.TestCase): struct.pack(">i", 1), # One offset requested ]) - encoded = KafkaProtocol.encode_offset_request("cid", 4, [ - OffsetRequest('topic1', 3, -1, 1), - OffsetRequest('topic1', 4, -1, 1), + encoded = KafkaProtocol.encode_offset_request(b"cid", 4, [ + OffsetRequest(b'topic1', 3, -1, 1), + OffsetRequest(b'topic1', 4, -1, 1), ]) self.assertEqual(encoded, expected) def test_decode_offset_response(self): - encoded = "".join([ + encoded = b"".join([ struct.pack(">i", 42), # Correlation ID struct.pack(">i", 1), # One topics - struct.pack(">h6s", 6, "topic1"), # First topic + struct.pack(">h6s", 6, b"topic1"),# First topic struct.pack(">i", 2), # Two partitions struct.pack(">i", 2), # Partition 2 @@ -574,24 +579,24 @@ class TestProtocol(unittest.TestCase): results = KafkaProtocol.decode_offset_response(encoded) self.assertEqual(set(results), set([ - OffsetResponse(topic = 'topic1', partition = 2, error = 0, offsets=(4,)), - OffsetResponse(topic = 'topic1', partition = 4, error = 0, offsets=(8,)), + OffsetResponse(topic = b'topic1', partition = 2, error = 0, offsets=(4,)), + OffsetResponse(topic = b'topic1', partition = 4, error = 0, offsets=(8,)), ])) def test_encode_offset_commit_request(self): - header = "".join([ + header = b"".join([ struct.pack('>i', 99), # Total message length struct.pack('>h', 8), # Message type = offset commit struct.pack('>h', 0), # API version struct.pack('>i', 42), # Correlation ID - struct.pack('>h9s', 9, "client_id"), # The client ID - struct.pack('>h8s', 8, "group_id"), # The group to commit for + struct.pack('>h9s', 9, b"client_id"),# The client ID + struct.pack('>h8s', 8, b"group_id"), # The group to commit for struct.pack('>i', 2), # Num topics ]) - topic1 = "".join([ - struct.pack(">h6s", 6, "topic1"), # Topic for the request + topic1 = b"".join([ + struct.pack(">h6s", 6, b"topic1"), # Topic for the request struct.pack(">i", 2), # Two partitions struct.pack(">i", 0), # Partition 0 struct.pack(">q", 123), # Offset 123 @@ -601,30 +606,30 @@ class TestProtocol(unittest.TestCase): struct.pack(">h", -1), # Null metadata ]) - topic2 = "".join([ - struct.pack(">h6s", 6, "topic2"), # Topic for the request + topic2 = b"".join([ + struct.pack(">h6s", 6, b"topic2"), # Topic for the request struct.pack(">i", 1), # One partition struct.pack(">i", 2), # Partition 2 struct.pack(">q", 345), # Offset 345 struct.pack(">h", -1), # Null metadata ]) - expected1 = "".join([ header, topic1, topic2 ]) - expected2 = "".join([ header, topic2, topic1 ]) + expected1 = b"".join([ header, topic1, topic2 ]) + expected2 = b"".join([ header, topic2, topic1 ]) - encoded = KafkaProtocol.encode_offset_commit_request("client_id", 42, "group_id", [ - OffsetCommitRequest("topic1", 0, 123, None), - OffsetCommitRequest("topic1", 1, 234, None), - OffsetCommitRequest("topic2", 2, 345, None), + encoded = KafkaProtocol.encode_offset_commit_request(b"client_id", 42, b"group_id", [ + OffsetCommitRequest(b"topic1", 0, 123, None), + OffsetCommitRequest(b"topic1", 1, 234, None), + OffsetCommitRequest(b"topic2", 2, 345, None), ]) self.assertIn(encoded, [ expected1, expected2 ]) def test_decode_offset_commit_response(self): - encoded = "".join([ + encoded = b"".join([ struct.pack(">i", 42), # Correlation ID struct.pack(">i", 1), # One topic - struct.pack(">h6s", 6, "topic1"), # First topic + struct.pack(">h6s", 6, b"topic1"),# First topic struct.pack(">i", 2), # Two partitions struct.pack(">i", 2), # Partition 2 @@ -636,82 +641,78 @@ class TestProtocol(unittest.TestCase): results = KafkaProtocol.decode_offset_commit_response(encoded) self.assertEqual(set(results), set([ - OffsetCommitResponse(topic = 'topic1', partition = 2, error = 0), - OffsetCommitResponse(topic = 'topic1', partition = 4, error = 0), + OffsetCommitResponse(topic = b'topic1', partition = 2, error = 0), + OffsetCommitResponse(topic = b'topic1', partition = 4, error = 0), ])) def test_encode_offset_fetch_request(self): - header = "".join([ + header = b"".join([ struct.pack('>i', 69), # Total message length struct.pack('>h', 9), # Message type = offset fetch struct.pack('>h', 0), # API version struct.pack('>i', 42), # Correlation ID - struct.pack('>h9s', 9, "client_id"), # The client ID - struct.pack('>h8s', 8, "group_id"), # The group to commit for + struct.pack('>h9s', 9, b"client_id"),# The client ID + struct.pack('>h8s', 8, b"group_id"), # The group to commit for struct.pack('>i', 2), # Num topics ]) - topic1 = "".join([ - struct.pack(">h6s", 6, "topic1"), # Topic for the request + topic1 = b"".join([ + struct.pack(">h6s", 6, b"topic1"), # Topic for the request struct.pack(">i", 2), # Two partitions struct.pack(">i", 0), # Partition 0 struct.pack(">i", 1), # Partition 1 ]) - topic2 = "".join([ - struct.pack(">h6s", 6, "topic2"), # Topic for the request + topic2 = b"".join([ + struct.pack(">h6s", 6, b"topic2"), # Topic for the request struct.pack(">i", 1), # One partitions struct.pack(">i", 2), # Partition 2 ]) - expected1 = "".join([ header, topic1, topic2 ]) - expected2 = "".join([ header, topic2, topic1 ]) + expected1 = b"".join([ header, topic1, topic2 ]) + expected2 = b"".join([ header, topic2, topic1 ]) - encoded = KafkaProtocol.encode_offset_fetch_request("client_id", 42, "group_id", [ - OffsetFetchRequest("topic1", 0), - OffsetFetchRequest("topic1", 1), - OffsetFetchRequest("topic2", 2), + encoded = KafkaProtocol.encode_offset_fetch_request(b"client_id", 42, b"group_id", [ + OffsetFetchRequest(b"topic1", 0), + OffsetFetchRequest(b"topic1", 1), + OffsetFetchRequest(b"topic2", 2), ]) self.assertIn(encoded, [ expected1, expected2 ]) def test_decode_offset_fetch_response(self): - encoded = "".join([ + encoded = b"".join([ struct.pack(">i", 42), # Correlation ID struct.pack(">i", 1), # One topics - struct.pack(">h6s", 6, "topic1"), # First topic + struct.pack(">h6s", 6, b"topic1"),# First topic struct.pack(">i", 2), # Two partitions struct.pack(">i", 2), # Partition 2 struct.pack(">q", 4), # Offset 4 - struct.pack(">h4s", 4, "meta"), # Metadata + struct.pack(">h4s", 4, b"meta"), # Metadata struct.pack(">h", 0), # No error struct.pack(">i", 4), # Partition 4 struct.pack(">q", 8), # Offset 8 - struct.pack(">h4s", 4, "meta"), # Metadata + struct.pack(">h4s", 4, b"meta"), # Metadata struct.pack(">h", 0), # No error ]) results = KafkaProtocol.decode_offset_fetch_response(encoded) self.assertEqual(set(results), set([ - OffsetFetchResponse(topic = 'topic1', partition = 2, offset = 4, error = 0, metadata = "meta"), - OffsetFetchResponse(topic = 'topic1', partition = 4, offset = 8, error = 0, metadata = "meta"), + OffsetFetchResponse(topic = b'topic1', partition = 2, offset = 4, error = 0, metadata = b"meta"), + OffsetFetchResponse(topic = b'topic1', partition = 4, offset = 8, error = 0, metadata = b"meta"), ])) @contextmanager def mock_create_message_fns(self): - patches = contextlib.nested( - mock.patch.object(kafka.protocol, "create_message", - return_value=sentinel.message), - mock.patch.object(kafka.protocol, "create_gzip_message", - return_value=sentinel.gzip_message), - mock.patch.object(kafka.protocol, "create_snappy_message", - return_value=sentinel.snappy_message), - ) - - with patches: - yield + with mock.patch.object(kafka.protocol, "create_message", + return_value=sentinel.message): + with mock.patch.object(kafka.protocol, "create_gzip_message", + return_value=sentinel.gzip_message): + with mock.patch.object(kafka.protocol, "create_snappy_message", + return_value=sentinel.snappy_message): + yield def test_create_message_set(self): messages = [1, 2, 3] -- cgit v1.2.1