| hg-reviewers |
This code didn't like passing in bytearray instances. Let's cast
bytearray to bytes so it works.
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| Path | Packages | |||
|---|---|---|---|---|
| M | mercurial/utils/cborutil.py (5 lines) | |||
| M | tests/test-cbor.py (8 lines) |
| """Attempt to decode bytes to CBOR values. | """Attempt to decode bytes to CBOR values. | ||||
| Returns a tuple with the following fields: | Returns a tuple with the following fields: | ||||
| * Bool indicating whether new values are available for retrieval. | * Bool indicating whether new values are available for retrieval. | ||||
| * Integer number of bytes decoded from the new input. | * Integer number of bytes decoded from the new input. | ||||
| * Integer number of bytes wanted to decode the next value. | * Integer number of bytes wanted to decode the next value. | ||||
| """ | """ | ||||
| # We /might/ be able to support passing a bytearray all the | |||||
| # way through. For now, let's cheat. | |||||
| if isinstance(b, bytearray): | |||||
| b = bytes(b) | |||||
| # Our strategy for buffering is to aggregate the incoming chunks in a | # Our strategy for buffering is to aggregate the incoming chunks in a | ||||
| # list until we've received enough data to decode the next item. | # list until we've received enough data to decode the next item. | ||||
| # This is slightly more complicated than using an ``io.BytesIO`` | # This is slightly more complicated than using an ``io.BytesIO`` | ||||
| # or continuously concatenating incoming data. However, because it | # or continuously concatenating incoming data. However, because it | ||||
| # isn't constantly reallocating backing memory for a growing buffer, | # isn't constantly reallocating backing memory for a growing buffer, | ||||
| # it prevents excessive memory thrashing and is significantly faster, | # it prevents excessive memory thrashing and is significantly faster, | ||||
| # especially in cases where the percentage of input chunks that don't | # especially in cases where the percentage of input chunks that don't | ||||
| # decode into a full item is high. | # decode into a full item is high. | ||||
| start = 0 | start = 0 | ||||
| while start < len(encoded): | while start < len(encoded): | ||||
| decoder.decode(encoded[start:start + step]) | decoder.decode(encoded[start:start + step]) | ||||
| start += step | start += step | ||||
| self.assertEqual(decoder.getavailable(), [source]) | self.assertEqual(decoder.getavailable(), [source]) | ||||
| def testbytearray(self): | |||||
| source = b''.join(cborutil.streamencode(b'foobar')) | |||||
| decoder = cborutil.bufferingdecoder() | |||||
| decoder.decode(bytearray(source)) | |||||
| self.assertEqual(decoder.getavailable(), [b'foobar']) | |||||
| class DecodeallTests(TestCase): | class DecodeallTests(TestCase): | ||||
| def testemptyinput(self): | def testemptyinput(self): | ||||
| self.assertEqual(cborutil.decodeall(b''), []) | self.assertEqual(cborutil.decodeall(b''), []) | ||||
| def testpartialinput(self): | def testpartialinput(self): | ||||
| encoded = b''.join([ | encoded = b''.join([ | ||||
| b'\x82', # array of 2 elements | b'\x82', # array of 2 elements | ||||
| b'\x01', # integer 1 | b'\x01', # integer 1 | ||||