from __future__ import absolute_import
import os
import sys
import unittest
# TODO migrate to canned cbor test strings and stop using thirdparty.cbor
tpp = os.path.normpath(
os.path.join(os.path.dirname(__file__), '..', 'mercurial', 'thirdparty')
)
if not os.path.exists(tpp):
# skip, not in a repo
sys.exit(80)
sys.path[0:0] = [tpp]
import cbor
del sys.path[0]
from mercurial.utils import cborutil
class TestCase(unittest.TestCase):
if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
# Python 3.7 deprecates the regex*p* version, but 2.7 lacks
# the regex version.
assertRaisesRegex = ( # camelcase-required
unittest.TestCase.assertRaisesRegexp
)
def loadit(it):
return cbor.loads(b''.join(it))
class BytestringTests(TestCase):
def testsimple(self):
self.assertEqual(
list(cborutil.streamencode(b'foobar')), [b'\x46', b'foobar']
)
self.assertEqual(loadit(cborutil.streamencode(b'foobar')), b'foobar')
self.assertEqual(cborutil.decodeall(b'\x46foobar'), [b'foobar'])
self.assertEqual(
cborutil.decodeall(b'\x46foobar\x45fizbi'), [b'foobar', b'fizbi']
)
def testlong(self):
source = b'x' * 1048576
self.assertEqual(loadit(cborutil.streamencode(source)), source)
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
def testfromiter(self):
# This is the example from RFC 7049 Section 2.2.2.
source = [b'\xaa\xbb\xcc\xdd', b'\xee\xff\x99']
self.assertEqual(
list(cborutil.streamencodebytestringfromiter(source)),
[
b'\x5f',
b'\x44',
b'\xaa\xbb\xcc\xdd',
b'\x43',
b'\xee\xff\x99',
b'\xff',
],
)
self.assertEqual(
loadit(cborutil.streamencodebytestringfromiter(source)),
b''.join(source),
)
self.assertEqual(
cborutil.decodeall(
b'\x5f\x44\xaa\xbb\xcc\xdd' b'\x43\xee\xff\x99\xff'
),
[b'\xaa\xbb\xcc\xdd', b'\xee\xff\x99', b''],
)
for i, chunk in enumerate(
cborutil.decodeall(
b'\x5f\x44\xaa\xbb\xcc\xdd' b'\x43\xee\xff\x99\xff'
)
):
self.assertIsInstance(chunk, cborutil.bytestringchunk)
if i == 0:
self.assertTrue(chunk.isfirst)
else:
self.assertFalse(chunk.isfirst)
if i == 2:
self.assertTrue(chunk.islast)
else:
self.assertFalse(chunk.islast)
def testfromiterlarge(self):
source = [b'a' * 16, b'b' * 128, b'c' * 1024, b'd' * 1048576]
self.assertEqual(
loadit(cborutil.streamencodebytestringfromiter(source)),
b''.join(source),
)
def testindefinite(self):
source = b'\x00\x01\x02\x03' + b'\xff' * 16384
it = cborutil.streamencodeindefinitebytestring(source, chunksize=2)
self.assertEqual(next(it), b'\x5f')
self.assertEqual(next(it), b'\x42')
self.assertEqual(next(it), b'\x00\x01')
self.assertEqual(next(it), b'\x42')
self.assertEqual(next(it), b'\x02\x03')
self.assertEqual(next(it), b'\x42')
self.assertEqual(next(it), b'\xff\xff')
dest = b''.join(
cborutil.streamencodeindefinitebytestring(source, chunksize=42)
)
self.assertEqual(cbor.loads(dest), source)
self.assertEqual(b''.join(cborutil.decodeall(dest)), source)
for chunk in cborutil.decodeall(dest):
self.assertIsInstance(chunk, cborutil.bytestringchunk)
self.assertIn(len(chunk), (0, 8, 42))
encoded = b'\x5f\xff'
b = cborutil.decodeall(encoded)
self.assertEqual(b, [b''])
self.assertTrue(b[0].isfirst)
self.assertTrue(b[0].islast)
def testdecodevariouslengths(self):
for i in (0, 1, 22, 23, 24, 25, 254, 255, 256, 65534, 65535, 65536):
source = b'x' * i
encoded = b''.join(cborutil.streamencode(source))
if len(source) < 24:
hlen = 1
elif len(source) < 256:
hlen = 2
elif len(source) < 65536:
hlen = 3
elif len(source) < 1048576:
hlen = 5
self.assertEqual(
cborutil.decodeitem(encoded),
(True, source, hlen + len(source), cborutil.SPECIAL_NONE),
)
def testpartialdecode(self):
encoded = b''.join(cborutil.streamencode(b'foobar'))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -6, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -5, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:6]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:7]),
(True, b'foobar', 7, cborutil.SPECIAL_NONE),
)
def testpartialdecodevariouslengths(self):
lens = [
2,
3,
10,
23,
24,
25,
31,
100,
254,
255,
256,
257,
16384,
65534,
65535,
65536,
65537,
131071,
131072,
131073,
1048575,
1048576,
1048577,
]
for size in lens:
if size < 24:
hlen = 1
elif size < 2 ** 8:
hlen = 2
elif size < 2 ** 16:
hlen = 3
elif size < 2 ** 32:
hlen = 5
else:
assert False
source = b'x' * size
encoded = b''.join(cborutil.streamencode(source))
res = cborutil.decodeitem(encoded[0:1])
if hlen > 1:
self.assertEqual(
res, (False, None, -(hlen - 1), cborutil.SPECIAL_NONE)
)
else:
self.assertEqual(
res,
(False, None, -(size + hlen - 1), cborutil.SPECIAL_NONE),
)
# Decoding partial header reports remaining header size.
for i in range(hlen - 1):
self.assertEqual(
cborutil.decodeitem(encoded[0 : i + 1]),
(False, None, -(hlen - i - 1), cborutil.SPECIAL_NONE),
)
# Decoding complete header reports item size.
self.assertEqual(
cborutil.decodeitem(encoded[0:hlen]),
(False, None, -size, cborutil.SPECIAL_NONE),
)
# Decoding single byte after header reports item size - 1
self.assertEqual(
cborutil.decodeitem(encoded[0 : hlen + 1]),
(False, None, -(size - 1), cborutil.SPECIAL_NONE),
)
# Decoding all but the last byte reports -1 needed.
self.assertEqual(
cborutil.decodeitem(encoded[0 : hlen + size - 1]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
# Decoding last byte retrieves value.
self.assertEqual(
cborutil.decodeitem(encoded[0 : hlen + size]),
(True, source, hlen + size, cborutil.SPECIAL_NONE),
)
def testindefinitepartialdecode(self):
encoded = b''.join(
cborutil.streamencodebytestringfromiter([b'foobar', b'biz'])
)
# First item should be begin of bytestring special.
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(True, None, 1, cborutil.SPECIAL_START_INDEFINITE_BYTESTRING),
)
# Second item should be the first chunk. But only available when
# we give it 7 bytes (1 byte header + 6 byte chunk).
self.assertEqual(
cborutil.decodeitem(encoded[1:2]),
(False, None, -6, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[1:3]),
(False, None, -5, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[1:4]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[1:5]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[1:6]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[1:7]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[1:8]),
(True, b'foobar', 7, cborutil.SPECIAL_NONE),
)
# Third item should be second chunk. But only available when
# we give it 4 bytes (1 byte header + 3 byte chunk).
self.assertEqual(
cborutil.decodeitem(encoded[8:9]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[8:10]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[8:11]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[8:12]),
(True, b'biz', 4, cborutil.SPECIAL_NONE),
)
# Fourth item should be end of indefinite stream marker.
self.assertEqual(
cborutil.decodeitem(encoded[12:13]),
(True, None, 1, cborutil.SPECIAL_INDEFINITE_BREAK),
)
# Now test the behavior when going through the decoder.
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:1]), (False, 1, 0)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:2]), (False, 1, 6)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:3]), (False, 1, 5)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:4]), (False, 1, 4)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:5]), (False, 1, 3)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:6]), (False, 1, 2)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:7]), (False, 1, 1)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:8]), (True, 8, 0)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:9]), (True, 8, 3)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:10]), (True, 8, 2)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:11]), (True, 8, 1)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:12]), (True, 12, 0)
)
self.assertEqual(
cborutil.sansiodecoder().decode(encoded[0:13]), (True, 13, 0)
)
decoder = cborutil.sansiodecoder()
decoder.decode(encoded[0:8])
values = decoder.getavailable()
self.assertEqual(values, [b'foobar'])
self.assertTrue(values[0].isfirst)
self.assertFalse(values[0].islast)
self.assertEqual(decoder.decode(encoded[8:12]), (True, 4, 0))
values = decoder.getavailable()
self.assertEqual(values, [b'biz'])
self.assertFalse(values[0].isfirst)
self.assertFalse(values[0].islast)
self.assertEqual(decoder.decode(encoded[12:]), (True, 1, 0))
values = decoder.getavailable()
self.assertEqual(values, [b''])
self.assertFalse(values[0].isfirst)
self.assertTrue(values[0].islast)
class StringTests(TestCase):
def testdecodeforbidden(self):
encoded = b'\x63foo'
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'string major type not supported'
):
cborutil.decodeall(encoded)
class IntTests(TestCase):
def testsmall(self):
self.assertEqual(list(cborutil.streamencode(0)), [b'\x00'])
self.assertEqual(cborutil.decodeall(b'\x00'), [0])
self.assertEqual(list(cborutil.streamencode(1)), [b'\x01'])
self.assertEqual(cborutil.decodeall(b'\x01'), [1])
self.assertEqual(list(cborutil.streamencode(2)), [b'\x02'])
self.assertEqual(cborutil.decodeall(b'\x02'), [2])
self.assertEqual(list(cborutil.streamencode(3)), [b'\x03'])
self.assertEqual(cborutil.decodeall(b'\x03'), [3])
self.assertEqual(list(cborutil.streamencode(4)), [b'\x04'])
self.assertEqual(cborutil.decodeall(b'\x04'), [4])
# Multiple value decode works.
self.assertEqual(
cborutil.decodeall(b'\x00\x01\x02\x03\x04'), [0, 1, 2, 3, 4]
)
def testnegativesmall(self):
self.assertEqual(list(cborutil.streamencode(-1)), [b'\x20'])
self.assertEqual(cborutil.decodeall(b'\x20'), [-1])
self.assertEqual(list(cborutil.streamencode(-2)), [b'\x21'])
self.assertEqual(cborutil.decodeall(b'\x21'), [-2])
self.assertEqual(list(cborutil.streamencode(-3)), [b'\x22'])
self.assertEqual(cborutil.decodeall(b'\x22'), [-3])
self.assertEqual(list(cborutil.streamencode(-4)), [b'\x23'])
self.assertEqual(cborutil.decodeall(b'\x23'), [-4])
self.assertEqual(list(cborutil.streamencode(-5)), [b'\x24'])
self.assertEqual(cborutil.decodeall(b'\x24'), [-5])
# Multiple value decode works.
self.assertEqual(
cborutil.decodeall(b'\x20\x21\x22\x23\x24'), [-1, -2, -3, -4, -5]
)
def testrange(self):
for i in range(-70000, 70000, 10):
encoded = b''.join(cborutil.streamencode(i))
self.assertEqual(encoded, cbor.dumps(i))
self.assertEqual(cborutil.decodeall(encoded), [i])
def testdecodepartialubyte(self):
encoded = b''.join(cborutil.streamencode(250))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, 250, 2, cborutil.SPECIAL_NONE),
)
def testdecodepartialbyte(self):
encoded = b''.join(cborutil.streamencode(-42))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, -42, 2, cborutil.SPECIAL_NONE),
)
def testdecodepartialushort(self):
encoded = b''.join(cborutil.streamencode(2 ** 15))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(True, 2 ** 15, 3, cborutil.SPECIAL_NONE),
)
def testdecodepartialshort(self):
encoded = b''.join(cborutil.streamencode(-1024))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(True, -1024, 3, cborutil.SPECIAL_NONE),
)
def testdecodepartialulong(self):
encoded = b''.join(cborutil.streamencode(2 ** 28))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(True, 2 ** 28, 5, cborutil.SPECIAL_NONE),
)
def testdecodepartiallong(self):
encoded = b''.join(cborutil.streamencode(-1048580))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(True, -1048580, 5, cborutil.SPECIAL_NONE),
)
def testdecodepartialulonglong(self):
encoded = b''.join(cborutil.streamencode(2 ** 32))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -8, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -7, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -6, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -5, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:6]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:7]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:8]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:9]),
(True, 2 ** 32, 9, cborutil.SPECIAL_NONE),
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'input data not fully consumed'
):
cborutil.decodeall(encoded[0:1])
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'input data not fully consumed'
):
cborutil.decodeall(encoded[0:2])
def testdecodepartiallonglong(self):
encoded = b''.join(cborutil.streamencode(-7000000000))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -8, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -7, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -6, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -5, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:6]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:7]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:8]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:9]),
(True, -7000000000, 9, cborutil.SPECIAL_NONE),
)
class ArrayTests(TestCase):
def testempty(self):
self.assertEqual(list(cborutil.streamencode([])), [b'\x80'])
self.assertEqual(loadit(cborutil.streamencode([])), [])
self.assertEqual(cborutil.decodeall(b'\x80'), [[]])
def testbasic(self):
source = [b'foo', b'bar', 1, -10]
chunks = [b'\x84', b'\x43', b'foo', b'\x43', b'bar', b'\x01', b'\x29']
self.assertEqual(list(cborutil.streamencode(source)), chunks)
self.assertEqual(cborutil.decodeall(b''.join(chunks)), [source])
def testemptyfromiter(self):
self.assertEqual(
b''.join(cborutil.streamencodearrayfromiter([])), b'\x9f\xff'
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'indefinite length uint not allowed'
):
cborutil.decodeall(b'\x9f\xff')
def testfromiter1(self):
source = [b'foo']
self.assertEqual(
list(cborutil.streamencodearrayfromiter(source)),
[b'\x9f', b'\x43', b'foo', b'\xff',],
)
dest = b''.join(cborutil.streamencodearrayfromiter(source))
self.assertEqual(cbor.loads(dest), source)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'indefinite length uint not allowed'
):
cborutil.decodeall(dest)
def testtuple(self):
source = (b'foo', None, 42)
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cbor.loads(encoded), list(source))
self.assertEqual(cborutil.decodeall(encoded), [list(source)])
def testpartialdecode(self):
source = list(range(4))
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(True, 4, 1, cborutil.SPECIAL_START_ARRAY),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, 4, 1, cborutil.SPECIAL_START_ARRAY),
)
source = list(range(23))
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(True, 23, 1, cborutil.SPECIAL_START_ARRAY),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, 23, 1, cborutil.SPECIAL_START_ARRAY),
)
source = list(range(24))
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, 24, 2, cborutil.SPECIAL_START_ARRAY),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(True, 24, 2, cborutil.SPECIAL_START_ARRAY),
)
source = list(range(256))
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(True, 256, 3, cborutil.SPECIAL_START_ARRAY),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(True, 256, 3, cborutil.SPECIAL_START_ARRAY),
)
def testnested(self):
source = [[], [], [[], [], []]]
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
source = [True, None, [True, 0, 2], [None], [], [[[]], -87]]
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
# A set within an array.
source = [None, {b'foo', b'bar', None, False}, set()]
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
# A map within an array.
source = [None, {}, {b'foo': b'bar', True: False}, [{}]]
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
def testindefinitebytestringvalues(self):
# Single value array whose value is an empty indefinite bytestring.
encoded = b'\x81\x5f\x40\xff'
with self.assertRaisesRegex(
cborutil.CBORDecodeError,
'indefinite length bytestrings not ' 'allowed as array values',
):
cborutil.decodeall(encoded)
class SetTests(TestCase):
def testempty(self):
self.assertEqual(
list(cborutil.streamencode(set())), [b'\xd9\x01\x02', b'\x80',]
)
self.assertEqual(cborutil.decodeall(b'\xd9\x01\x02\x80'), [set()])
def testset(self):
source = {b'foo', None, 42}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cbor.loads(encoded), source)
self.assertEqual(cborutil.decodeall(encoded), [source])
def testinvalidtag(self):
# Must use array to encode sets.
encoded = b'\xd9\x01\x02\xa0'
with self.assertRaisesRegex(
cborutil.CBORDecodeError,
'expected array after finite set ' 'semantic tag',
):
cborutil.decodeall(encoded)
def testpartialdecode(self):
# Semantic tag item will be 3 bytes. Set header will be variable
# depending on length.
encoded = b''.join(cborutil.streamencode({i for i in range(23)}))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(True, 23, 4, cborutil.SPECIAL_START_SET),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(True, 23, 4, cborutil.SPECIAL_START_SET),
)
encoded = b''.join(cborutil.streamencode({i for i in range(24)}))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(True, 24, 5, cborutil.SPECIAL_START_SET),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:6]),
(True, 24, 5, cborutil.SPECIAL_START_SET),
)
encoded = b''.join(cborutil.streamencode({i for i in range(256)}))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:6]),
(True, 256, 6, cborutil.SPECIAL_START_SET),
)
def testinvalidvalue(self):
encoded = b''.join(
[
b'\xd9\x01\x02', # semantic tag
b'\x81', # array of size 1
b'\x5f\x43foo\xff', # indefinite length bytestring "foo"
]
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError,
'indefinite length bytestrings not ' 'allowed as set values',
):
cborutil.decodeall(encoded)
encoded = b''.join([b'\xd9\x01\x02', b'\x81', b'\x80',]) # empty array
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'collections not allowed as set values'
):
cborutil.decodeall(encoded)
encoded = b''.join([b'\xd9\x01\x02', b'\x81', b'\xa0',]) # empty map
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'collections not allowed as set values'
):
cborutil.decodeall(encoded)
encoded = b''.join(
[
b'\xd9\x01\x02',
b'\x81',
b'\xd9\x01\x02\x81\x01', # set with integer 1
]
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'collections not allowed as set values'
):
cborutil.decodeall(encoded)
class BoolTests(TestCase):
def testbasic(self):
self.assertEqual(list(cborutil.streamencode(True)), [b'\xf5'])
self.assertEqual(list(cborutil.streamencode(False)), [b'\xf4'])
self.assertIs(loadit(cborutil.streamencode(True)), True)
self.assertIs(loadit(cborutil.streamencode(False)), False)
self.assertEqual(cborutil.decodeall(b'\xf4'), [False])
self.assertEqual(cborutil.decodeall(b'\xf5'), [True])
self.assertEqual(
cborutil.decodeall(b'\xf4\xf5\xf5\xf4'), [False, True, True, False]
)
class NoneTests(TestCase):
def testbasic(self):
self.assertEqual(list(cborutil.streamencode(None)), [b'\xf6'])
self.assertIs(loadit(cborutil.streamencode(None)), None)
self.assertEqual(cborutil.decodeall(b'\xf6'), [None])
self.assertEqual(cborutil.decodeall(b'\xf6\xf6'), [None, None])
class MapTests(TestCase):
def testempty(self):
self.assertEqual(list(cborutil.streamencode({})), [b'\xa0'])
self.assertEqual(loadit(cborutil.streamencode({})), {})
self.assertEqual(cborutil.decodeall(b'\xa0'), [{}])
def testemptyindefinite(self):
self.assertEqual(
list(cborutil.streamencodemapfromiter([])), [b'\xbf', b'\xff']
)
self.assertEqual(loadit(cborutil.streamencodemapfromiter([])), {})
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'indefinite length uint not allowed'
):
cborutil.decodeall(b'\xbf\xff')
def testone(self):
source = {b'foo': b'bar'}
self.assertEqual(
list(cborutil.streamencode(source)),
[b'\xa1', b'\x43', b'foo', b'\x43', b'bar'],
)
self.assertEqual(loadit(cborutil.streamencode(source)), source)
self.assertEqual(cborutil.decodeall(b'\xa1\x43foo\x43bar'), [source])
def testmultiple(self):
source = {
b'foo': b'bar',
b'baz': b'value1',
}
self.assertEqual(loadit(cborutil.streamencode(source)), source)
self.assertEqual(
loadit(cborutil.streamencodemapfromiter(source.items())), source
)
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
def testcomplex(self):
source = {
b'key': 1,
2: -10,
}
self.assertEqual(loadit(cborutil.streamencode(source)), source)
self.assertEqual(
loadit(cborutil.streamencodemapfromiter(source.items())), source
)
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
def testnested(self):
source = {b'key1': None, b'key2': {b'sub1': b'sub2'}, b'sub2': {}}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
source = {
b'key1': [],
b'key2': [None, False],
b'key3': {b'foo', b'bar'},
b'key4': {},
}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(cborutil.decodeall(encoded), [source])
def testillegalkey(self):
encoded = b''.join(
[
# map header + len 1
b'\xa1',
# indefinite length bytestring "foo" in key position
b'\x5f\x03foo\xff',
]
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError,
'indefinite length bytestrings not ' 'allowed as map keys',
):
cborutil.decodeall(encoded)
encoded = b''.join([b'\xa1', b'\x80', b'\x43foo',]) # empty array
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'collections not supported as map keys'
):
cborutil.decodeall(encoded)
def testillegalvalue(self):
encoded = b''.join(
[
b'\xa1', # map headers
b'\x43foo', # key
b'\x5f\x03bar\xff', # indefinite length value
]
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError,
'indefinite length bytestrings not ' 'allowed as map values',
):
cborutil.decodeall(encoded)
def testpartialdecode(self):
source = {b'key1': b'value1'}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(True, 1, 1, cborutil.SPECIAL_START_MAP),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, 1, 1, cborutil.SPECIAL_START_MAP),
)
source = {b'key%d' % i: None for i in range(23)}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(True, 23, 1, cborutil.SPECIAL_START_MAP),
)
source = {b'key%d' % i: None for i in range(24)}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(True, 24, 2, cborutil.SPECIAL_START_MAP),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(True, 24, 2, cborutil.SPECIAL_START_MAP),
)
source = {b'key%d' % i: None for i in range(256)}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(True, 256, 3, cborutil.SPECIAL_START_MAP),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(True, 256, 3, cborutil.SPECIAL_START_MAP),
)
source = {b'key%d' % i: None for i in range(65536)}
encoded = b''.join(cborutil.streamencode(source))
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -4, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -3, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:3]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:4]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:5]),
(True, 65536, 5, cborutil.SPECIAL_START_MAP),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:6]),
(True, 65536, 5, cborutil.SPECIAL_START_MAP),
)
class SemanticTagTests(TestCase):
def testdecodeforbidden(self):
for i in range(500):
if i == cborutil.SEMANTIC_TAG_FINITE_SET:
continue
tag = cborutil.encodelength(cborutil.MAJOR_TYPE_SEMANTIC, i)
encoded = tag + cborutil.encodelength(cborutil.MAJOR_TYPE_UINT, 42)
# Partial decode is incomplete.
if i < 24:
pass
elif i < 256:
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
elif i < 65536:
self.assertEqual(
cborutil.decodeitem(encoded[0:1]),
(False, None, -2, cborutil.SPECIAL_NONE),
)
self.assertEqual(
cborutil.decodeitem(encoded[0:2]),
(False, None, -1, cborutil.SPECIAL_NONE),
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, r'semantic tag \d+ not allowed'
):
cborutil.decodeitem(encoded)
class SpecialTypesTests(TestCase):
def testforbiddentypes(self):
for i in range(256):
if i == cborutil.SUBTYPE_FALSE:
continue
elif i == cborutil.SUBTYPE_TRUE:
continue
elif i == cborutil.SUBTYPE_NULL:
continue
encoded = cborutil.encodelength(cborutil.MAJOR_TYPE_SPECIAL, i)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, r'special type \d+ not allowed'
):
cborutil.decodeitem(encoded)
class SansIODecoderTests(TestCase):
def testemptyinput(self):
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])
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):
def testemptyinput(self):
self.assertEqual(cborutil.decodeall(b''), [])
def testpartialinput(self):
encoded = b''.join(
[b'\x82', b'\x01',] # array of 2 elements # integer 1
)
with self.assertRaisesRegex(
cborutil.CBORDecodeError, 'input data not complete'
):
cborutil.decodeall(encoded)
if __name__ == '__main__':
import silenttestrunner
silenttestrunner.main(__name__)