diff --git a/tests/corekit/_roundtrip.py b/tests/corekit/_roundtrip.py new file mode 100644 index 0000000000..398967e62a --- /dev/null +++ b/tests/corekit/_roundtrip.py @@ -0,0 +1,151 @@ +# -*- coding: utf-8 -*- +"""Shared known-failure tables for the corekit round-trip modules. C.f. #1202. + +The owner's rule is that every round trip the library offers reproduces its +input exactly. The ``test_*_roundtrip_unit`` modules beside this one enumerate +the corekit round trips -- a field's ``pack``/``unpack``, a schema's +``pack``/``unpack`` and ``from_dict``/``to_dict``, an :class:`~pcapkit.corekit. +infoclass.Info`'s ``from_dict``/``to_dict`` -- and report one :class:`Outcome` +per case. A case that does not close is listed in its module's +``KNOWN_FAILURES``, one :class:`Gap` per root cause, in the style of +:mod:`tests.protocols._edge_roundtrip`. + +The table is asserted in both directions: a case that fails and is not listed +turns its module red, and so does a listed case that passes, so a fix has to +delete its entry in the same change. A pattern that matches no case is stale and +fails too. + +""" + +from __future__ import annotations + +import fnmatch +import importlib.util +import unittest +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import time_limit + +if TYPE_CHECKING: + from typing import Any, Callable, Iterable, Optional + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Whole seconds one case may take; a working case takes milliseconds. +CASE_TIMEOUT = 30 + + +class Outcome(NamedTuple): + """What one check found.""" + + #: ``'OK'`` or a module-specific failure status. + status: 'str' + #: Failure detail; empty when the case closed. + detail: 'str' = '' + + +#: The outcome of a case that closed. +OK = Outcome('OK') + + +class Gap(NamedTuple): + """One root cause, and every case it stops.""" + + #: GitHub issue tracking the defect. + issue: 'Optional[int]' + #: The defect, with the ``file:line`` that causes it. + defect: 'str' + #: Status every case below reports, or a tuple of which each reports one. + status: 'str | tuple[str, ...]' + #: Substring every failure detail must contain, or a tuple of which each + #: detail must contain one. + fragment: 'str | tuple[str, ...]' + #: Labels of the cases this defect stops, as :mod:`fnmatch` patterns. Each + #: pattern must match at least one case, and every case it matches must fail + #: as recorded, so a pattern is only as wide as the defect. + cases: 'tuple[str, ...]' + + +def describe(exc: 'BaseException') -> 'str': + """``Type: message``, as a failure detail.""" + return f'{type(exc).__name__}: {exc}' + + +def diff(got: 'bytes', want: 'bytes') -> 'str': + """Two octet strings as a failure detail.""" + return f'{len(got)} octets {got.hex()} != {len(want)} octets {want.hex()}' + + +def run(func: 'Callable[..., Outcome]', *args: 'Any') -> 'Outcome': + """Run ``func(*args)``, one check, under :data:`CASE_TIMEOUT`.""" + try: + with time_limit(CASE_TIMEOUT): + return func(*args) + except TimeoutError as exc: + return Outcome('TIMEOUT', str(exc)) + + +def _as_tuple(value: 'str | tuple[str, ...]') -> 'tuple[str, ...]': + return (value,) if isinstance(value, str) else value + + +class KnownFailureTable: + """The table assertions, mixed into a :class:`unittest.TestCase`.""" + + #: Known failures, one entry per root cause. + KNOWN_FAILURES = () # type: tuple[Gap, ...] + #: Statuses a case in this module may report. + STATUSES = ('OK',) # type: tuple[str, ...] + + def gap_table(self: 'Any', labels: 'Iterable[str]') -> 'dict[str, Gap]': + """Every label a ``KNOWN_FAILURES`` pattern matches, to its entry.""" + labels = sorted(labels) + table = {} # type: dict[str, Gap] + for gap in self.KNOWN_FAILURES: + for pattern in gap.cases: + for label in fnmatch.filter(labels, pattern): + self.assertIs(table.setdefault(label, gap), gap, + f'{label} is listed under two root causes') + return table + + def check_table(self: 'Any', labels: 'Iterable[str]') -> 'None': + """Every pattern names a case, and every entry is well formed.""" + labels = sorted(labels) + self.assertEqual(len(labels), len(set(labels)), 'duplicate case labels') + stale = [pattern for gap in self.KNOWN_FAILURES for pattern in gap.cases + if not fnmatch.filter(labels, pattern)] + self.assertEqual(stale, [], 'KNOWN_FAILURES patterns that match no case') + for gap in self.KNOWN_FAILURES: + self.assertTrue(gap.cases, f'empty KNOWN_FAILURES entry: {gap.defect}') + for status in _as_tuple(gap.status): + self.assertIn(status, self.STATUSES) + self.assertNotEqual(status, 'OK', gap.defect) + + def check_outcome(self: 'Any', label: 'str', outcome: 'Outcome', + gaps: 'dict[str, Gap]') -> 'None': + """Assert ``outcome`` is ``OK``, or fails as its entry records.""" + gap = gaps.get(label) + if gap is None: + self.assertEqual( + outcome.status, 'OK', + f'{label} does not round-trip: {outcome.status}: {outcome.detail}. ' + f'If this is a new defect, add it to KNOWN_FAILURES under its root cause.') + return + self.assertIn( + outcome.status, _as_tuple(gap.status), + f'{label} was recorded as failing with {gap.status} (#{gap.issue}: ' + f'{gap.defect}) but came back {outcome.status}: {outcome.detail}. ' + f'If the defect is fixed, delete the label from its entry.') + fragments = [fragment for fragment in _as_tuple(gap.fragment) if fragment] + if fragments: + self.assertTrue( + any(fragment in outcome.detail for fragment in fragments), + f'{label} fails, but not in the recorded way ({gap.defect}); ' + f'detail was {outcome.detail!r}') + + +def skip_without_runtime(case: 'unittest.TestCase') -> 'None': + """Skip ``case`` when the runtime dependencies are not installed.""" + if not HAS_RUNTIME: + case.skipTest('runtime dependencies not installed') diff --git a/tests/corekit/test_field_roundtrip_unit.py b/tests/corekit/test_field_roundtrip_unit.py new file mode 100644 index 0000000000..7b6d153f85 --- /dev/null +++ b/tests/corekit/test_field_roundtrip_unit.py @@ -0,0 +1,701 @@ +# -*- coding: utf-8 -*- +"""Every corekit field packs and unpacks symmetrically. C.f. #1202. + +For each concrete field class under :mod:`pcapkit.corekit.fields`, two +directions are checked on representative and boundary inputs: + +* **value**: ``unpack(pack(v)) == v``, and packing the result again gives the + same octets; +* **wire**: ``pack(unpack(b)) == b``. + +A field that a schema only ever uses in context -- a forward match, a +conditional, a nested schema left with a remainder, an option list followed by +its padding -- is also checked inside a minimal schema, as +``Schema.unpack(b).pack() == b`` and ``from_dict(to_dict())`` packing ``b``. + +Malformed wire input a field is entitled to refuse is listed in +:data:`REJECTED` and must raise an in-library exception. Every other failing +case is a defect, listed in ``KNOWN_FAILURES`` by root cause (see +:mod:`tests.corekit._roundtrip`). + +Nothing here imports :mod:`pcapkit` at module level; the cases are built inside +each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import collections +import enum +import inspect +import ipaddress +import unittest +import warnings +from typing import TYPE_CHECKING, NamedTuple + +from tests._support import reimport_once_per_class +from tests.corekit._roundtrip import (OK, Gap, KnownFailureTable, Outcome, describe, diff, run, + skip_without_runtime) + +if TYPE_CHECKING: + from typing import Any, Callable + + +class FieldCase(NamedTuple): + """One field and the inputs it is round-tripped on.""" + + #: Unique label prefix, ``FieldClass/variant``. + label: 'str' + #: Builds the (unresolved) field. + make: 'Callable[[], Any]' + #: Values for the value direction. + values: 'tuple[Any, ...]' = () + #: Octets for the wire direction. + wires: 'tuple[bytes, ...]' = () + #: Packet context the field is resolved and (un)packed with. + packet: 'dict[str, Any]' = {} + + +class SchemaCase(NamedTuple): + """One minimal schema exercising a field in context, and its wire input.""" + + #: Unique label, ``schema/what``. + label: 'str' + #: Builds the schema class. + make: 'Callable[[], Any]' + #: Wire octets. + raw: 'bytes' + + +#: Field classes that are not round-tripped on their own, and why. +NOT_CONCRETE = { + 'FieldBase': 'internal base class; its template is the placeholder ``0s``', + 'Field': 'base class: ``Field.__init__`` sets no ``_template``, so a bare ' + '``Field`` cannot be packed; every concrete subclass sets one', +} + +#: Wire labels a field must refuse with an in-library exception, to that +#: exception's class name. +REJECTED = { + 'IPv6InterfaceField/prefix/wire/' + '00' * 16 + '81': 'FieldValueError', + 'IPv6InterfaceField/prefix/wire/' + '00' * 16 + 'ff': 'FieldValueError', +} + + +def named(field: 'Any', name: 'str' = 'items') -> 'Any': + """Name ``field`` as a schema class body would (``__set_name__``). + + :class:`~pcapkit.corekit.fields.collections.ListField` replaces + ``FieldBase.__init__`` without setting a placeholder name, and + :meth:`OptionField.unpack ` + reads ``self.name``, so a list field used outside a schema needs one + (#1491). + + """ + field.name = name + return field + + +def _bounds(width: 'int', signed: 'bool') -> 'tuple[int, ...]': + """The boundary values of a ``width``-octet integer, and one inside.""" + bits = width * 8 + if signed: + return (0, 1, -1, (1 << (bits - 1)) - 1, -(1 << (bits - 1))) + return (0, 1, (1 << bits) - 1, 1 << (bits - 1)) + + +def _wires(width: 'int') -> 'tuple[bytes, ...]': + """Representative octets of a ``width``-octet field.""" + if width == 0: + return (b'',) + return (bytes(width), b'\xff' * width, b'\x80' + bytes(width - 1), + bytes(range(1, width + 1))) + + +def field_cases() -> 'tuple[FieldCase, ...]': # pylint: disable=too-many-locals,too-many-statements + """Every field case, built against the current :mod:`pcapkit` import.""" + import aenum + + from pcapkit.const.reg.transtype import TransType + from pcapkit.corekit.fields import collections as fcoll + from pcapkit.corekit.fields import ipaddress as fip + from pcapkit.corekit.fields import misc as fmisc + from pcapkit.corekit.fields import numbers as fnum + from pcapkit.corekit.fields import strings as fstr + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.schema.schema import Schema, schema_final + + cases = [] # type: list[FieldCase] + + # -- numbers ------------------------------------------------------------ + fixed = {'Int8Field': 1, 'UInt8Field': 1, 'Int16Field': 2, 'UInt16Field': 2, + 'Int32Field': 4, 'UInt32Field': 4, 'Int64Field': 8, 'UInt64Field': 8} + for name, width in fixed.items(): + cls = getattr(fnum, name) + for order in ('big', 'little'): + cases.append(FieldCase( + f'{name}/{order}', lambda cls=cls, order=order: cls(byteorder=order), + _bounds(width, cls.__signed__), _wires(width))) + + for width in (1, 2, 3, 4, 5, 7, 8, 9, 16): + for signed in (False, True): + for order in ('big', 'little'): + sign = 'signed' if signed else 'unsigned' + cases.append(FieldCase( + f'NumberField/{width}-{sign}-{order}', + lambda width=width, signed=signed, order=order: fnum.NumberField( + length=lambda pkt: pkt['n'], signed=signed, byteorder=order), + _bounds(width, signed), _wires(width), {'n': width})) + cases.append(FieldCase( + 'NumberField/3-fixed', lambda: fnum.NumberField(length=3), + _bounds(3, False), _wires(3))) + cases.append(FieldCase( + 'NumberField/bit-length-12', lambda: fnum.UInt16Field(bit_length=12), + (0, 1, 0xFFF), (b'\x00\x00', b'\x0f\xff', b'\xff\xff', b'\xf0\x00'))) + + class Stdlib(enum.IntEnum): + """A stdlib registry with gaps.""" + A = 0 + B = 1 + C = 255 + + class Aenum(aenum.IntEnum): + """An :mod:`aenum` registry with gaps.""" + A = 0 + B = 2 + + for label, namespace, values in (('stdlib', Stdlib, tuple(Stdlib)), + ('aenum', Aenum, tuple(Aenum)), + ('none', None, (0, 7, 255)), + ('transtype', TransType, (TransType.TCP, TransType.UDP))): + cases.append(FieldCase( + f'EnumField/{label}', + lambda namespace=namespace: fnum.EnumField(length=1, namespace=namespace), + values, (b'\x00', b'\x01', b'\x02', b'\x06', b'\x7f', b'\x90', b'\xff'))) + cases.append(FieldCase( + 'EnumField/2-octet', lambda: fnum.EnumField(length=2, namespace=TransType), + (TransType.TCP,), (b'\x00\x06', b'\x01\x00', b'\xff\xff'))) + cases.append(FieldCase( + 'EnumField/bit-length-3', lambda: fnum.EnumField(length=1, bit_length=3, namespace=None), + (0, 7), (b'\x00', b'\x07', b'\xff'))) + + # -- strings ------------------------------------------------------------ + for width in (0, 1, 4): + cases.append(FieldCase( + f'BytesField/{width}', lambda width=width: fstr.BytesField(length=width), + (bytes(width), b'\xff' * width), _wires(width))) + cases.append(FieldCase( + 'BytesField/callable', lambda: fstr.BytesField(length=lambda pkt: pkt['n']), + (b'abcde', bytes(5)), _wires(5), {'n': 5})) + + texts = { + '': b'', 'abc': b'abc', 'utf-8': 'café'.encode(), 'latin-1': 'café'.encode('latin-1'), + 'invalid': b'\xff\xfe\x00', 'nul': b'\x00\x00', 'embedded-nul': b'a\x00b', + 'all-octets': bytes(range(256)), + } + for label, raw in texts.items(): + cases.append(FieldCase( + f'StringField/detect-{label or "empty"}', + lambda raw=raw: fstr.StringField(length=len(raw)), (), (raw,))) + for encoding, values, wires in ( + ('utf-8', ('', 'abc', 'café', '€', '\U0001f600'), (b'\xc3', b'\xe2\x82', b'\xff')), + ('ascii', ('', 'abc'), (b'\x80', b'abc\xff')), + ('latin-1', ('', 'café', 'ÿ'), (b'\xff\xfe',)), + ('utf-16', ('abc', '€'), (b'\xff\xfea\x00', b'a\x00', b'\x00'))): + for value in values: + raw = value.encode(encoding) + cases.append(FieldCase( + f'StringField/{encoding}/{value.encode("unicode_escape").decode() or "empty"}', + lambda encoding=encoding, raw=raw: fstr.StringField(length=len(raw), encoding=encoding), + (value,), (raw,))) + for raw in wires: + cases.append(FieldCase( + f'StringField/{encoding}/bad-{raw.hex()}', + lambda encoding=encoding, raw=raw: fstr.StringField(length=len(raw), encoding=encoding), + (), (raw,))) + for errors in ('ignore', 'replace'): + for raw in (b'abc', b'\xff\xfe', b'caf\xe9'): + cases.append(FieldCase( + f'StringField/errors-{errors}/{raw.hex()}', + lambda errors=errors, raw=raw: fstr.StringField(length=len(raw), encoding='utf-8', + errors=errors), + (), (raw,))) + for raw in (b'a%20b', b'%zz', b'%41', b'+', b'%E9', b'a b', b'%25', b'%C3%A9'): + cases.append(FieldCase( + f'StringField/unquote/{raw.decode()}', + lambda raw=raw: fstr.StringField(length=len(raw), unquote=True), (), (raw,))) + for value in ('a b', 'café', '%', '/x?y=1&z'): + import urllib.parse # pylint: disable=import-outside-toplevel + raw = urllib.parse.quote(value).encode() + cases.append(FieldCase( + f'StringField/unquote-value/{urllib.parse.quote(value)}', + lambda raw=raw: fstr.StringField(length=len(raw), unquote=True), (value,), ())) + + full = {'a': (0, 1), 'b': (1, 3), 'c': (4, 4)} + cases.append(FieldCase( + 'BitField/1-octet', lambda: fstr.BitField(length=1, namespace=full), + ({'a': 0, 'b': 0, 'c': 0}, {'a': 1, 'b': 7, 'c': 15}, {'a': 1, 'b': 2, 'c': 5}), + _wires(1) + (b'\xa5',))) + wide = {'flag': (0, 1), 'id': (1, 31)} + cases.append(FieldCase( + 'BitField/4-octet', lambda: fstr.BitField(length=4, namespace=wide), + ({'flag': 0, 'id': 0}, {'flag': 1, 'id': (1 << 31) - 1}), _wires(4))) + partial = {'a': (0, 1), 'c': (8, 4)} + cases.append(FieldCase( + 'BitField/partial', lambda: fstr.BitField(length=2, namespace=partial), + ({'a': 1, 'c': 15},), (b'\x00\x00', b'\x80\xf0', b'\xff\xff', b'\x01\x00'))) + + for width in (0, 3): + cases.append(FieldCase( + f'PaddingField/{width}', lambda width=width: fstr.PaddingField(length=width), + (bytes(width), b'\x01' * width), _wires(width))) + cases.append(FieldCase( + 'PaddingField/callable', lambda: fstr.PaddingField(length=lambda pkt: pkt['n']), + (b'\xde\xad',), _wires(2), {'n': 2})) + + # -- addresses ---------------------------------------------------------- + cases.append(FieldCase( + 'IPv4AddressField/plain', fip.IPv4AddressField, + tuple(ipaddress.IPv4Address(a) for a in ('0.0.0.0', '255.255.255.255', '10.0.0.1')), + _wires(4))) + cases.append(FieldCase( + 'IPv6AddressField/plain', fip.IPv6AddressField, + tuple(ipaddress.IPv6Address(a) for a in ('::', 'fe80::1', '::ffff:1.2.3.4', + 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff')), + _wires(16))) + cases.append(FieldCase( + 'IPv4InterfaceField/masks', fip.IPv4InterfaceField, + tuple(ipaddress.IPv4Interface(a) for a in ('0.0.0.0/0', '10.0.0.1/24', '10.0.0.1/32', + '255.255.255.255/1')), + tuple(bytes.fromhex(h) for h in ( + '0000000000000000', '0a000001ffffff00', '0a000001000000ff', '0a000001ff00ff00', + 'ffffffffffffffff', '0a00000100000000', '0a0000017fffffff', '0a00000180000001')))) + cases.append(FieldCase( + 'IPv6InterfaceField/prefix', fip.IPv6InterfaceField, + tuple(ipaddress.IPv6Interface(a) for a in ('::/0', 'fe80::1/64', '::1/128')), + tuple(fill + bytes([prefix]) for fill in (bytes(16), b'\xff' * 16) + for prefix in (0, 1, 64, 127, 128)) + (bytes(16) + b'\x81', bytes(16) + b'\xff'))) + + # -- misc --------------------------------------------------------------- + cases.append(FieldCase('NoValueField/plain', fmisc.NoValueField, (None,), (b'',))) + cases.append(FieldCase( + 'ConditionalField/true', + lambda: fmisc.ConditionalField(fnum.UInt16Field(), lambda pkt: pkt['flag']), + _bounds(2, False), _wires(2), {'flag': True})) + cases.append(FieldCase( + 'ConditionalField/false', + lambda: fmisc.ConditionalField(fnum.UInt16Field(default=7), lambda pkt: pkt['flag']), + (), (b'',), {'flag': False})) + cases.append(FieldCase( + 'PayloadField/bytes', fmisc.PayloadField, (b'', b'abc'), (b'', b'abc', bytes(64)))) + cases.append(FieldCase( + 'PayloadField/raw', lambda: fmisc.PayloadField(protocol=Raw), (), (b'abc', bytes(64)))) + + def switch() -> 'Any': + return fmisc.SwitchField(selector=lambda pkt: ( + fnum.UInt16Field() if pkt['kind'] == 0 else fstr.BytesField(length=3))) + + cases.append(FieldCase('SwitchField/number', switch, (0, 0xFFFF), _wires(2), {'kind': 0})) + cases.append(FieldCase('SwitchField/bytes', switch, (b'abc',), _wires(3), {'kind': 1})) + cases.append(FieldCase( + 'ForwardMatchField/inner', lambda: fmisc.ForwardMatchField(fnum.UInt16Field()), + (), _wires(2))) + + @schema_final + class Inner(Schema): + """Three octets: a byte and a short.""" + + x: 'int' = fnum.UInt8Field() + y: 'int' = fnum.UInt16Field() + + cases.append(FieldCase( + 'SchemaField/fixed', lambda: fmisc.SchemaField(length=3, schema=Inner), + (Inner(x=1, y=2), Inner(x=0xFF, y=0xFFFF)), _wires(3))) + cases.append(FieldCase( + 'SchemaField/variable', lambda: fmisc.SchemaField(schema=Inner), + (Inner(x=0, y=0),), _wires(3))) + + cases.append(FieldCase( + 'ListField/bytes', lambda: named(fcoll.ListField(length=lambda pkt: pkt['n'])), + (b'abcd',), _wires(4), {'n': 4})) + cases.append(FieldCase( + 'ListField/numbers', + lambda: named(fcoll.ListField(length=lambda pkt: pkt['n'], item_type=fnum.UInt16Field())), + ([0, 1, 0xFFFF],), _wires(6), {'n': 6})) + cases.append(FieldCase( + 'ListField/schemas', + lambda: named(fcoll.ListField(length=lambda pkt: pkt['n'], + item_type=fmisc.SchemaField(length=3, schema=Inner))), + ([Inner(x=1, y=2), Inner(x=3, y=4)],), _wires(6), {'n': 6})) + + class Base(Schema): + """Option header: a one-octet type and a one-octet body length.""" + + kind: 'int' = fnum.UInt8Field() + size: 'int' = fnum.UInt8Field() + + @schema_final + class Body(Base): + """An option carrying ``size`` octets.""" + + body: 'bytes' = fstr.BytesField(length=lambda pkt: pkt['size']) + + @schema_final + class End(Base): + """End of option list.""" + + registry = collections.defaultdict(lambda: Body, {0: End, 1: Body}) + + cases.append(FieldCase( + 'OptionField/tlv', + lambda: named(fcoll.OptionField(length=lambda pkt: pkt['n'], base_schema=Base, + type_name='kind', registry=registry, eool=0)), + ([Body(kind=1, size=2, body=b'ab'), End(kind=0, size=0)],), + (b'\x01\x02ab\x00\x00', b'\x01\x00\x05\x01z', b'\x00\x00'), {'n': 6})) + return tuple(cases) + + +def schema_cases() -> 'tuple[SchemaCase, ...]': + """Minimal schemas holding the fields that only make sense in context.""" + from pcapkit.corekit.fields import collections as fcoll + from pcapkit.corekit.fields import misc as fmisc + from pcapkit.corekit.fields import numbers as fnum + from pcapkit.corekit.fields import strings as fstr + from pcapkit.protocols.schema.schema import Schema, schema_final + + def forward() -> 'Any': + @schema_final + class Forward(Schema): + """A forward match sizing the field after it.""" + + test: 'int' = fmisc.ForwardMatchField(fnum.UInt8Field()) + size: 'int' = fnum.UInt8Field() + body: 'bytes' = fstr.BytesField(length=lambda pkt: pkt['test']) + return Forward + + def conditional() -> 'Any': + @schema_final + class Conditional(Schema): + """A field present only when a flag is set.""" + + flag: 'int' = fnum.UInt8Field() + extra: 'int' = fmisc.ConditionalField(fnum.UInt16Field(), lambda pkt: pkt['flag'] == 1) + tail: 'int' = fnum.UInt8Field() + return Conditional + + def remainder() -> 'Any': + @schema_final + class Inner(Schema): + """Two octets.""" + + x: 'int' = fnum.UInt16Field() + + @schema_final + class Outer(Schema): + """A nested schema whose span holds more than it reads (#1380).""" + + size: 'int' = fnum.UInt8Field() + inner: 'Inner' = fmisc.SchemaField(length=lambda pkt: pkt['size'], schema=Inner) + tail: 'int' = fnum.UInt8Field() + return Outer + + def options() -> 'Any': + class Base(Schema): + """Option header.""" + + kind: 'int' = fnum.UInt8Field() + size: 'int' = fnum.UInt8Field() + + @schema_final + class Body(Base): + """An option carrying ``size`` octets.""" + + body: 'bytes' = fstr.BytesField(length=lambda pkt: pkt['size']) + + @schema_final + class End(Base): + """End of option list.""" + + registry = collections.defaultdict(lambda: Body, {0: End, 1: Body}) + + @schema_final + class Options(Schema): + """An option area of eight octets, then its padding (#1289).""" + + options: 'list[Base]' = fcoll.OptionField( + length=8, base_schema=Base, type_name='kind', registry=registry, eool=0) + padding: 'bytes' = fstr.PaddingField( + length=lambda pkt: pkt.get('__option_padding__', 0)) + return Options + + def payload() -> 'Any': + @schema_final + class Header(Schema): + """A header and its payload.""" + + size: 'int' = fnum.UInt16Field() + payload: 'bytes' = fmisc.PayloadField(length=lambda pkt: pkt['size']) + return Header + + def short() -> 'Any': + @schema_final + class Short(Schema): + """Fixed fields, for a truncated read (#1458).""" + + a: 'int' = fnum.UInt16Field() + b: 'int' = fnum.UInt32Field() + c: 'int' = fnum.UInt16Field() + return Short + + return ( + SchemaCase('forward/empty', forward, b'\x00'), + SchemaCase('forward/body', forward, b'\x03abc'), + SchemaCase('conditional/absent', conditional, b'\x00\x09'), + SchemaCase('conditional/present', conditional, b'\x01\xbe\xef\x09'), + SchemaCase('remainder/none', remainder, b'\x02\x12\x34\x09'), + SchemaCase('remainder/kept', remainder, b'\x04\x12\x34\xaa\xbb\x09'), + SchemaCase('remainder/zero-filled', remainder, b'\x04\x12\x34\x00\x00\x09'), + SchemaCase('options/eool-then-padding', options, b'\x01\x02ab\x00\x00\xde\xad'), + SchemaCase('options/full', options, b'\x01\x06abcdef'), + SchemaCase('payload/empty', payload, b'\x00\x00'), + SchemaCase('payload/bytes', payload, b'\x00\x03abc'), + SchemaCase('short/inside-second', short, b'\x00\x01\x02\x03'), + SchemaCase('short/inside-last', short, b'\x00\x01\x02\x03\x04\x05\x06'), + ) + + +def _in_library(exc: 'BaseException') -> 'bool': + from pcapkit.utilities.exceptions import BaseError + return isinstance(exc, BaseError) + + +def _hex(raw: 'bytes') -> 'str': + return raw.hex() or 'empty' + + +def check_value(case: 'FieldCase', value: 'Any') -> 'Outcome': + """``unpack(pack(v)) == v``, and the result packs to the same octets.""" + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + packed = case.make()(dict(case.packet)).pack(value, dict(case.packet)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('PACK', describe(exc)) + try: + got = case.make()(dict(case.packet)).unpack(packed, dict(case.packet)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('UNPACK', describe(exc)) + if got != value: + return Outcome('VALUE', f'{got!r} != {value!r} (packed {packed.hex()})') + try: + again = case.make()(dict(case.packet)).pack(got, dict(case.packet)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('REPACK', describe(exc)) + if again != packed: + return Outcome('REPACK', diff(again, packed)) + return OK + + +def check_wire(case: 'FieldCase', raw: 'bytes') -> 'Outcome': + """``pack(unpack(b)) == b``.""" + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + value = case.make()(dict(case.packet)).unpack(raw, dict(case.packet)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('REJECTED' if _in_library(exc) else 'UNPACK', describe(exc)) + if case.label.startswith('ForwardMatchField/'): + # A forward match is non-capturing: it reads what its inner field + # reads and packs nothing, leaving the octets to the next field. + inner = case.make().field(dict(case.packet)).unpack(raw, dict(case.packet)) + if value != inner: + return Outcome('VALUE', f'{value!r} != {inner!r}') + return OK + try: + packed = case.make()(dict(case.packet)).pack(value, dict(case.packet)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('PACK', describe(exc)) + if packed != raw: + return Outcome('WIRE', diff(packed, raw)) + return OK + + +def check_schema(case: 'SchemaCase') -> 'Outcome': + """``unpack(b).pack() == b``, and ``from_dict(to_dict())`` packs ``b``.""" + cls = case.make() + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + parsed = cls.unpack(case.raw, len(case.raw), None) + except Exception as exc: # pylint: disable=broad-except + return Outcome('UNPACK', describe(exc)) + try: + packed = parsed.pack() + except Exception as exc: # pylint: disable=broad-except + return Outcome('PACK', describe(exc)) + if packed != case.raw: + return Outcome('WIRE', diff(packed, case.raw)) + try: + rebuilt = type(parsed).from_dict(parsed.to_dict()).pack() + except Exception as exc: # pylint: disable=broad-except + return Outcome('DICT', describe(exc)) + if rebuilt != case.raw: + return Outcome('DICT', diff(rebuilt, case.raw)) + return OK + + +class FieldRoundTripTests(KnownFailureTable, unittest.TestCase): + """Field ``pack``/``unpack`` symmetry, in both directions.""" + + STATUSES = ('OK', 'PACK', 'UNPACK', 'VALUE', 'REPACK', 'WIRE', 'DICT', 'REJECTED', 'TIMEOUT') + + KNOWN_FAILURES = ( + Gap(1487, 'BitField.post_process reads only the bits its namespace names, and ' + 'pre_process seeds the rest as zeros, so octets with a bit outside every ' + 'subfield set pack back with it cleared ' + '(pcapkit/corekit/fields/strings.py:311, :334). Not reachable from a shipped ' + 'schema: all 102 BitFields cover every bit ' + '(test_every_schema_bitfield_covers_every_bit)', + 'WIRE', '', ('BitField/partial/wire/ffff', 'BitField/partial/wire/0100')), + Gap(1488, 'NumberField with bit_length narrower than its octets masks the unpacked value ' + 'to bit_length bits, so set high bits are dropped and pack writes them as ' + 'zeros (pcapkit/corekit/fields/numbers.py:318, :322). Shipped use: vlan.TCI ' + '(pcapkit/protocols/schema/link/vlan.py:47-51), which no protocol parses with', + 'WIRE', '', ('NumberField/bit-length-12/wire/ffff', 'NumberField/bit-length-12/wire/f000', + 'EnumField/bit-length-3/wire/ff')), + Gap(1489, 'ListField without an item_type unpacks to bytes ' + '(pcapkit/corekit/fields/collections.py:179), but ListField.pack iterates its ' + 'value item by item and refuses each int of those bytes ' + '(pcapkit/corekit/fields/collections.py:146, :154). Not reachable through ' + 'Schema.pack, which writes a bytes value itself ' + '(pcapkit/protocols/schema/schema.py:846)', + 'PACK', 'has invalid value', ('ListField/bytes/*',)), + ) + + def setUp(self) -> None: + skip_without_runtime(self) + reimport_once_per_class(self) + + # -- labels ------------------------------------------------------------- + + @staticmethod + def _value_labels(cases: 'tuple[FieldCase, ...]') -> 'list[tuple[str, FieldCase, Any]]': + return [(f'{case.label}/value/{index}', case, value) + for case in cases for index, value in enumerate(case.values)] + + @staticmethod + def _wire_labels(cases: 'tuple[FieldCase, ...]') -> 'list[tuple[str, FieldCase, bytes]]': + return [(f'{case.label}/wire/{_hex(raw)}', case, raw) + for case in cases for raw in case.wires] + + def _labels(self) -> 'list[str]': + cases = field_cases() + return ([label for label, _, _ in self._value_labels(cases)] + + [label for label, _, _ in self._wire_labels(cases)] + + [f'schema/{case.label}' for case in schema_cases()]) + + # -- the tables --------------------------------------------------------- + + def test_tables_name_real_cases(self) -> None: + labels = self._labels() + self.check_table(labels) + self.assertEqual(sorted(set(REJECTED) - set(labels)), [], 'stale REJECTED labels') + self.assertEqual(sorted(set(self.gap_table(labels)) & set(REJECTED)), [], + 'a label cannot be both rejected and a known failure') + + def test_every_concrete_field_class_has_a_case(self) -> None: + """A field class added tomorrow fails here until it gets a case.""" + import pcapkit.corekit.fields as package + from pcapkit.corekit.fields.field import FieldBase + + classes = set() + for module in ('field', 'numbers', 'strings', 'ipaddress', 'misc', 'collections'): + mod = getattr(package, module) + for name, obj in vars(mod).items(): + if (inspect.isclass(obj) and issubclass(obj, FieldBase) + and obj.__module__ == mod.__name__ and not name.startswith('_')): + classes.add(name) + covered = {case.label.split('/')[0] for case in field_cases()} + self.assertEqual(sorted(classes - covered - set(NOT_CONCRETE)), []) + self.assertEqual(sorted(set(NOT_CONCRETE) & covered), []) + self.assertEqual(sorted(set(NOT_CONCRETE) - classes), [], 'stale NOT_CONCRETE entries') + + def test_every_schema_bitfield_covers_every_bit(self) -> None: + """Why the partial-namespace gap above is unreachable from the library.""" + import importlib + import pkgutil + + import pcapkit.protocols.schema as package + from pcapkit.corekit.fields.misc import ConditionalField, ForwardMatchField + from pcapkit.corekit.fields.strings import BitField + from pcapkit.protocols.schema.schema import Schema + + total, uncovered = 0, [] + for info in pkgutil.walk_packages(package.__path__, f'{package.__name__}.'): + module = importlib.import_module(info.name) + for obj in vars(module).values(): + if not (inspect.isclass(obj) and issubclass(obj, Schema) and obj.__module__ == module.__name__): + continue + for name, field in obj.__fields__.items(): + if isinstance(field, ForwardMatchField): + # A forward match peeks and packs nothing; its bits are + # read again, for real, by the fields after it. + continue + if isinstance(field, ConditionalField): + field = field.field + if not isinstance(field, BitField): + continue + total += 1 + bits = set() + for start, size in field._namespace.values(): # pylint: disable=protected-access + bits.update(range(start, start + size)) + if len(bits) != field.length * 8: + uncovered.append(f'{obj.__qualname__}.{name}') + self.assertGreater(total, 0) + self.assertEqual(uncovered, []) + + # -- the checks --------------------------------------------------------- + + def test_unpack_of_pack_is_the_value(self) -> None: + cases = field_cases() + gaps = self.gap_table(self._labels()) + for label, case, value in self._value_labels(cases): + with self.subTest(case=label): + self.check_outcome(label, run(check_value, case, value), gaps) + + def test_pack_of_unpack_is_the_wire(self) -> None: + cases = field_cases() + gaps = self.gap_table(self._labels()) + for label, case, raw in self._wire_labels(cases): + if label in REJECTED: + continue + with self.subTest(case=label): + self.check_outcome(label, run(check_wire, case, raw), gaps) + + def test_malformed_wire_is_rejected_in_library(self) -> None: + cases = {label: (case, raw) for label, case, raw in self._wire_labels(field_cases())} + for label, exc in REJECTED.items(): + case, raw = cases[label] + with self.subTest(case=label): + outcome = run(check_wire, case, raw) + self.assertEqual(outcome.status, 'REJECTED', outcome.detail) + self.assertTrue(outcome.detail.startswith(f'{exc}: '), outcome.detail) + + def test_fields_in_a_schema_round_trip(self) -> None: + gaps = self.gap_table(self._labels()) + for case in schema_cases(): + label = f'schema/{case.label}' + with self.subTest(case=label): + self.check_outcome(label, run(check_schema, case), gaps) + + def test_a_decoded_string_keeps_its_octets(self) -> None: + """The lossless decode the string cases rely on (#1336), checked directly.""" + from pcapkit.corekit.fields.strings import DecodedString, StringField + + value = StringField(length=4)({}).unpack(b'caf\xe9', {}) + self.assertIsInstance(value, DecodedString) + self.assertEqual(value.raw, b'caf\xe9') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/corekit/test_info_roundtrip_unit.py b/tests/corekit/test_info_roundtrip_unit.py new file mode 100644 index 0000000000..2d73b866cc --- /dev/null +++ b/tests/corekit/test_info_roundtrip_unit.py @@ -0,0 +1,375 @@ +# -*- coding: utf-8 -*- +"""Every :class:`~pcapkit.corekit.infoclass.Info` survives ``to_dict``/``from_dict``. C.f. #1202. + +The documented inverse of :meth:`Info.to_dict +` is :meth:`Info.from_dict +`: nested values annotated with an +:class:`Info` subclass are rebuilt into it, and a key held more than once keeps +every value (#1466). For every :class:`Info` subclass the library defines -- the +``Data_*`` models of :mod:`pcapkit.protocols.data`, the reassembly and +traceflow models, and the rest -- an instance is synthesised from its +annotations and must come back: + +* equal, with the same ``to_dict()``, the same ``items(multi=True)``, and every + nested value of the same :class:`Info` class (``plain``); +* with a key repeated through an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` still repeated (``multi``); +* with the bookkeeping keys the parsers attach carried through: every + :class:`~pcapkit.protocols.data.data.Data` with ``__short_read__`` (#1465), every + PCAP-NG block with ``__truncated_raw__`` (#1475). + +``to_dict()`` must also be an :class:`~pcapkit.corekit.multidict.InfoDict` at +every level. :class:`InfoMechanicsTests` covers the shapes a synthesised +instance does not: a key renamed because it shadows a method, ``Optional`` +nesting, and repeated nested values. + +A failing case is a defect and goes in ``KNOWN_FAILURES`` (see +:mod:`tests.corekit._roundtrip`). Nothing imports :mod:`pcapkit` at module +level. The module has no ``from __future__ import annotations``, so the models +:class:`InfoMechanicsTests` declares carry real annotations that +:meth:`~pcapkit.corekit.infoclass.Info.from_dict` can resolve. + +""" + +import importlib +import inspect +import pkgutil +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class +from tests.corekit._roundtrip import (OK, Gap, KnownFailureTable, Outcome, describe, run, + skip_without_runtime) + +if TYPE_CHECKING: + from typing import Any + +#: Packages whose modules define :class:`Info` subclasses but are not imported by +#: ``import pcapkit`` alone. +PACKAGES = ('pcapkit.protocols.data', 'pcapkit.foundation.reassembly.data', + 'pcapkit.foundation.traceflow.data') + +#: Values a model's ``__post_init__`` needs in a shape no marker has, by label; +#: each is a callable, so it is built against the current import. +VALUES = { + 'protocols.data.misc.pcapng:NameResolutionBlock': { + 'records': lambda: importlib.import_module('pcapkit.corekit.multidict').OrderedMultiDict(), + }, +} + +#: Nesting depth past which a nested value is a marker rather than an instance; +#: some models are recursive. +MAX_DEPTH = 3 + + +def info_classes() -> 'dict[str, type]': + """Every :class:`Info` subclass defined under :mod:`pcapkit`, by label.""" + from pcapkit.corekit.infoclass import Info + + for package in PACKAGES: + module = importlib.import_module(package) + for info in pkgutil.walk_packages(module.__path__, f'{package}.'): + importlib.import_module(info.name) + + found = {} # type: dict[str, type] + stack = [Info] + while stack: + current = stack.pop() + for sub in current.__subclasses__(): + if sub.__module__.startswith('pcapkit.'): + found[f'{sub.__module__.replace("pcapkit.", "", 1)}:{sub.__qualname__}'] = sub + stack.append(sub) + return found + + +def annotated_keys(cls: 'type') -> 'list[str]': + """The keys :func:`~pcapkit.corekit.infoclass.info_final` builds ``__init__`` from.""" + from pcapkit.corekit.infoclass import Info + + keys = [] # type: list[str] + for base in reversed(cls.__mro__): + if base is Info or not issubclass(base, Info): + continue + # The same lookup info_final makes; on Python 3.14 a class's own + # annotations are no longer stored in its ``__dict__``. + for key in base.__dict__.get('__annotations__', getattr(base, '__annotations__', {})): + if key not in keys and not key.startswith('__'): + keys.append(key) + return keys + + +def synthesise(cls: 'type', depth: 'int' = 0) -> 'Any': + """An instance of ``cls``: a marker per key, a nested instance per nested key.""" + from pcapkit.corekit.infoclass import _nested_info_types + + nested = _nested_info_types(cls) + preset = VALUES.get(f'{cls.__module__.replace("pcapkit.", "", 1)}:{cls.__qualname__}', {}) + values = {} # type: dict[str, Any] + for key in annotated_keys(cls): + if key in preset: + values[key] = preset[key]() + elif key in nested and depth < MAX_DEPTH: + values[key] = synthesise(nested[key], depth + 1) + else: + values[key] = f'{cls.__qualname__}.{key}' + # The generated ``__init__`` takes the annotated keys as parameters, but a + # model may annotate a key no parameter can be named -- IPv4's + # ``OptionType`` injects ``class`` -- and is then built through + # ``from_dict`` as the parsers build it. + params = inspect.signature(cls.__init__).parameters + if cls.__dict__.get('__final__') and all(key in params for key in values): + return cls(**values) + return cls.from_dict(values) + + +def _structure(info: 'Any') -> 'Any': + """The class of ``info`` and of every nested :class:`Info`, recursively.""" + from pcapkit.corekit.infoclass import Info + + return (type(info).__qualname__, + tuple((key, _structure(value) if isinstance(value, Info) else type(value).__qualname__) + for key, value in info.items(multi=True))) + + +def _plain_dicts(value: 'Any', path: 'str' = '') -> 'list[str]': + """Paths in a ``to_dict()`` result of every mapping that should be an InfoDict. + + A nested :class:`Info` exports as an InfoDict. A + :class:`~pcapkit.corekit.multidict.MultiDict` held as a field value -- a + PCAP-NG block's ``records``, say -- is a value, not an export, and is kept. + + """ + from pcapkit.corekit.multidict import InfoDict, MultiDict + + if not isinstance(value, InfoDict): + return [f'{path or ""}: {type(value).__module__}.{type(value).__qualname__}'] + found = [] # type: list[str] + for key, item in value.items(multi=True): + if isinstance(item, InfoDict) or (isinstance(item, dict) and not isinstance(item, MultiDict)): + found.extend(_plain_dicts(item, f'{path}.{key}')) + return found + + +def check(info: 'Any') -> 'Outcome': + """``from_dict(to_dict())`` reproduces ``info`` in every respect listed above.""" + try: + exported = info.to_dict() + except Exception as exc: # pylint: disable=broad-except + return Outcome('TO_DICT', describe(exc)) + plain = _plain_dicts(exported) + if plain: + return Outcome('TO_DICT', f'not an InfoDict: {plain}') + try: + rebuilt = type(info).from_dict(exported) + except Exception as exc: # pylint: disable=broad-except + return Outcome('FROM_DICT', describe(exc)) + if rebuilt != info: + return Outcome('EQUAL', f'{rebuilt!r} != {info!r}') + if list(rebuilt.items(multi=True)) != list(info.items(multi=True)): + return Outcome('MULTI', f'{list(rebuilt.items(multi=True))!r} != {list(info.items(multi=True))!r}') + if _structure(rebuilt) != _structure(info): + return Outcome('TYPE', f'{_structure(rebuilt)!r} != {_structure(info)!r}') + if rebuilt.to_dict() != exported or list(rebuilt.to_dict().items(multi=True)) != list(exported.items(multi=True)): + return Outcome('TO_DICT', 'to_dict() of the rebuilt instance differs') + return OK + + +def _with(info: 'Any', key: 'str', value: 'Any', repeat: 'bool' = False) -> 'Any': + """``info`` rebuilt with ``key`` set to ``value``, or added again if ``repeat``.""" + from pcapkit.corekit.multidict import OrderedMultiDict + + items = OrderedMultiDict() # type: OrderedMultiDict[str, Any] + for name, item in info.to_dict().items(multi=True): + items.add(name, item) + items.add(key, value) + if not repeat: + return type(info).from_dict(list(items.items(multi=True))) + return type(info).from_dict(items) + + +def run_case(cls: 'type', variant: 'str') -> 'Outcome': + """Synthesise ``cls`` and check ``variant`` of it.""" + from pcapkit.corekit.infoclass import _nested_info_types + + try: + info = synthesise(cls) + except Exception as exc: # pylint: disable=broad-except + return Outcome('BUILD', describe(exc)) + try: + if variant == 'multi': + keys = annotated_keys(cls) + nested = _nested_info_types(cls) + key = next((key for key in keys if key in nested), keys[0]) + again = synthesise(nested[key], 1) if key in nested else f'{cls.__qualname__}.{key}.again' + info = _with(info, key, again, repeat=True) + if len(info.to_dict().getlist(key)) != 2: + return Outcome('MULTI', f'{key!r} holds {info.to_dict().getlist(key)!r}') + elif variant == 'short-read': + info = _with(info, '__short_read__', (annotated_keys(cls)[0], 1)) + if info.to_dict().get('__short_read__') != (annotated_keys(cls)[0], 1): + return Outcome('TO_DICT', '__short_read__ is not carried by to_dict()') + elif variant == 'truncated-raw': + info = _with(info, '__truncated_raw__', b'\x0a\x0d\x0d') + if info.to_dict().get('__truncated_raw__') != b'\x0a\x0d\x0d': + return Outcome('TO_DICT', '__truncated_raw__ is not carried by to_dict()') + except Exception as exc: # pylint: disable=broad-except + return Outcome('BUILD', describe(exc)) + return check(info) + + +class InfoRoundTripTests(KnownFailureTable, unittest.TestCase): + """``Info.from_dict(info.to_dict())`` for every :class:`Info` subclass.""" + + STATUSES = ('OK', 'BUILD', 'TO_DICT', 'FROM_DICT', 'EQUAL', 'MULTI', 'TYPE', 'TIMEOUT') + + KNOWN_FAILURES = ( + Gap(1490, 'A non-final model declared under a non-final parent is never finalised ' + 'once the parent has been: Info.__new__ auto-finalises only while ' + '__finalised__ reads NONE, and it reads the BASE the parent was given ' + 'through inheritance (pcapkit/corekit/infoclass.py:435, :223). The class keeps ' + 'the __excluded__ list it copied at class creation, before info_final added ' + 'the bookkeeping names to the parent\'s (pcapkit/corekit/infoclass.py:272-280, ' + ':146), so to_dict() emits __map__, __map_reverse__ and __multi__ as keys. ' + 'Order-dependent: the census instantiates each parent first', + 'TO_DICT', "'.__map__: builtins.dict'", + ('protocols.data.internet.hopopt:QuickStartOption/*', + 'protocols.data.internet.hopopt:SMFDPDOption/*', + 'protocols.data.internet.ipv6_opts:QuickStartOption/*', + 'protocols.data.internet.ipv6_opts:SMFDPDOption/*', + 'protocols.data.internet.ipv4:QSOption/*', + 'protocols.data.transport.tcp:MPTCPJoin/*')), + ) + + def setUp(self) -> None: + skip_without_runtime(self) + reimport_once_per_class(self) + + def _cases(self) -> 'dict[str, tuple[type, str]]': + from pcapkit.protocols.data.data import Data + from pcapkit.protocols.data.misc.pcapng import PCAPNG + + cases = {} # type: dict[str, tuple[type, str]] + for label, cls in info_classes().items(): + if not annotated_keys(cls): + continue + variants = ['plain', 'multi'] + if issubclass(cls, Data): + variants.append('short-read') + if issubclass(cls, PCAPNG): + variants.append('truncated-raw') + for variant in variants: + cases[f'{label}/{variant}'] = (cls, variant) + return cases + + def test_tables_name_real_cases(self) -> None: + self.check_table(self._cases()) + + def test_the_census_covers_the_data_models(self) -> None: + """The census reaches the protocol models, and every keyed class is a case.""" + classes = info_classes() + keyed = {label for label, cls in classes.items() if annotated_keys(cls)} + covered = {label.rsplit('/', 1)[0] for label in self._cases()} + self.assertEqual(keyed, covered) + self.assertGreater(sum(label.startswith('protocols.data.') for label in keyed), 400) + + def test_from_dict_of_to_dict_reproduces_the_instance(self) -> None: + cases = self._cases() + gaps = self.gap_table(cases) + for label, (cls, variant) in sorted(cases.items()): + with self.subTest(case=label): + self.check_outcome(label, run(run_case, cls, variant), gaps) + + +class InfoMechanicsTests(unittest.TestCase): + """The :class:`Info` shapes the synthesised census cannot produce.""" + + def setUp(self) -> None: + skip_without_runtime(self) + reimport_once_per_class(self) + + def _classes(self) -> 'tuple[type, type]': + from typing import Optional + + from pcapkit.corekit.infoclass import Info, info_final + + @info_final + class Leaf(Info): + """A nested model.""" + + x: int + + @info_final + class Node(Info): + """Nested, optional, renamed and plain keys together.""" + + leaf: Leaf + maybe: Optional[Leaf] + items: int + keys: str + plain: bytes + + return Leaf, Node + + def test_renamed_optional_and_nested_keys(self) -> None: + Leaf, Node = self._classes() + for maybe in (Leaf(x=2), None): + with self.subTest(maybe=maybe): + info = Node(leaf=Leaf(x=1), maybe=maybe, items=3, keys='k', plain=b'\x00\xff') + self.assertEqual(check(info), OK) + rebuilt = Node.from_dict(info.to_dict()) + self.assertEqual(rebuilt['items'], 3) + self.assertEqual(list(rebuilt), ['leaf', 'maybe', 'items', 'keys', 'plain']) + + def test_repeated_nested_values_stay_nested_and_ordered(self) -> None: + """#1466: every value of a repeated key, in order, rebuilt into its class.""" + from pcapkit.corekit.multidict import OrderedMultiDict + + Leaf, Node = self._classes() + source = OrderedMultiDict() # type: OrderedMultiDict[str, Any] + for key, value in (('leaf', {'x': 1}), ('maybe', None), ('items', 3), ('leaf', {'x': 2}), + ('keys', 'k'), ('leaf', {'x': 3}), ('plain', b'')): + source.add(key, value) + info = Node.from_dict(source) + leaves = [value for key, value in info.items(multi=True) if key == 'leaf'] + self.assertEqual([type(leaf) for leaf in leaves], [Leaf] * 3) + self.assertEqual([leaf.x for leaf in leaves], [1, 2, 3]) + self.assertEqual([leaf['x'] for leaf in info.to_dict().getlist('leaf')], [1, 2, 3]) + self.assertEqual(check(info), OK) + + def test_an_intermediate_model_after_its_parent(self) -> None: + """Pins the root cause of the census gap above directly (#1490). + + Expected-failure style: it asserts the defect, so a fix turns it red and + has to flip it together with the census entry. + + """ + from pcapkit.corekit.infoclass import Info + + class Parent(Info): + """A non-final model.""" + + a: int + + class Child(Parent): + """A non-final model under it.""" + + b: int + + Parent.from_dict({'a': 1}) + child = Child.from_dict({'a': 1, 'b': 2}) + self.assertEqual(list(Parent.from_dict({'a': 1}).to_dict()), ['a']) + self.assertEqual(list(child.to_dict()), ['__map__', '__map_reverse__', '__multi__', 'a', 'b']) + + def test_bookkeeping_keys_survive_a_second_round(self) -> None: + """#1465 and #1475: the dunder keys survive ``to_dict`` -> ``from_dict`` twice.""" + _, Node = self._classes() + info = Node.from_dict({'leaf': {'x': 1}, 'maybe': None, 'items': 0, 'keys': '', 'plain': b'', + '__short_read__': ('plain', 0), '__truncated_raw__': b'\x01'}) + twice = Node.from_dict(Node.from_dict(info.to_dict()).to_dict()) + self.assertEqual(twice.to_dict()['__short_read__'], ('plain', 0)) + self.assertEqual(twice.to_dict()['__truncated_raw__'], b'\x01') + self.assertEqual(check(info), OK) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/corekit/test_schema_roundtrip_unit.py b/tests/corekit/test_schema_roundtrip_unit.py new file mode 100644 index 0000000000..00bb7e74ef --- /dev/null +++ b/tests/corekit/test_schema_roundtrip_unit.py @@ -0,0 +1,671 @@ +# -*- coding: utf-8 -*- +"""Every protocol schema packs, unpacks and re-packs to the same octets. C.f. #1202. + +Each concrete :class:`~pcapkit.protocols.schema.schema.Schema` subclass under +:mod:`pcapkit.protocols.schema` is enumerated, so a schema added tomorrow gets a +case tomorrow. For each one a **minimal valid instance** ``s`` is built and +packed to ``b``, and four things must hold: + +* ``Schema.unpack(b)`` succeeds and accounts for exactly ``b`` (``SELF``); +* the parsed schema, packed again, is ``b`` (``REPACK``); +* ``from_dict(to_dict())`` of the parsed schema packs ``b`` (``DICT``); +* ``from_dict(to_dict())`` of ``s`` itself packs ``b`` (``DICT``). + +The minimal instance is built from the field types: zero for a number, the +first registry member (or the schema's own code) for an enumeration, empty for a +variable-width field, zeros for a fixed one, a nested minimal instance for a +:class:`~pcapkit.corekit.fields.misc.SchemaField`. A number another field's +length, condition or selector reads -- found by reading the names those +callbacks subscript -- is a *driver*, and the drivers are searched over small +values until the instance closes; a name the schema reads from its enclosing +layer is supplied the same way, as packet context. Where none closes, the first +instance that packed is the one reported. :data:`OVERRIDES` holds the few +values no field type implies (a magic number, a NUL-terminated name). + +A *dispatcher* -- a schema whose ``post_process`` returns the schema it +selected, such as TCP's ``_MPTCP`` -- is checked through its targets instead: +each target's minimal octets, parsed through the dispatcher, must come back as +that target and pack the same octets. + +A case that does not close is a defect and goes in ``KNOWN_FAILURES`` (see +:mod:`tests.corekit._roundtrip`). Nothing imports :mod:`pcapkit` at module +level. + +""" + +from __future__ import annotations + +import importlib +import inspect +import ipaddress +import itertools +import pkgutil +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class +from tests.corekit._roundtrip import (OK, Gap, KnownFailureTable, Outcome, describe, diff, run, + skip_without_runtime) + +if TYPE_CHECKING: + from typing import Any, Callable, Optional + +#: Package prefix stripped from a schema's module to make its label. +PREFIX = 'pcapkit.protocols.schema.' + +#: Driver values tried for one driver, for two, and for three or more. +ONE_DRIVER = tuple(range(65)) +TWO_DRIVERS = (0, 1, 2, 3, 4, 6, 8, 12, 16, 20, 24, 32) +MANY_DRIVERS = (0, 1, 2, 4, 8, 16) +#: Ceiling on driver combinations tried per schema. +MAX_COMBINATIONS = 400 + +#: Field values no field type implies, by schema label. A value is either the +#: value itself or a callable taking the schema's module. +OVERRIDES = { + 'misc.pcap.header:Header': {'magic_number': b'\xa1\xb2\xc3\xd4'}, + # ``match`` left unset, so ``pre_pack`` seeds the byte-order magic. + 'misc.pcapng:SectionHeaderBlock': {'match': None, 'magic': 0x1A2B3C4D}, + 'misc.pcapng:IPv4Record': {'resol': 'a\x00'}, + 'misc.pcapng:IPv6Record': {'resol': 'a\x00'}, + # The hash-assist value's first bit is the H-DPD mode bit [RFC 6621]. + 'internet.hopopt:SMFHashBasedDPDOption': {'hav': b'\x80'}, + 'internet.ipv6_opts:SMFHashBasedDPDOption': {'hav': b'\x80'}, +} # type: dict[str, dict[str, Any]] + +#: Packet context a dispatch target needs from its enclosing layer. +DISPATCH_CONTEXT = { + 'transport.tcp:MPTCPJoinSYN': {'flags': {'syn': 1, 'ack': 0}}, + 'transport.tcp:MPTCPJoinSYNACK': {'flags': {'syn': 1, 'ack': 1}}, + 'transport.tcp:MPTCPJoinACK': {'flags': {'syn': 0, 'ack': 1}}, +} # type: dict[str, dict[str, Any]] + +#: Dispatchers, to the base class whose concrete subclasses they select. +DISPATCHERS = { + 'internet.hopopt:_QuickStartOption': 'internet.hopopt:QuickStartOption', + 'internet.hopopt:_SMFDPDOption': 'internet.hopopt:SMFDPDOption', + 'internet.ipv6_opts:_QuickStartOption': 'internet.ipv6_opts:QuickStartOption', + 'internet.ipv6_opts:_SMFDPDOption': 'internet.ipv6_opts:SMFDPDOption', + 'internet.ipv4:_QSOption': 'internet.ipv4:QSOption', + 'transport.tcp:_MPTCP': 'transport.tcp:MPTCP', +} + +#: Names of a bit-field subfield that carries a schema's own registry code. +CODE_SUBFIELDS = ('subtype', 'func', 'mode') + +#: Schemas no minimal instance can be built for, to why. Each must still fail +#: to build, so one that starts building leaves this table. +UNBUILDABLE = {} # type: dict[str, str] + + +class AutoZero(int): + """An :obj:`int` that answers any subscript with zero. + + It stands in for a name a schema reads from its enclosing layer, whose + shape the schema alone does not say: ``pkt['flags']['bit_3']`` and + ``pkt['length'] - 2`` both work on it. + + """ + + def __getitem__(self, key: 'Any') -> 'int': + return 0 + + def get(self, key: 'Any', default: 'Any' = None) -> 'int': # pylint: disable=unused-argument + """Zero, as for a subscript.""" + return 0 + + +def label_of(cls: 'type') -> 'str': + """``module:Qualname``, relative to :data:`PREFIX`.""" + return f'{cls.__module__.replace(PREFIX, "")}:{cls.__qualname__}' + + +def schema_classes() -> 'dict[str, type]': + """Every :class:`Schema` subclass defined under :mod:`pcapkit.protocols.schema`.""" + import pcapkit.protocols.schema as package + from pcapkit.protocols.schema.schema import Schema + + found = {} # type: dict[str, type] + for info in pkgutil.walk_packages(package.__path__, f'{package.__name__}.'): + module = importlib.import_module(info.name) + for obj in vars(module).values(): + if inspect.isclass(obj) and issubclass(obj, Schema) and obj.__module__ == module.__name__: + found[label_of(obj)] = obj + return found + + +def _strings(fn: 'Any', depth: 'int' = 0, out: 'Optional[set[str]]' = None) -> 'set[str]': + """The string constants and names a callback's code mentions.""" + out = set() if out is None else out + code = getattr(fn, '__code__', None) + if depth > 4 or code is None: + return out + stack = [code] + while stack: + current = stack.pop() + out.update(current.co_names) + for const in current.co_consts: + if isinstance(const, str): + out.add(const) + elif hasattr(const, 'co_consts'): + stack.append(const) + for cell in getattr(fn, '__closure__', None) or (): + try: + value = cell.cell_contents + except ValueError: + continue + if callable(value): + _strings(value, depth + 1, out) + return out + + +def _callbacks(field: 'Any') -> 'list[Any]': + from pcapkit.corekit.fields.field import FieldBase + + found = [getattr(field, attr) for attr in ('_length_callback', '_condition', '_selector', '_callback') + if getattr(field, attr, None) is not None] + inner = getattr(field, '_field', None) + if isinstance(inner, FieldBase): + found.extend(_callbacks(inner)) + return found + + +def _unwrap(field: 'Any') -> 'Any': + from pcapkit.corekit.fields.misc import ConditionalField + + return field.field if isinstance(field, ConditionalField) else field + + +def drivers(cls: 'type') -> 'list[tuple[str, Optional[str], int]]': + """``(field, subfield, max)`` for every number another field's callback reads.""" + from pcapkit.corekit.fields.numbers import EnumField, NumberField + from pcapkit.corekit.fields.strings import BitField + + names = set() # type: set[str] + for field in cls.__fields__.values(): + for fn in _callbacks(field): + names |= _strings(fn) + + found = [] # type: list[tuple[str, Optional[str], int]] + for name, field in cls.__fields__.items(): + field = _unwrap(field) + if name not in names or (isinstance(field, EnumField) and field._namespace is not None): # pylint: disable=protected-access + continue + if isinstance(field, NumberField) and not hasattr(field, '_flags'): + width = field._bit_length if field._bit_length >= 0 else 8 * max(field._length, 1) # pylint: disable=protected-access + found.append((name, None, (1 << width) - 1)) + elif isinstance(field, BitField): + for sub, (_, size) in field._namespace.items(): # pylint: disable=protected-access + if sub in names: + found.append((name, sub, (1 << size) - 1)) + return found + + +def own_code(cls: 'type') -> 'Any': + """The registry code ``cls`` is registered under, or an unassigned one.""" + from pcapkit.protocols.schema.schema import EnumSchema + + if not issubclass(cls, EnumSchema): + return None + try: + registry = cls.registry + except Exception: # pylint: disable=broad-except + return None + for code, target in list(registry.items()): + if target is cls: + return code + try: + default = registry.default_factory() if registry.default_factory else None + except Exception: # pylint: disable=broad-except + default = None + if default is cls: + taken = {int(code) for code in registry if isinstance(code, int)} + return next(value for value in itertools.count(1) if value not in taken) + return None + + +def registered_code(cls: 'type') -> 'Any': + """The code ``cls`` or its nearest registered ancestor is registered under.""" + for base in cls.__mro__: + try: + items = list(cls.registry.items()) + except Exception: # pylint: disable=broad-except + return None + for code, target in items: + if target is base: + return code + return None + + +def _code_type(cls: 'type') -> 'Optional[type]': + try: + return type(next(iter(cls.registry))) + except Exception: # pylint: disable=broad-except + return None + + +def outer_code(cls: 'type') -> 'Any': + """The code a dispatcher is registered under in its own module's registries.""" + from pcapkit.protocols.schema.schema import EnumSchema + + module = importlib.import_module(cls.__module__) + for obj in vars(module).values(): + if inspect.isclass(obj) and issubclass(obj, EnumSchema) and obj is not EnumSchema: + try: + items = list(obj.registry.items()) + except Exception: # pylint: disable=broad-except + continue + for code, target in items: + if target is cls: + return code + return None + + +def minimal(field: 'Any', cls: 'type', depth: 'int', ctx: 'dict[str, Any]') -> 'Any': # pylint: disable=too-many-return-statements,too-many-branches + """The minimal value of ``field``, as the field type implies it.""" + from pcapkit.corekit.fields import ipaddress as fip + from pcapkit.corekit.fields.collections import ListField + from pcapkit.corekit.fields.field import NO_VALUE + from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, NoValueField, + PayloadField, SchemaField, SwitchField) + from pcapkit.corekit.fields.numbers import EnumField, NumberField + from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField, StringField + + if isinstance(field, (ConditionalField, ForwardMatchField)): + return minimal(field.field, cls, depth, ctx) + if isinstance(field, NoValueField): + return None + if isinstance(field, SwitchField): + try: + return minimal(field._selector(ctx), cls, depth, ctx) # pylint: disable=protected-access + except Exception: # pylint: disable=broad-except + return None + if field.default is not NO_VALUE and not isinstance(field, SchemaField): + return field.default + if hasattr(field, '_flags'): + return {name: 0 for name in field._flags} # pylint: disable=protected-access + if isinstance(field, EnumField): + namespace = field._namespace # pylint: disable=protected-access + code, kind = own_code(cls), _code_type(cls) + if code is not None and namespace is not None and kind is not None and ( + issubclass(kind, namespace) or issubclass(namespace, kind)): + return code + if namespace is None: + return 0 + members = getattr(namespace, '_value2member_map_', {}) + return members[0] if 0 in members else next(iter(namespace), 0) + if isinstance(field, NumberField): + return 0 + if isinstance(field, BitField): + return {name: 0 for name in field._namespace} # pylint: disable=protected-access + if isinstance(field, PaddingField): + return b'' + if isinstance(field, (StringField, BytesField)): + width = field._length if field._length_callback is None else -1 # pylint: disable=protected-access + if isinstance(field, StringField): + return '\x00' * max(width, 0) + return bytes(max(width, 0)) + for kind, value in ((fip.IPv4AddressField, ipaddress.IPv4Address(0)), + (fip.IPv6AddressField, ipaddress.IPv6Address(0)), + (fip.IPv4InterfaceField, ipaddress.IPv4Interface('0.0.0.0/0')), + (fip.IPv6InterfaceField, ipaddress.IPv6Interface('::/0'))): + if isinstance(field, kind): + return value + if isinstance(field, ListField): + # A declared area is filled with zeros, which every option registry here + # reads as padding or end-of-option-list; an undeclared one is empty. + try: + width = field(ctx).length + except Exception: # pylint: disable=broad-except + width = -1 + return bytes(width) if width > 0 else [] + if isinstance(field, PayloadField): + return b'' + if isinstance(field, SchemaField): + if field.default is not NO_VALUE: + return field.default + if depth > 3: + return None + try: + return build(field.schema, depth + 1)[0] + except Exception: # pylint: disable=broad-except + return None + return None + + +def _base_values(cls: 'type', depth: 'int', ctx: 'dict[str, Any]') -> 'dict[str, Any]': + from pcapkit.corekit.fields.numbers import EnumField + from pcapkit.corekit.fields.strings import BitField + + values = {} # type: dict[str, Any] + for name, field in cls.__fields__.items(): + values[name] = minimal(field, cls, depth, {**ctx, **values, '__length__': -1}) + + # A dispatch target carries the dispatcher's code in its type field, and its + # own code in the bit-field subfield the dispatcher selects on. + for dispatcher, base in DISPATCHERS.items(): + if not label_of(cls).startswith(dispatcher.split(':')[0] + ':'): + continue + base_cls = getattr(importlib.import_module(cls.__module__), base.split(':')[1]) + if not issubclass(cls, base_cls): + continue + outer = outer_code(getattr(importlib.import_module(cls.__module__), dispatcher.split(':')[1])) + code = registered_code(cls) + code = own_code(cls) if code is None else code + for name, field in cls.__fields__.items(): + field = _unwrap(field) + if outer is not None and isinstance(field, EnumField) and field._namespace is not None \ + and isinstance(outer, field._namespace): # pylint: disable=protected-access + values[name] = outer + if code is not None and isinstance(field, BitField): + for sub in CODE_SUBFIELDS: + if sub in field._namespace: # pylint: disable=protected-access + values[name] = {**values[name], sub: int(code)} + + module = importlib.import_module(cls.__module__) + for name, value in OVERRIDES.get(label_of(cls), {}).items(): + values[name] = value(module) if callable(value) else value + return values + + +def _attempt(cls: 'type', depth: 'int', ctx: 'dict[str, Any]', combo: 'tuple[int, ...]', + drv: 'list[tuple[str, Optional[str], int]]') -> 'tuple[Any, bytes, dict[str, Any]]': + packet = dict(ctx) + for (name, _, _), value in zip(drv, combo): + if name in ctx: + packet[name] = AutoZero(value) + values = _base_values(cls, depth, packet) + for (name, sub, _), value in zip(drv, combo): + if name not in cls.__fields__: + continue + if sub is None: + values[name] = value + else: + values[name] = {**values[name], sub: value} + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + schema = cls.from_dict({key: value for key, value in values.items() if value is not None}) + raw = schema.pack(dict(packet)) + return schema, raw, packet + + +def build(cls: 'type', depth: 'int' = 0, extra: 'Optional[dict[str, Any]]' = None, + lengths: 'bool' = False, accept: 'Optional[Callable[..., Outcome]]' = None) -> 'Any': + """``(schema, octets, context)`` for a minimal instance of ``cls``. + + At ``depth`` 0 the first driver combination whose instance closes -- by + :func:`check`, or by ``accept`` -- is returned; if none closes, + ``(schema, outcome)`` of the first that packed. ``lengths`` adds the + schema's own ``length``/``len`` field to the drivers, for a dispatch target + whose length only the dispatcher reads. + + Raises: + RuntimeError: If no combination packs at all. + + """ + ctx = dict(extra or {}) + first = None # type: Any + for _ in range(6): + drv = drivers(cls) + [(name, None, 0xFFFF) for name in ctx if name not in (extra or {})] + if lengths: + drv += [(name, None, 0xFF) for name in ('length', 'len') + if name in cls.__fields__ and all(name != item[0] for item in drv)] + choices = ONE_DRIVER if len(drv) == 1 else TWO_DRIVERS if len(drv) == 2 else MANY_DRIVERS + missing = None + tried = 0 + for combo in itertools.product(choices, repeat=len(drv)): + if any(value > top for (_, _, top), value in zip(drv, combo)): + continue + tried += 1 + if tried > MAX_COMBINATIONS: + break + try: + schema, raw, packet = _attempt(cls, depth, ctx, combo, drv) + except KeyError as exc: + key = exc.args[0] if exc.args else None + if isinstance(key, str) and key not in ctx and key not in cls.__fields__: + missing = key + break + first = first if first is not None else exc + continue + except Exception as exc: # pylint: disable=broad-except + first = first if first is not None else exc + continue + if depth: + return schema, raw, packet + outcome = (accept or check)(cls, schema, raw, packet) + if outcome.status == 'OK': + return schema, raw, packet + if not isinstance(first, tuple): + first = (schema, outcome) + if missing is None: + break + ctx[missing] = AutoZero(0) + if not depth and isinstance(first, tuple): + return first + raise RuntimeError(f'no minimal instance packs (context {sorted(ctx)}): ' + f'{describe(first) if isinstance(first, BaseException) else first}') + + +def check(cls: 'type', schema: 'Any', raw: 'bytes', ctx: 'dict[str, Any]') -> 'Outcome': + """The four assertions of the module docstring, for one built instance.""" + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + parsed = cls.unpack(raw, len(raw), dict(ctx)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('REPARSE', describe(exc)) + if bytes(parsed) != raw: + return Outcome('SELF', diff(bytes(parsed), raw)) + try: + repacked = parsed.pack(dict(ctx)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('REPACK', describe(exc)) + if repacked != raw: + return Outcome('REPACK', diff(repacked, raw)) + for what, source in (('parsed', parsed), ('built', schema)): + try: + rebuilt = type(source).from_dict(source.to_dict()).pack(dict(ctx)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('DICT', f'{what}: {describe(exc)}') + if rebuilt != raw: + return Outcome('DICT', f'{what}: {diff(rebuilt, raw)}') + return OK + + +def _referenced(cls: 'type') -> 'set[str]': + """Every name a callback of any schema in ``cls``'s module reads. + + The whole module rather than ``cls`` alone, since a nested schema reads its + enclosing schema's fields through the packet context -- HTTP/2's ``DATA`` + frame reads the frame header's ``flags``. + + """ + from pcapkit.protocols.schema.schema import Schema + + names = set(OVERRIDES.get(label_of(cls), ())) + for obj in vars(importlib.import_module(cls.__module__)).values(): + if inspect.isclass(obj) and issubclass(obj, Schema): + for field in obj.__fields__.values(): + for fn in _callbacks(field): + names |= _strings(fn) + return names + + +def patterned(cls: 'type', schema: 'Any', ctx: 'dict[str, Any]') -> 'Outcome': + """The minimal instance with every free value made non-zero, checked again. + + Zeros survive a reversed, shifted or dropped octet unchanged, so a minimal + instance alone cannot see such a defect. A value is *free* when no callback + in the schema's module reads its name and :data:`OVERRIDES` does not set it: a plain number becomes ``1``, a fixed-width + octet string ``01 02 ...``, every subfield of a bit field ``1``, an address + ``10.0.0.1`` or ``fe80::1``. + + """ + from pcapkit.corekit.fields import ipaddress as fip + from pcapkit.corekit.fields.numbers import EnumField, NumberField + from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField + + names = _referenced(cls) + values = dict(schema.to_dict()) + for name, field in cls.__fields__.items(): + field = _unwrap(field) + if name in names or name not in values or values[name] is None: + continue + if isinstance(field, EnumField) or hasattr(field, '_flags'): + continue + if isinstance(field, NumberField): + values[name] = 1 + elif isinstance(field, BitField): + values[name] = {sub: 1 for sub in field._namespace} # pylint: disable=protected-access + elif isinstance(field, BytesField) and not isinstance(field, PaddingField) \ + and field._length_callback is None and field._length > 0: # pylint: disable=protected-access + values[name] = bytes(index % 255 + 1 for index in range(field._length)) # pylint: disable=protected-access + elif isinstance(field, fip.IPv4AddressField): + values[name] = ipaddress.IPv4Address('10.0.0.1') + elif isinstance(field, fip.IPv6AddressField): + values[name] = ipaddress.IPv6Address('fe80::1') + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + built = type(schema).from_dict(values) + raw = built.pack(dict(ctx)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('MAKE', describe(exc)) + return check(type(schema), built, raw, ctx) + + +def run_patterned(cls: 'type') -> 'Outcome': + """Build ``cls`` and check its patterned variant.""" + try: + out = build(cls) + except RuntimeError as exc: + return Outcome('UNBUILT', str(exc)) + if len(out) == 2: + return Outcome('UNBUILT', f'minimal instance does not close: {out[1].status}') + return patterned(cls, out[0], out[2]) + + +def run_schema(cls: 'type') -> 'Outcome': + """Build ``cls`` and check it.""" + try: + out = build(cls) + except RuntimeError as exc: + return Outcome('UNBUILT', str(exc)) + if len(out) == 2: + return out[1] + return OK + + +def check_dispatch(dispatcher: 'type', target: 'type', raw: 'bytes', ctx: 'dict[str, Any]') -> 'Outcome': + """``raw``, parsed through ``dispatcher``, is ``target`` and packs ``raw``.""" + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + try: + parsed = dispatcher.unpack(raw, len(raw), dict(ctx)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('DISPATCH', describe(exc)) + if type(parsed) is not target: + return Outcome('DISPATCH', f'parsed as {type(parsed).__qualname__}') + try: + repacked = parsed.pack(dict(ctx)) + rebuilt = type(parsed).from_dict(parsed.to_dict()).pack(dict(ctx)) + except Exception as exc: # pylint: disable=broad-except + return Outcome('REPACK', describe(exc)) + if repacked != raw: + return Outcome('REPACK', diff(repacked, raw)) + if rebuilt != raw: + return Outcome('DICT', diff(rebuilt, raw)) + return OK + + +def run_dispatch(dispatcher: 'type', target: 'type') -> 'Outcome': + """A minimal ``target`` that closes on its own, parsed through ``dispatcher``.""" + def accept(cls: 'type', schema: 'Any', raw: 'bytes', ctx: 'dict[str, Any]') -> 'Outcome': + outcome = check(cls, schema, raw, ctx) + return check_dispatch(dispatcher, cls, raw, ctx) if outcome.status == 'OK' else outcome + + try: + out = build(target, extra=DISPATCH_CONTEXT.get(label_of(target), {}), lengths=True, accept=accept) + except RuntimeError as exc: + return Outcome('UNBUILT', str(exc)) + return out[1] if len(out) == 2 else OK + + +class SchemaRoundTripTests(KnownFailureTable, unittest.TestCase): + """Schema ``pack``/``unpack`` and ``from_dict``/``to_dict`` symmetry.""" + + STATUSES = ('OK', 'UNBUILT', 'MAKE', 'REPARSE', 'SELF', 'REPACK', 'DICT', 'DISPATCH', 'TIMEOUT') + + KNOWN_FAILURES = () # type: tuple[Gap, ...] + + def setUp(self) -> None: + skip_without_runtime(self) + reimport_once_per_class(self) + + def _cases(self) -> 'tuple[dict[str, type], dict[str, tuple[type, type]]]': + classes = schema_classes() + plain = {label: cls for label, cls in classes.items() + if cls.__fields__ and label not in DISPATCHERS} + dispatch = {} # type: dict[str, tuple[type, type]] + for dispatcher, base in DISPATCHERS.items(): + for label, cls in classes.items(): + if cls.__fields__ and issubclass(cls, classes[base]) and cls.__dict__.get('__final__'): + dispatch[f'dispatch/{dispatcher}/{label}'] = (classes[dispatcher], cls) + return plain, dispatch + + def _labels(self) -> 'list[str]': + plain, dispatch = self._cases() + return list(plain) + [f'{label}/patterned' for label in plain] + list(dispatch) + + def test_tables_name_real_cases(self) -> None: + labels = self._labels() + self.check_table(labels) + _, dispatch = self._cases() + classes = schema_classes() + for table in (OVERRIDES, DISPATCHERS, DISPATCH_CONTEXT, UNBUILDABLE): + self.assertEqual(sorted(set(table) - set(classes)), [], 'stale table entries') + self.assertEqual(sorted(set(DISPATCHERS.values()) - set(classes)), []) + for dispatcher in DISPATCHERS: + self.assertTrue(any(label.startswith(f'dispatch/{dispatcher}/') for label in dispatch), + f'{dispatcher} has no targets') + + def test_the_census_is_the_whole_package(self) -> None: + """Every schema with fields is a case, a dispatcher, or the base classes.""" + plain, _ = self._cases() + classes = schema_classes() + fieldless = sorted(label for label, cls in classes.items() if not cls.__fields__) + self.assertEqual(len(plain) + len(DISPATCHERS) + len(fieldless), len(classes)) + self.assertGreater(len(plain), 400) + + def test_minimal_instance_round_trips(self) -> None: + plain, _ = self._cases() + gaps = self.gap_table(self._labels()) + for label, cls in sorted(plain.items()): + with self.subTest(case=label): + outcome = run(run_schema, cls) + if label in UNBUILDABLE: + self.assertEqual(outcome.status, 'UNBUILT', f'{label} builds now: drop it from UNBUILDABLE') + continue + self.check_outcome(label, outcome, gaps) + + def test_patterned_instance_round_trips(self) -> None: + plain, _ = self._cases() + gaps = self.gap_table(self._labels()) + for label, cls in sorted(plain.items()): + if label in UNBUILDABLE: + continue + with self.subTest(case=f'{label}/patterned'): + self.check_outcome(f'{label}/patterned', run(run_patterned, cls), gaps) + + def test_dispatchers_select_and_rebuild_each_target(self) -> None: + _, dispatch = self._cases() + gaps = self.gap_table(self._labels()) + for label, (dispatcher, target) in sorted(dispatch.items()): + with self.subTest(case=label): + self.check_outcome(label, run(run_dispatch, dispatcher, target), gaps) + + +if __name__ == '__main__': + unittest.main()