view tests/test-cbor.py @ 41762:cde37ed080c9

tests: fixed test too dependent on actual exception wording

On one of the machines I use to run the tests prior to submission,
the default Python is 2.7.5, with the following wording:

  must be encoded string without NULL bytes, not str

This third form (and possible future ones) are motivation to
use a wider catching regexp.

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', # empty array
            b'\x43foo',
        ])

        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', # array of 2 elements
            b'\x01', # integer 1
        ])

        with self.assertRaisesRegex(cborutil.CBORDecodeError,
                                    'input data not complete'):
            cborutil.decodeall(encoded)

if __name__ == '__main__':
    import silenttestrunner
    silenttestrunner.main(__name__)