xref: /freebsd/contrib/lib9p/pytest/sequencer.py (revision 134e17798c9af53632b372348ab828e75e65bf46)

#! /usr/bin/env python

from __future__ import print_function

#__all__ = ['EncDec', 'EncDecSimple', 'EncDecTyped', 'EncDecA',
#    'SequenceError', 'Sequencer']

import abc
import struct
import sys

_ProtoStruct = {
    '1': struct.Struct('<B'),
    '2': struct.Struct('<H'),
    '4': struct.Struct('<I'),
    '8': struct.Struct('<Q'),
    '_string_': None,   # handled specially
}
for _i in (1, 2, 4, 8):
    _ProtoStruct[_i] = _ProtoStruct[str(_i)]
del _i

class EncDec(object):
    __metaclass__ = abc.ABCMeta
    """
    Base class for en/de-coders, which are put into sequencers.

    All have a name and arbitrary user-supplied auxiliary data
    (default=None).

    All provide a pack() and unpack().  The pack() function
    returns a "bytes" value.  This is internally implemented as a
    function apack() that returns a list of struct.pack() bytes,
    and pack() just joins them up as needed.

    The pack/unpack functions take a dictionary of variable names
    and values, and a second dictionary for conditionals, but at
    this level conditionals don't apply: they are just being
    passed through.  Variable names do apply to array encoders

    EncDec also provide b2s() and s2b() static methods, which
    convert strings to bytes and vice versa, as reversibly as
    possible (using surrogateescape encoding). In Python2 this is
    a no-op since the string type *is* the bytes type (<type
    'unicode'>) is the unicode-ized string type).

    EncDec also provides b2u() and u2b() to do conversion to/from
    Unicode.

    These are partly for internal use (all strings get converted
    to UTF-8 byte sequences when coding a _string_ type) and partly
    for doctests, where we just want some py2k/py3k compat hacks.
    """
    def __init__(self, name, aux):
        self.name = name
        self.aux = aux

    @staticmethod
    def b2u(byte_sequence):
        "transform bytes to unicode"
        return byte_sequence.decode('utf-8', 'surrogateescape')

    @staticmethod
    def u2b(unicode_sequence):
        "transform unicode to bytes"
        return unicode_sequence.encode('utf-8', 'surrogateescape')

    if sys.version_info[0] >= 3:
        b2s = b2u
        @staticmethod
        def s2b(string):
            "transform string to bytes (leaves raw byte sequence unchanged)"
            if isinstance(string, bytes):
                return string
            return string.encode('utf-8', 'surrogateescape')
    else:
        @staticmethod
        def b2s(byte_sequence):
            "transform bytes to string - no-op in python2.7"
            return byte_sequence
        @staticmethod
        def s2b(string):
            "transform string or unicode to bytes"
            if isinstance(string, unicode):
                return string.encode('utf-8', 'surrogateescape')
            return string

    def pack(self, vdict, cdict, val):
        "encode value <val> into a byte-string"
        return b''.join(self.apack(vdict, cdict, val))

    @abc.abstractmethod
    def apack(self, vdict, cdict, val):
        "encode value <val> into [bytes1, b2, ..., bN]"

    @abc.abstractmethod
    def unpack(self, vdict, cdict, bstring, offset, noerror=False):
        "unpack bytes from <bstring> at <offset>"


class EncDecSimple(EncDec):
    r"""
    Encode/decode a simple (but named) field.  The field is not an
    array, which requires using EncDecA, nor a typed object
    like a qid or stat instance -- those require a Sequence and
    EncDecTyped.

    The format is one of '1'/1, '2'/2, '4'/4, '8'/8, or '_string_'.

    Note: using b2s here is purely a doctest/tetsmod python2/python3
    compat hack.  The output of e.pack is <type 'bytes'>; b2s
    converts it to a string, purely for display purposes.  (It might
    be better to map py2 output to bytes but they just print as a
    string anyway.)  In normal use, you should not call b2s here.

    >>> e = EncDecSimple('eggs', 2)
    >>> e.b2s(e.pack({}, {}, 0))
    '\x00\x00'
    >>> e.b2s(e.pack({}, {}, 256))
    '\x00\x01'

    Values that cannot be packed produce a SequenceError:

    >>> e.pack({}, {}, None)
    Traceback (most recent call last):
        ...
    SequenceError: failed while packing 'eggs'=None
    >>> e.pack({}, {}, -1)
    Traceback (most recent call last):
        ...
    SequenceError: failed while packing 'eggs'=-1

    Unpacking both returns a value, and tells how many bytes it
    used out of the bytestring or byte-array argument.  If there
    are not enough bytes remaining at the starting offset, it
    raises a SequenceError, unless noerror=True (then unset
    values are None)

    >>> e.unpack({}, {}, b'\x00\x01', 0)
    (256, 2)
    >>> e.unpack({}, {}, b'', 0)
    Traceback (most recent call last):
        ...
    SequenceError: out of data while unpacking 'eggs'
    >>> e.unpack({}, {}, b'', 0, noerror=True)
    (None, 2)

    Note that strings can be provided as regular strings, byte
    strings (same as regular strings in py2k), or Unicode strings
    (same as regular strings in py3k).  Unicode strings will be
    converted to UTF-8 before being packed.  Since this leaves
    7-bit characters alone, these examples work in both py2k and
    py3k.  (Note: the UTF-8 encoding of u'\u1234' is
    '\0xe1\0x88\0xb4' or 225, 136, 180. The b2i trick below is
    another py2k vs py3k special case just for doctests: py2k
    tries to display the utf-8 encoded data as a string.)

    >>> e = EncDecSimple('spam', '_string_')
    >>> e.b2s(e.pack({}, {}, 'p3=unicode,p2=bytes'))
    '\x13\x00p3=unicode,p2=bytes'

    >>> e.b2s(e.pack({}, {}, b'bytes'))
    '\x05\x00bytes'

    >>> import sys
    >>> ispy3k = sys.version_info[0] >= 3

    >>> b2i = lambda x: x if ispy3k else ord(x)
    >>> [b2i(x) for x in e.pack({}, {}, u'\u1234')]
    [3, 0, 225, 136, 180]

    The byte length of the utf-8 data cannot exceed 65535 since
    the encoding has the length as a 2-byte field (a la the
    encoding for 'eggs' here).  A too-long string produces
    a SequenceError as well.

    >>> e.pack({}, {}, 16384 * 'spam')
    Traceback (most recent call last):
        ...
    SequenceError: string too long (len=65536) while packing 'spam'

    Unpacking strings produces byte arrays.  (Of course,
    in py2k these are also known as <type 'str'>.)

    >>> unpacked = e.unpack({}, {}, b'\x04\x00data', 0)
    >>> etype = bytes if ispy3k else str
    >>> print(isinstance(unpacked[0], etype))
    True
    >>> e.b2s(unpacked[0])
    'data'
    >>> unpacked[1]
    6

    You may use e.b2s() to conver them to unicode strings in py3k,
    or you may set e.autob2s.  This still only really does
    anything in py3k, since py2k strings *are* bytes, so it's
    really just intended for doctest purposes (see EncDecA):

    >>> e.autob2s = True
    >>> e.unpack({}, {}, b'\x07\x00stringy', 0)
    ('stringy', 9)
    """
    def __init__(self, name, fmt, aux=None):
        super(EncDecSimple, self).__init__(name, aux)
        self.fmt = fmt
        self.struct = _ProtoStruct[fmt]
        self.autob2s = False

    def __repr__(self):
        if self.aux is None:
            return '{0}({1!r}, {2!r})'.format(self.__class__.__name__,
                self.name, self.fmt)
        return '{0}({1!r}, {2!r}, {3!r})'.format(self.__class__.__name__,
            self.name, self.fmt, self.aux)

    __str__ = __repr__

    def apack(self, vdict, cdict, val):
        "encode a value"
        try:
            if self.struct:
                return [self.struct.pack(val)]
            sval = self.s2b(val)
            if len(sval) > 65535:
                raise SequenceError('string too long (len={0:d}) '
                    'while packing {1!r}'.format(len(sval), self.name))
            return [EncDecSimple.string_len.pack(len(sval)), sval]
        # Include AttributeError in case someone tries to, e.g.,
        # pack name=None and self.s2b() tries to use .encode on it.
        except (struct.error, AttributeError):
            raise SequenceError('failed '
                'while packing {0!r}={1!r}'.format(self.name, val))

    def _unpack1(self, via, bstring, offset, noerror):
        "internal function to unpack single item"
        try:
            tup = via.unpack_from(bstring, offset)
        except struct.error as err:
            if 'unpack_from requires a buffer of at least' in str(err):
                if noerror:
                    return None, offset + via.size
                raise SequenceError('out of data '
                    'while unpacking {0!r}'.format(self.name))
            # not clear what to do here if noerror
            raise SequenceError('failed '
                'while unpacking {0!r}'.format(self.name))
        assert len(tup) == 1
        return tup[0], offset + via.size

    def unpack(self, vdict, cdict, bstring, offset, noerror=False):
        "decode a value; return the value and the new offset"
        if self.struct:
            return self._unpack1(self.struct, bstring, offset, noerror)
        slen, offset = self._unpack1(EncDecSimple.string_len, bstring, offset,
            noerror)
        if slen is None:
            return None, offset
        nexto = offset + slen
        if len(bstring) < nexto:
            if noerror:
                val = None
            else:
                raise SequenceError('out of data '
                    'while unpacking {0!r}'.format(self.name))
        else:
            val = bstring[offset:nexto]
            if self.autob2s:
                val = self.b2s(val)
        return val, nexto

# string length: 2 byte unsigned field
EncDecSimple.string_len = _ProtoStruct[2]

class EncDecTyped(EncDec):
    r"""
    EncDec for typed objects (which are build from PFODs, which are
    a sneaky class variant of OrderedDict similar to namedtuple).

    Calling the klass() function with no arguments must create an
    instance with all-None members.

    We also require a Sequencer to pack and unpack the members of
    the underlying pfod.

    >>> qid_s = Sequencer('qid')
    >>> qid_s.append_encdec(None, EncDecSimple('type', 1))
    >>> qid_s.append_encdec(None, EncDecSimple('version', 4))
    >>> qid_s.append_encdec(None, EncDecSimple('path', 8))
    >>> len(qid_s)
    3

    >>> from pfod import pfod
    >>> qid = pfod('qid', ['type', 'version', 'path'])
    >>> len(qid._fields)
    3
    >>> qid_inst = qid(1, 2, 3)
    >>> qid_inst
    qid(type=1, version=2, path=3)

    >>> e = EncDecTyped(qid, 'aqid', qid_s)
    >>> e.b2s(e.pack({}, {}, qid_inst))
    '\x01\x02\x00\x00\x00\x03\x00\x00\x00\x00\x00\x00\x00'
    >>> e.unpack({}, {},
    ... b'\x01\x02\x00\x00\x00\x03\x00\x00\x00\x00\x00\x00\x00', 0)
    (qid(type=1, version=2, path=3), 13)

    If an EncDecTyped instance has a conditional sequencer, note
    that unpacking will leave un-selected items set to None (see
    the Sequencer example below):

    >>> breakfast = pfod('breakfast', 'eggs spam ham')
    >>> breakfast()
    breakfast(eggs=None, spam=None, ham=None)
    >>> bfseq = Sequencer('breakfast')
    >>> bfseq.append_encdec(None, EncDecSimple('eggs', 1))
    >>> bfseq.append_encdec('yuck', EncDecSimple('spam', 1))
    >>> bfseq.append_encdec(None, EncDecSimple('ham', 1))
    >>> e = EncDecTyped(breakfast, 'bfname', bfseq)
    >>> e.unpack({}, {'yuck': False}, b'\x02\x01\x04', 0)
    (breakfast(eggs=2, spam=None, ham=1), 2)

    This used just two of the three bytes: eggs=2, ham=1.

    >>> e.unpack({}, {'yuck': True}, b'\x02\x01\x04', 0)
    (breakfast(eggs=2, spam=1, ham=4), 3)

    This used the third byte, so ham=4.
    """
    def __init__(self, klass, name, sequence, aux=None):
        assert len(sequence) == len(klass()._fields) # temporary
        super(EncDecTyped, self).__init__(name, aux)
        self.klass = klass
        self.name = name
        self.sequence = sequence

    def __repr__(self):
        if self.aux is None:
            return '{0}({1!r}, {2!r}, {3!r})'.format(self.__class__.__name__,
                self.klass, self.name, self.sequence)
        return '{0}({1!r}, {2!r}, {3!r}, {4!r})'.format(self.__class__.__name__,
            self.klass, self.name, self.sequence, self.aux)

    __str__ = __repr__

    def apack(self, vdict, cdict, val):
        """
        Pack each of our instance variables.

        Note that some packing may be conditional.
        """
        return self.sequence.apack(val, cdict)

    def unpack(self, vdict, cdict, bstring, offset, noerror=False):
        """
        Unpack each instance variable, into a new object of
        self.klass.  Return the new instance and new offset.

        Note that some unpacking may be conditional.
        """
        obj = self.klass()
        offset = self.sequence.unpack_from(obj, cdict, bstring, offset, noerror)
        return obj, offset

class EncDecA(EncDec):
    r"""
    EncDec for arrays (repeated objects).

    We take the name of repeat count variable, and a sub-coder
    (Sequencer instance).  For instance, we can en/de-code
    repeat='nwname' copies of name='wname', or nwname of
    name='wqid', in a Twalk en/de-code.

    Note that we don't pack or unpack the repeat count itself --
    that must be done by higher level code.  We just get its value
    from vdict.

    >>> subcode = EncDecSimple('wname', '_string_')
    >>> e = EncDecA('nwname', 'wname', subcode)
    >>> e.b2s(e.pack({'nwname': 2}, {}, ['A', 'BC']))
    '\x01\x00A\x02\x00BC'

    >>> subcode.autob2s = True # so that A and BC decode to py3k str
    >>> e.unpack({'nwname': 2}, {}, b'\x01\x00A\x02\x00BC', 0)
    (['A', 'BC'], 7)

    When using noerror, the first sub-item that fails to decode
    completely starts the None-s.  Strings whose length fails to
    decode are assumed to be zero bytes long as well, for the
    purpose of showing the expected packet length:

    >>> e.unpack({'nwname': 2}, {}, b'\x01\x00A\x02\x00', 0, noerror=True)
    (['A', None], 7)
    >>> e.unpack({'nwname': 2}, {}, b'\x01\x00A\x02', 0, noerror=True)
    (['A', None], 5)
    >>> e.unpack({'nwname': 3}, {}, b'\x01\x00A\x02', 0, noerror=True)
    (['A', None, None], 7)

    As a special case, supplying None for the sub-coder
    makes the repeated item pack or unpack a simple byte
    string.  (Note that autob2s is not supported here.)
    A too-short byte string is simply truncated!

    >>> e = EncDecA('count', 'data', None)
    >>> e.b2s(e.pack({'count': 5}, {}, b'12345'))
    '12345'
    >>> x = list(e.unpack({'count': 3}, {}, b'123', 0))
    >>> x[0] = e.b2s(x[0])
    >>> x
    ['123', 3]
    >>> x = list(e.unpack({'count': 3}, {}, b'12', 0, noerror=True))
    >>> x[0] = e.b2s(x[0])
    >>> x
    ['12', 3]
    """
    def __init__(self, repeat, name, sub, aux=None):
        super(EncDecA, self).__init__(name, aux)
        self.repeat = repeat
        self.name = name
        self.sub = sub

    def __repr__(self):
        if self.aux is None:
            return '{0}({1!r}, {2!r}, {3!r})'.format(self.__class__.__name__,
                self.repeat, self.name, self.sub)
        return '{0}({1!r}, {2!r}, {3!r}, {4!r})'.format(self.__class__.__name__,
            self.repeat, self.name, self.sub, self.aux)

    __str__ = __repr__

    def apack(self, vdict, cdict, val):
        "pack each val[i], for i in range(vdict[self.repeat])"
        num = vdict[self.repeat]
        assert num == len(val)
        if self.sub is None:
            assert isinstance(val, bytes)
            return [val]
        parts = []
        for i in val:
            parts.extend(self.sub.apack(vdict, cdict, i))
        return parts

    def unpack(self, vdict, cdict, bstring, offset, noerror=False):
        "unpack repeatedly, per self.repeat, into new array."
        num = vdict[self.repeat]
        if num is None and noerror:
            num = 0
        else:
            assert num >= 0
        if self.sub is None:
            nexto = offset + num
            if len(bstring) < nexto and not noerror:
                raise SequenceError('out of data '
                    'while unpacking {0!r}'.format(self.name))
            return bstring[offset:nexto], nexto
        array = []
        for i in range(num):
            obj, offset = self.sub.unpack(vdict, cdict, bstring, offset,
                noerror)
            array.append(obj)
        return array, offset

class SequenceError(Exception):
    "sequence error: item too big, or ran out of data"
    pass

class Sequencer(object):
    r"""
    A sequencer is an object that packs (marshals) or unpacks
    (unmarshals) a series of objects, according to their EncDec
    instances.

    The objects themselves (and their values) come from, or
    go into, a dictionary: <vdict>, the first argument to
    pack/unpack.

    Some fields may be conditional.  The conditions are in a
    separate dictionary (the second or <cdict> argument).

    Some objects may be dictionaries or PFODs, e.g., they may
    be a Plan9 qid or stat structure.  These have their own
    sub-encoding.

    As with each encoder, we have both an apack() function
    (returns a list of parts) and a plain pack().  Users should
    mostly stick with plain pack().

    >>> s = Sequencer('monty')
    >>> s
    Sequencer('monty')
    >>> e = EncDecSimple('eggs', 2)
    >>> s.append_encdec(None, e)
    >>> s.append_encdec(None, EncDecSimple('spam', 1))
    >>> s[0]
    (None, EncDecSimple('eggs', 2))
    >>> e.b2s(s.pack({'eggs': 513, 'spam': 65}, {}))
    '\x01\x02A'

    When particular fields are conditional, they appear in
    packed output, or are taken from the byte-string during
    unpacking, only if their condition is true.

    As with struct, use unpack_from to start at an arbitrary
    offset and/or omit verification that the entire byte-string
    is consumed.

    >>> s = Sequencer('python')
    >>> s.append_encdec(None, e)
    >>> s.append_encdec('.u', EncDecSimple('spam', 1))
    >>> s[1]
    ('.u', EncDecSimple('spam', 1))
    >>> e.b2s(s.pack({'eggs': 513, 'spam': 65}, {'.u': True}))
    '\x01\x02A'
    >>> e.b2s(s.pack({'eggs': 513, 'spam': 65}, {'.u': False}))
    '\x01\x02'

    >>> d = {}
    >>> s.unpack(d, {'.u': True}, b'\x01\x02A')
    >>> print(d['eggs'], d['spam'])
    513 65
    >>> d = {}
    >>> s.unpack(d, {'.u': False}, b'\x01\x02A', 0)
    Traceback (most recent call last):
        ...
    SequenceError: 1 byte(s) unconsumed
    >>> s.unpack_from(d, {'.u': False}, b'\x01\x02A', 0)
    2
    >>> print(d)
    {'eggs': 513}

    The incoming dictionary-like object may be pre-initialized
    if you like; only sequences that decode are filled-in:

    >>> d = {'eggs': None, 'spam': None}
    >>> s.unpack_from(d, {'.u': False}, b'\x01\x02A', 0)
    2
    >>> print(d['eggs'], d['spam'])
    513 None

    Some objects may be arrays; if so their EncDec is actually
    an EncDecA, the repeat count must be in the dictionary, and
    the object itself must have a len() and be index-able:

    >>> s = Sequencer('arr')
    >>> s.append_encdec(None, EncDecSimple('n', 1))
    >>> ae = EncDecSimple('array', 2)
    >>> s.append_encdec(None, EncDecA('n', 'array', ae))
    >>> ae.b2s(s.pack({'n': 2, 'array': [257, 514]}, {}))
    '\x02\x01\x01\x02\x02'

    Unpacking an array creates a list of the number of items.
    The EncDec encoder that decodes the number of items needs to
    occur first in the sequencer, so that the dictionary will have
    acquired the repeat-count variable's value by the time we hit
    the array's encdec:

    >>> d = {}
    >>> s.unpack(d, {}, b'\x01\x04\x00')
    >>> d['n'], d['array']
    (1, [4])
    """
    def __init__(self, name):
        self.name = name
        self._codes = []
        self.debug = False # or sys.stderr

    def __repr__(self):
        return '{0}({1!r})'.format(self.__class__.__name__, self.name)

    __str__ = __repr__

    def __len__(self):
        return len(self._codes)

    def __iter__(self):
        return iter(self._codes)

    def __getitem__(self, index):
        return self._codes[index]

    def dprint(self, *args, **kwargs):
        if not self.debug:
            return
        if isinstance(self.debug, bool):
            dest = sys.stdout
        else:
            dest = self.debug
        print(*args, file=dest, **kwargs)

    def append_encdec(self, cond, code):
        "add EncDec en/de-coder, conditional on cond"
        self._codes.append((cond, code))

    def apack(self, vdict, cdict):
        """
        Produce packed representation of each field.
        """
        packed_data = []
        for cond, code in self._codes:
            # Skip this item if it's conditional on a false thing.
            if cond is not None and not cdict[cond]:
                self.dprint('skip %r - %r is False' % (code, cond))
                continue

            # Pack the item.
            self.dprint('pack %r - no cond or %r is True' % (code, cond))
            packed_data.extend(code.apack(vdict, cdict, vdict[code.name]))

        return packed_data

    def pack(self, vdict, cdict):
        """
        Flatten packed data.
        """
        return b''.join(self.apack(vdict, cdict))

    def unpack_from(self, vdict, cdict, bstring, offset=0, noerror=False):
        """
        Unpack from byte string.

        The values are unpacked into a dictionary vdict;
        some of its entries may themselves be ordered
        dictionaries created by typedefed codes.

        Raises SequenceError if the string is too short,
        unless you set noerror, in which case we assume
        you want see what you can get out of the data.
        """
        for cond, code in self._codes:
            # Skip this item if it's conditional on a false thing.
            if cond is not None and not cdict[cond]:
                self.dprint('skip %r - %r is False' % (code, cond))
                continue

            # Unpack the item.
            self.dprint('unpack %r - no cond or %r is True' % (code, cond))
            obj, offset = code.unpack(vdict, cdict, bstring, offset, noerror)
            vdict[code.name] = obj

        return offset

    def unpack(self, vdict, cdict, bstring, noerror=False):
        """
        Like unpack_from but unless noerror=True, requires that
        we completely use up the given byte string.
        """
        offset = self.unpack_from(vdict, cdict, bstring, 0, noerror)
        if not noerror and offset != len(bstring):
            raise SequenceError('{0} byte(s) unconsumed'.format(
                len(bstring) - offset))

if __name__ == '__main__':
    import doctest
    doctest.testmod()