THRIFT-335. python: Initial implementation of TCompactProtocol
Seems to work. No interoperability testing with other languages yet.
git-svn-id: https://svn.apache.org/repos/asf/incubator/thrift/trunk@1001827 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/test/py/TestEof.py b/test/py/TestEof.py
index 7d64289..7ff0b42 100755
--- a/test/py/TestEof.py
+++ b/test/py/TestEof.py
@@ -28,14 +28,18 @@
from thrift.transport import TTransport
from thrift.transport import TSocket
from thrift.protocol import TBinaryProtocol
+from thrift.protocol import TCompactProtocol
import unittest
import time
class TestEof(unittest.TestCase):
- def setUp(self):
+ def make_data(self, pfactory=None):
trans = TTransport.TMemoryBuffer()
- prot = TBinaryProtocol.TBinaryProtocol(trans)
+ if pfactory:
+ prot = pfactory.getProtocol(trans)
+ else:
+ prot = TBinaryProtocol.TBinaryProtocol(trans)
x = Xtruct()
x.string_thing = "Zero"
@@ -49,11 +53,11 @@
x.write(prot)
- self.data = trans.getvalue()
+ return trans.getvalue()
def testTransportReadAll(self):
"""Test that readAll on any type of transport throws an EOFError"""
- trans = TTransport.TMemoryBuffer(self.data)
+ trans = TTransport.TMemoryBuffer(self.make_data())
trans.readAll(1)
try:
@@ -64,7 +68,7 @@
self.fail("Should have gotten EOFError")
def eofTestHelper(self, pfactory):
- trans = TTransport.TMemoryBuffer(self.data)
+ trans = TTransport.TMemoryBuffer(self.make_data(pfactory))
prot = pfactory.getProtocol(trans)
x = Xtruct()
@@ -89,8 +93,9 @@
"""Teest the ability of TBinaryProtocol to deal with the removal of every byte in the file"""
# TODO: we should make sure this covers more of the code paths
- for i in xrange(0, len(self.data) + 1):
- trans = TTransport.TMemoryBuffer(self.data[0:i])
+ data = self.make_data(pfactory)
+ for i in xrange(0, len(data) + 1):
+ trans = TTransport.TMemoryBuffer(data[0:i])
prot = pfactory.getProtocol(trans)
try:
x = Xtruct()
@@ -111,6 +116,11 @@
self.eofTestHelper(TBinaryProtocol.TBinaryProtocolAcceleratedFactory())
self.eofTestHelperStress(TBinaryProtocol.TBinaryProtocolAcceleratedFactory())
+ def testCompactProtocolEof(self):
+ """Test that TCompactProtocol throws an EOFError when it reaches the end of the stream"""
+ self.eofTestHelper(TCompactProtocol.TCompactProtocolFactory())
+ self.eofTestHelperStress(TCompactProtocol.TCompactProtocolFactory())
+
def suite():
suite = unittest.TestSuite()
loader = unittest.TestLoader()