Mercurial > hg
view tests/test-cbor.py @ 43198:c16fe77e340a
pathcopies: give up any optimization based on `introrev`
Between 8a0136f69027 and d98fb3f42f33, we sped up the search for the
introduction revision during path copies. However, further checking show that
finding the introduction revision is still expensive and that we are better off
without it. So we simply drop it and only rely on the linkrev optimisation.
I ran `perfpathcopies` on 6989 pair of revision in the pypy
repository (`hg perfhelper-pathcopies`. The result is massively in favor of
dropping this condition. The result of the copy tracing are unchanged.
Attempt to use a smaller changes preserving linkrev usage were unsuccessful, it
can return wrong result. The following changesets broke test-mv-cp-st-diff.t
- if not f.isintroducedafter(limit):
+ if limit >= 0 and f.linkrev() < limit:
return None
Here are various numbers (before this changeset/after this changesets)
source destination before after saved-time ratio
worth cases e66f24650daf 695dfb0f493b 1.062843 1.246369 -0.183526 1.172675
c979853a3b6a 8d60fe293e79 1.036985 1.196414 -0.159429 1.153743
22349fa2fc33 fbb1c9fd86c0 0.879926 1.038682 -0.158756 1.180420
682b98f3e672 a4878080a536 0.909952 1.063801 -0.153849 1.169074
5adabc9b9848 920958a93997 0.993622 1.147452 -0.153830 1.154817
worse 1% dbfbfcf077e9 aea8f2fd3593 1.016595 1.082999 -0.066404 1.065320
worse 5% c95f1ced15f2 7d29d5e39734 0.453694 0.471156 -0.017462 1.038488
worse 10% 3e144ed1d5b7 2aef0e942480 0.035140 0.037535 -0.002395 1.068156
worse 25% 321fc60db035 801748ba582a 0.009267 0.009325 -0.000058 1.006259
median 2088ce763fc2 e6991321d78b 0.000665 0.000651 0.000014 0.978947
best 25% 915631a97de6 385b31354be6 0.040743 0.040363 0.000380 0.990673
best 10% ad495c36a765 19c10384d3e7 0.431658 0.411490 0.020168 0.953278
best 5% d13ae7d283ae 813c99f810ac 1.141404 1.075346 0.066058 0.942126
best 1% 81593cb4a496 99ae11866969 1.833297 0.063823 1.769474 0.034813
best cases c3b14617fbd7 743a0fcaa4eb 1101.811740 2.735970 1099.075770 0.002483
c3b14617fbd7 9ba6ab77fd29 1116.753953 2.800729 1113.953224 0.002508
058b99d6e81f 57e249b7a3ea 1246.128485 3.042762 1243.085723 0.002442
9a8c361aab49 0354a250d371 1253.111894 3.085796 1250.026098 0.002463
442dbbc53c68 3ec1002a818c 1261.786294 3.138607 1258.647687 0.002487
As one can see, the average case is not really impacted. However, the worth case
we get after this changeset are much better than the one we had before it. We
have 30 pairs where improvements are above 10 minutes.
This reflect in the combined time for all pairs
before: 26256s
after: 1300s (-95%)
If we remove these pathological 30 cases, we still see a significant improvements:
before: 1631s
after: 1245s (-24%)
| author | Pierre-Yves David <pierre-yves.david@octobus.net> |
|---|---|
| date | Thu, 10 Oct 2019 03:49:33 +0200 |
| parents | 2372284d9457 |
| children | 6dbb18e1ac8d |
line wrap: on
line source
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__)