diff 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

--- a/tests/test-cbor.py	Fri Aug 31 15:54:17 2018 -0700
+++ b/tests/test-cbor.py	Wed Aug 29 14:29:01 2018 -0700
@@ -941,6 +941,30 @@
         decoder = cborutil.sansiodecoder()
         self.assertEqual(decoder.decode(b''), (False, 0, 0))
 
+class BufferingDecoderTests(TestCase):
+    def testsimple(self):
+        source = [
+            b'foobar',
+            b'x' * 128,
+            {b'foo': b'bar'},
+            True,
+            False,
+            None,
+            [None for i in range(128)],
+        ]
+
+        encoded = b''.join(cborutil.streamencode(source))
+
+        for step in range(1, 32):
+            decoder = cborutil.bufferingdecoder()
+            start = 0
+
+            while start < len(encoded):
+                decoder.decode(encoded[start:start + step])
+                start += step
+
+            self.assertEqual(decoder.getavailable(), [source])
+
 class DecodeallTests(TestCase):
     def testemptyinput(self):
         self.assertEqual(cborutil.decodeall(b''), [])