Reinstate test_integrate, make test_protocol more explicit, create testutil
This commit is contained in:
@@ -50,7 +50,7 @@ class KafkaProtocol(object):
|
||||
request_key, # ApiKey
|
||||
0, # ApiVersion
|
||||
correlation_id, # CorrelationId
|
||||
len(client_id),
|
||||
len(client_id), # ClientId size
|
||||
client_id) # ClientId
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -20,27 +20,20 @@ from kafka.codec import (
|
||||
from kafka.protocol import (
|
||||
create_gzip_message, create_message, create_snappy_message, KafkaProtocol
|
||||
)
|
||||
|
||||
ITERATIONS = 1000
|
||||
STRLEN = 100
|
||||
|
||||
|
||||
def random_string():
|
||||
return os.urandom(random.randint(1, STRLEN))
|
||||
|
||||
from .testutil import *
|
||||
|
||||
class TestCodec(unittest.TestCase):
|
||||
@unittest.skipUnless(has_gzip(), "Gzip not available")
|
||||
def test_gzip(self):
|
||||
for i in xrange(ITERATIONS):
|
||||
s1 = random_string()
|
||||
for i in xrange(1000):
|
||||
s1 = random_string(100)
|
||||
s2 = gzip_decode(gzip_encode(s1))
|
||||
self.assertEquals(s1, s2)
|
||||
|
||||
@unittest.skipUnless(has_snappy(), "Snappy not available")
|
||||
def test_snappy(self):
|
||||
for i in xrange(ITERATIONS):
|
||||
s1 = random_string()
|
||||
for i in xrange(1000):
|
||||
s1 = random_string(100)
|
||||
s2 = snappy_decode(snappy_encode(s1))
|
||||
self.assertEquals(s1, s2)
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -86,7 +86,13 @@ class TestProtocol(unittest.TestCase):
|
||||
self.assertEqual(msg.value, expect)
|
||||
|
||||
def test_encode_message_header(self):
|
||||
expect = '\x00\n\x00\x00\x00\x00\x00\x04\x00\x07client3'
|
||||
expect = (
|
||||
"\x00\n" # API Key
|
||||
"\x00\x00" # API Version
|
||||
"\x00\x00\x00\x04" # CorrelationId
|
||||
"\x00\x07" # Client length
|
||||
"client3" # Client Id
|
||||
)
|
||||
encoded = KafkaProtocol._encode_message_header("client3", 4, 10)
|
||||
self.assertEqual(encoded, expect)
|
||||
|
||||
@@ -111,10 +117,27 @@ class TestProtocol(unittest.TestCase):
|
||||
def test_encode_message_set(self):
|
||||
message_set = [create_message("v1", "k1"), create_message("v2", "k2")]
|
||||
encoded = KafkaProtocol._encode_message_set(message_set)
|
||||
expect = ("\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x12W\xe7In\x00"
|
||||
"\x00\x00\x00\x00\x02k1\x00\x00\x00\x02v1\x00\x00\x00\x00"
|
||||
"\x00\x00\x00\x00\x00\x00\x00\x12\xff\x06\x02I\x00\x00\x00"
|
||||
"\x00\x00\x02k2\x00\x00\x00\x02v2")
|
||||
expect = (
|
||||
"\x00\x00\x00\x00\x00\x00\x00\x00" # Msgset1, Offset (Meaningless)
|
||||
"\x00\x00\x00\x12" # Msgset1, Msg Size
|
||||
"\x57\xe7\x49\x6e" # Msg1, CRC
|
||||
"\x00" # Msg1, Magic
|
||||
"\x00" # Msg1, Flags
|
||||
"\x00\x00\x00\x02" # Msg1, key size
|
||||
"k1" # Msg1, key
|
||||
"\x00\x00\x00\x02" # Msg1, value size
|
||||
"v1" # Msg1, value
|
||||
"\x00\x00\x00\x00\x00\x00\x00\x00" # Msgset2, Offset (Meaningless)
|
||||
"\x00\x00\x00\x12" # Msgset2, Msg Size
|
||||
"\xff\x06\x02\x49" # Msg2, CRC
|
||||
"\x00" # Msg2, Magic
|
||||
"\x00" # Msg2, flags
|
||||
"\x00\x00\x00\x02" # Msg2, key size
|
||||
"k2" # Msg2, key
|
||||
"\x00\x00\x00\x02" # Msg2, value size
|
||||
"v2" # MSg2, value
|
||||
)
|
||||
|
||||
self.assertEqual(encoded, expect)
|
||||
|
||||
def test_decode_message(self):
|
||||
|
||||
10
test/testutil.py
Normal file
10
test/testutil.py
Normal file
@@ -0,0 +1,10 @@
|
||||
import os
|
||||
import random
|
||||
import string
|
||||
|
||||
def random_string(l):
|
||||
s = "".join(random.choice(string.letters) for i in xrange(l))
|
||||
return s
|
||||
|
||||
def skip_integration():
|
||||
return os.environ.get('SKIP_INTEGRATION')
|
||||
Reference in New Issue
Block a user