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'])