( )⚙ 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
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

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 10729

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: