( )⚙ D4433 cborutil: remove readindefinitebytestringtoiter()

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4433

cborutil: remove readindefinitebytestringtoiter()
ClosedPublic

Authored by indygreg on Aug 31 2018, 6:57 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGa40d3da89b7d: cborutil: remove readindefinitebytestringtoiter()
Summary

This was implemented as part of implementing streaming encoding.
It was never used outside of tests.

Now that we have a full CBOR decoder, it can be used for incremental
decoding of indefinite-length byte strings.

This also removes the last use of the vendored cbor2 package from this
module.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Aug 31 2018, 6:57 PM
Closed by commit rHGa40d3da89b7d: cborutil: remove readindefinitebytestringtoiter() (authored by indygreg). · Explain WhySep 3 2018, 9:38 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/utils/cborutil.py (52 lines)
Mtests/test-cbor.py (11 lines)

Diff 10690

mercurial/utils/cborutil.py

# cborutil.py - CBOR extensions # cborutil.py - CBOR extensions
# #
# Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com> # Copyright 2018 Gregory Szorc <gregory.szorc@gmail.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import struct import struct
import sys import sys
from ..thirdparty.cbor.cbor2 import (
decoder as decodermod,
)
# Very short very of RFC 7049... # Very short very of RFC 7049...
# #
# Each item begins with a byte. The 3 high bits of that byte denote the # Each item begins with a byte. The 3 high bits of that byte denote the
# "major type." The lower 5 bits denote the "subtype." Each major type # "major type." The lower 5 bits denote the "subtype." Each major type
# has its own encoding mechanism. # has its own encoding mechanism.
# #
# Most types have lengths. However, bytestring, string, array, and map # Most types have lengths. However, bytestring, string, array, and map
# can be indefinite length. These are denotes by a subtype with value 31. # can be indefinite length. These are denotes by a subtype with value 31.
""" """
fn = STREAM_ENCODERS.get(v.__class__) fn = STREAM_ENCODERS.get(v.__class__)
if not fn: if not fn:
raise ValueError('do not know how to encode %s' % type(v)) raise ValueError('do not know how to encode %s' % type(v))
return fn(v) return fn(v)
def readindefinitebytestringtoiter(fh, expectheader=True):
"""Read an indefinite bytestring to a generator.
Receives an object with a ``read(X)`` method to read N bytes.
If ``expectheader`` is True, it is expected that the first byte read
will represent an indefinite length bytestring. Otherwise, we
expect the first byte to be part of the first bytestring chunk.
"""
read = fh.read
decodeuint = decodermod.decode_uint
byteasinteger = decodermod.byte_as_integer
if expectheader:
initial = decodermod.byte_as_integer(read(1))
majortype = initial >> 5
subtype = initial & SUBTYPE_MASK
if majortype != MAJOR_TYPE_BYTESTRING:
raise decodermod.CBORDecodeError(
'expected major type %d; got %d' % (MAJOR_TYPE_BYTESTRING,
majortype))
if subtype != SUBTYPE_INDEFINITE:
raise decodermod.CBORDecodeError(
'expected indefinite subtype; got %d' % subtype)
# The indefinite bytestring is composed of chunks of normal bytestrings.
# Read chunks until we hit a BREAK byte.
while True:
# We need to sniff for the BREAK byte.
initial = byteasinteger(read(1))
if initial == BREAK_INT:
break
length = decodeuint(fh, initial & SUBTYPE_MASK)
chunk = read(length)
if len(chunk) != length:
raise decodermod.CBORDecodeError(
'failed to read bytestring chunk: got %d bytes; expected %d' % (
len(chunk), length))
yield chunk
class CBORDecodeError(Exception): class CBORDecodeError(Exception):
"""Represents an error decoding CBOR.""" """Represents an error decoding CBOR."""
if sys.version_info.major >= 3: if sys.version_info.major >= 3:
def _elementtointeger(b, i): def _elementtointeger(b, i):
return b[i] return b[i]
else: else:
def _elementtointeger(b, i): def _elementtointeger(b, i):

tests/test-cbor.py

from __future__ import absolute_import from __future__ import absolute_import
import io
import unittest import unittest
from mercurial.thirdparty import ( from mercurial.thirdparty import (
cbor, cbor,
) )
from mercurial.utils import ( from mercurial.utils import (
cborutil, cborutil,
) )
self.assertIn(len(chunk), (0, 8, 42)) self.assertIn(len(chunk), (0, 8, 42))
encoded = b'\x5f\xff' encoded = b'\x5f\xff'
b = cborutil.decodeall(encoded) b = cborutil.decodeall(encoded)
self.assertEqual(b, [b'']) self.assertEqual(b, [b''])
self.assertTrue(b[0].isfirst) self.assertTrue(b[0].isfirst)
self.assertTrue(b[0].islast) self.assertTrue(b[0].islast)
def testreadtoiter(self):
source = io.BytesIO(b'\x5f\x44\xaa\xbb\xcc\xdd\x43\xee\xff\x99\xff')
it = cborutil.readindefinitebytestringtoiter(source)
self.assertEqual(next(it), b'\xaa\xbb\xcc\xdd')
self.assertEqual(next(it), b'\xee\xff\x99')
with self.assertRaises(StopIteration):
next(it)
def testdecodevariouslengths(self): def testdecodevariouslengths(self):
for i in (0, 1, 22, 23, 24, 25, 254, 255, 256, 65534, 65535, 65536): for i in (0, 1, 22, 23, 24, 25, 254, 255, 256, 65534, 65535, 65536):
source = b'x' * i source = b'x' * i
encoded = b''.join(cborutil.streamencode(source)) encoded = b''.join(cborutil.streamencode(source))
if len(source) < 24: if len(source) < 24:
hlen = 1 hlen = 1
elif len(source) < 256: elif len(source) < 256: