Mercurial > hg
comparison tests/test-cbor.py @ 39413:babad5ebaf0a
cborutil: add a buffering decoder
The sansiodecoder leaves it up to the callers to feed in data that
wasn't fully consumed last time.
This commit implements a decoder that performs buffering of
leftover chunks from the previous invocation. It otherwise
behaves identically to sansiodecoder.
Differential Revision: https://phab.mercurial-scm.org/D4434
author | Gregory Szorc <gregory.szorc@gmail.com> |
---|---|
date | Wed, 29 Aug 2018 14:29:01 -0700 |
parents | a40d3da89b7d |
children | b638219a23c3 |
comparison
equal
deleted
inserted
replaced
39412:a40d3da89b7d | 39413:babad5ebaf0a |
---|---|
939 class SansIODecoderTests(TestCase): | 939 class SansIODecoderTests(TestCase): |
940 def testemptyinput(self): | 940 def testemptyinput(self): |
941 decoder = cborutil.sansiodecoder() | 941 decoder = cborutil.sansiodecoder() |
942 self.assertEqual(decoder.decode(b''), (False, 0, 0)) | 942 self.assertEqual(decoder.decode(b''), (False, 0, 0)) |
943 | 943 |
944 class BufferingDecoderTests(TestCase): | |
945 def testsimple(self): | |
946 source = [ | |
947 b'foobar', | |
948 b'x' * 128, | |
949 {b'foo': b'bar'}, | |
950 True, | |
951 False, | |
952 None, | |
953 [None for i in range(128)], | |
954 ] | |
955 | |
956 encoded = b''.join(cborutil.streamencode(source)) | |
957 | |
958 for step in range(1, 32): | |
959 decoder = cborutil.bufferingdecoder() | |
960 start = 0 | |
961 | |
962 while start < len(encoded): | |
963 decoder.decode(encoded[start:start + step]) | |
964 start += step | |
965 | |
966 self.assertEqual(decoder.getavailable(), [source]) | |
967 | |
944 class DecodeallTests(TestCase): | 968 class DecodeallTests(TestCase): |
945 def testemptyinput(self): | 969 def testemptyinput(self): |
946 self.assertEqual(cborutil.decodeall(b''), []) | 970 self.assertEqual(cborutil.decodeall(b''), []) |
947 | 971 |
948 def testpartialinput(self): | 972 def testpartialinput(self): |