diff --git a/docs/source/pcapkit/corekit/infoclass.rst b/docs/source/pcapkit/corekit/infoclass.rst index f22187854a..3fc0c7d9c5 100644 --- a/docs/source/pcapkit/corekit/infoclass.rst +++ b/docs/source/pcapkit/corekit/infoclass.rst @@ -5,7 +5,10 @@ Info Class :mod:`pcapkit.corekit.infoclass` contains the :obj:`dict`-like class :class:`~pcapkit.corekit.infoclass.Info`, modelled on -:func:`dataclasses.dataclass` (:pep:`557`). +:func:`dataclasses.dataclass` (:pep:`557`), and the immutable multi-mapping +classes :class:`~pcapkit.corekit.infoclass.MultiInfo` and +:class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` that a finalised +:class:`~pcapkit.corekit.infoclass.Info` holds its option lists in. .. autoclass:: pcapkit.corekit.infoclass.Info :members: @@ -24,6 +27,14 @@ Info Class .. autodecorator:: pcapkit.corekit.infoclass.info_final +.. autoclass:: pcapkit.corekit.infoclass.MultiInfo + :no-members: + :show-inheritance: + +.. autoclass:: pcapkit.corekit.infoclass.OrderedMultiInfo + :no-members: + :show-inheritance: + Internal Definitions -------------------- @@ -31,9 +42,20 @@ Internal Definitions :no-members: :show-inheritance: +.. autoclass:: pcapkit.corekit.infoclass._MultiInfo + :no-members: + :show-inheritance: + +.. autoclass:: pcapkit.corekit.infoclass._OrderedMultiDict + :no-members: + :show-inheritance: + Type Variables -------------- +.. data:: pcapkit.corekit.infoclass.KT + :type: typing.Any + .. data:: pcapkit.corekit.infoclass.VT :type: typing.Any diff --git a/docs/source/pcapkit/corekit/multidict.rst b/docs/source/pcapkit/corekit/multidict.rst index ea18f2e6b3..67ae8aab10 100644 --- a/docs/source/pcapkit/corekit/multidict.rst +++ b/docs/source/pcapkit/corekit/multidict.rst @@ -35,15 +35,6 @@ values per key, based on the `Werkzeug`_ project. :no-special-members: __init__ :show-inheritance: -.. autoclass:: pcapkit.corekit.multidict.InfoDict - :no-members: - :no-special-members: __init__ - :show-inheritance: - - .. automethod:: keys - .. automethod:: values - .. automethod:: items - Type Variables -------------- diff --git a/pcapkit/all.py b/pcapkit/all.py index 21e61578ea..fe77f29d9a 100644 --- a/pcapkit/all.py +++ b/pcapkit/all.py @@ -53,6 +53,7 @@ # pcapkit.corekit 'EnumRegistry', # Enum Registry 'Info', 'info_final', # Info Class + 'MultiInfo', 'OrderedMultiInfo', 'ProtoChain', # ProtoChain 'VersionInfo', # Version 'NumberField', 'Int32Field', 'UInt32Field', # numeric protocol fields diff --git a/pcapkit/corekit/__init__.py b/pcapkit/corekit/__init__.py index e05bed5f39..c9058a5646 100644 --- a/pcapkit/corekit/__init__.py +++ b/pcapkit/corekit/__init__.py @@ -19,7 +19,7 @@ class :class:`~pcapkit.corekit.infoclass.Info`, from pcapkit.corekit.io import SeekableReader from pcapkit.corekit.enum import EnumRegistry from pcapkit.corekit.fields import * -from pcapkit.corekit.infoclass import Info, info_final +from pcapkit.corekit.infoclass import Info, MultiInfo, OrderedMultiInfo, info_final from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict from pcapkit.corekit.protochain import ProtoChain @@ -29,6 +29,7 @@ class :class:`~pcapkit.corekit.infoclass.Info`, 'EnumRegistry', 'Info', 'info_final', + 'MultiInfo', 'OrderedMultiInfo', 'ProtoChain', diff --git a/pcapkit/corekit/infoclass.py b/pcapkit/corekit/infoclass.py index fede71e905..b411e364f6 100644 --- a/pcapkit/corekit/infoclass.py +++ b/pcapkit/corekit/infoclass.py @@ -5,13 +5,17 @@ .. module:: pcapkit.corekit.infoclass :mod:`pcapkit.corekit.infoclass` contains :obj:`dict` like class -:class:`~pcapkit.corekit.infoclass.Info` only, which is originally +:class:`~pcapkit.corekit.infoclass.Info`, which is originally designed to work alike :func:`dataclasses.dataclass` as introduced -in :pep:`557`. +in :pep:`557`, and the immutable multi-mapping classes +:class:`~pcapkit.corekit.infoclass.MultiInfo` and +:class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` that a finalised +:class:`~pcapkit.corekit.infoclass.Info` holds its option lists in. """ import abc import collections.abc +import contextlib import enum import itertools import sys @@ -20,7 +24,7 @@ from typing import TYPE_CHECKING, Generic, TypeVar from pcapkit.corekit.enum import EnumLookup -from pcapkit.corekit.multidict import InfoDict, MultiDict +from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict, iter_multi_items from pcapkit.utilities.compat import Mapping, final from pcapkit.utilities.exceptions import InfoError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import InfoWarning, warn @@ -30,8 +34,9 @@ from typing_extensions import Self -__all__ = ['Info', 'info_final'] +__all__ = ['Info', 'info_final', 'MultiInfo', 'OrderedMultiInfo'] +KT = TypeVar('KT') VT = TypeVar('VT') ST = TypeVar('ST', bound='Type[Info]') @@ -147,8 +152,10 @@ def info_final(cls: 'ST', *, _finalised: 'bool' = True) -> 'ST': # NOTE: We only generate ``__init__`` method for subclasses of the # ``Info`` class, rather than itself, plus that such class does not - # override the ``__init__`` method of the meta class. - if '__init__' not in cls.__dict__ and cls is not Info: + # override the ``__init__`` method of the meta class. A multi-mapping + # :class:`Info` (:class:`_MultiInfo`) declares no fields and keeps the + # mapping constructor it inherits (:issue:`1484`). + if '__init__' not in cls.__dict__ and cls is not Info and not issubclass(cls, _MultiInfo): args_ = [] # type: list[str] dict_ = [] # type: list[str] @@ -462,6 +469,11 @@ def __update__(self, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT] # value and records the new one in :attr:`__multi__`. Anything else # replaces the values held. + # NOTE: A :class:`MultiDict` or :class:`OrderedMultiDict` held as a + # *value* -- the option lists the parsers build -- is stored as a + # :class:`MultiInfo` or :class:`OrderedMultiInfo`, so that a finalised + # instance holds no mutable mapping (:issue:`1484`). + __name__ = type(self).__name__ # pylint: disable=redefined-builtin multi_iter = () # type: Iterable[tuple[str, Any]] @@ -479,7 +491,7 @@ def __update__(self, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT] new_key = f'_{__name__}{key}' if key in self.__builtin__ else key if new_key in self.__dict__ and new_key not in self.__excluded__: anchor = next(reversed(self.__dict__)) - self.__multi__.setdefault(anchor, []).append((new_key, value)) + self.__multi__.setdefault(anchor, []).append((new_key, _freeze(value))) else: self.__update__({key: value}) @@ -508,7 +520,7 @@ def __update__(self, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT] # if isinstance(key, str): # key = re.sub(r'\W', '_', key) - self.__dict__[key] = value + self.__dict__[key] = _freeze(value) __init__ = __update__ @@ -530,7 +542,9 @@ def __repr__(self) -> 'str': continue out_key = self.__map_reverse__.get(key, key) - if isinstance(value, Info): + # NOTE: a multi-mapping :class:`Info`, i.e. an option list, is + # written out in full, as the multi-mapping it is. + if isinstance(value, Info) and not isinstance(value, _MultiInfo): temp.append(f'{out_key}={type(value).__name__}(...)') else: temp.append(f'{out_key}={value!r}') @@ -601,6 +615,12 @@ def from_dict(cls, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT]]] such nested values. A key without such an annotation keeps the mapping as is -- see :meth:`to_dict`. + Every :class:`~pcapkit.corekit.multidict.MultiDict` or + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` value is stored + as a :class:`MultiInfo` or :class:`OrderedMultiInfo` respectively, or + as the subclass of either that the key's annotation names, such as + :class:`~pcapkit.protocols.data.application.httpv2.Settings`. + The rebuild is as unchecked as the rest of this method: a mapping whose keys are not the nested class's own is accepted and becomes an instance carrying those keys, exactly as a direct call to the nested class's own @@ -616,25 +636,41 @@ def from_dict(cls, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT]]] self = cls.__new__(cls) self.__update__(dict_, **kwargs) - for (name, info_cls) in _nested_info_types(cls).items(): + for (name, info_cls) in _nested_types(cls).items(): key = self.__map__.get(name, name) - value = self.__dict__.get(key) - if isinstance(value, Mapping) and not isinstance(value, Info): - self.__dict__[key] = info_cls.from_dict(value) + if key in self.__dict__: + self.__dict__[key] = _rebuild(info_cls, self.__dict__[key]) for values in self.__multi__.values(): for (index, (rkey, rvalue)) in enumerate(values): - if rkey == key and isinstance(rvalue, Mapping) and not isinstance(rvalue, Info): - values[index] = (rkey, info_cls.from_dict(rvalue)) + if rkey == key: + values[index] = (rkey, _rebuild(info_cls, rvalue)) return self - def to_dict(self) -> 'InfoDict[str, VT]': + def to_dict(self) -> 'OrderedMultiDict[str, VT]': """Convert :class:`Info` into :obj:`dict`. - The :obj:`dict` is an :class:`~pcapkit.corekit.multidict.InfoDict`. A key - held more than once -- as given by a + The :obj:`dict` is an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` of the fields, in + order -- a :class:`_OrderedMultiDict`, i.e. one whose ``!=`` is the + negation of its ``==``, so two equal exports compare ``!=`` + :data:`False`. A key held more than once -- as given by a :class:`~pcapkit.corekit.multidict.MultiDict` to :meth:`from_dict`, e.g. - a repeated IPv6 extension header -- keeps every value for - ``items(multi=True)``, while the :obj:`dict` interface sees the first. + a repeated IPv6 extension header -- keeps every value, in place, for + ``items(multi=True)`` and :meth:`~pcapkit.corekit.multidict.MultiDict.getlist`, + while indexing sees the first. A nested :class:`Info` becomes its own + :meth:`to_dict`: an :class:`~pcapkit.corekit.multidict.OrderedMultiDict` + for a field model or an :class:`OrderedMultiInfo`, and a + :class:`~pcapkit.corekit.multidict.MultiDict` for a :class:`MultiInfo`; + the values those two hold are not converted. + + Note: + ``dict(result)``, ``{**result}`` and :func:`json.dumps` see the + *first* value of each key only, and silently drop the rest of a + repeated key; read every value with ``items(multi=True)`` or + :meth:`~pcapkit.corekit.multidict.MultiDict.getlist`. The result + compares equal to another + :class:`~pcapkit.corekit.multidict.MultiDict` only, never to a + plain :obj:`dict`. Important: We only convert nested :class:`Info` objects into :obj:`dict` if @@ -649,10 +685,12 @@ def to_dict(self) -> 'InfoDict[str, VT]': any other key -- every key of a bare :class:`Info`, and annotations naming a type imported only under :data:`~typing.TYPE_CHECKING` -- the conversion is one-way, and the nested value comes back as the - :class:`~pcapkit.corekit.multidict.InfoDict` it was written as. + :class:`OrderedMultiInfo` that :meth:`from_dict` stores the + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` it was + written as in. """ - dict_ = InfoDict() # type: InfoDict[str, Any] + dict_ = _OrderedMultiDict() # type: OrderedMultiDict[str, Any] for (key, value) in self.__items_multi(): out_key = self.__map_reverse__.get(key, key) if isinstance(value, Info): @@ -672,9 +710,326 @@ def to_dict(self) -> 'InfoDict[str, VT]': return dict_ +def _ne(self: 'OrderedMultiDict[Any, Any]', other: 'object') -> 'bool': + """``self != other``, as the negation of ``self == other``. + + The ``__ne__`` that :class:`_OrderedMultiDict` and :class:`OrderedMultiInfo` + add to Werkzeug's :class:`~pcapkit.corekit.multidict.OrderedMultiDict`. A + :data:`NotImplemented` from ``__eq__`` is passed on, so that Python asks + ``other`` in turn, exactly as it does for ``==``. + + """ + result = self.__eq__(other) + if result is NotImplemented: + return result + return not result + + +class _OrderedMultiDict(OrderedMultiDict[KT, VT]): + """The :class:`~pcapkit.corekit.multidict.OrderedMultiDict` that :meth:`Info.to_dict` returns. + + It adds one method, :meth:`__ne__` (:func:`_ne`). The + :mod:`~pcapkit.corekit.multidict` classes are Werkzeug's, kept verbatim, + and Werkzeug's :class:`~pcapkit.corekit.multidict.OrderedMultiDict` + defines ``__eq__`` but no ``__ne__``: ``!=`` falls back to :obj:`dict`'s, + which compares the internal buckets, so two equal exports compared + ``!=`` as well as ``==`` -- and, since ``__eq__`` compares the values with + ``!=``, two exports holding a nested export compared unequal + (:issue:`1484`). Its copies, deep copies and pickles stay of this class, as + Werkzeug's build them from ``type(self)``, and :meth:`__repr__` writes it as + the :class:`~pcapkit.corekit.multidict.OrderedMultiDict` it is. + + """ + + __ne__ = _ne + + def __repr__(self) -> 'str': + return f'{OrderedMultiDict.__name__}({list(self.items(multi=True))!r})' + + +class _MultiInfo(Info[VT], Generic[KT, VT]): + """An :class:`Info` that holds its values as a multi-mapping. + + The common base of :class:`MultiInfo` and :class:`OrderedMultiInfo`. Each + of those also derives from the :mod:`~pcapkit.corekit.multidict` class it + is named after, *after* :class:`Info`, so it is an :class:`Info` -- + :func:`isinstance` holds, and it inherits :meth:`Info.__new__`'s + finalisation, :meth:`Info.from_dict`, :meth:`Info.__init_subclass__` and + :attr:`Info.__additional__`/:attr:`Info.__excluded__` -- while its values + live in the multi-mapping rather than in :attr:`~object.__dict__`. + + That is the whole carve-out from :class:`Info`, and it is confined to this + class and the two below it: + + * :meth:`__new__` gives the multi-mapping its empty storage, and + :meth:`__update__` *adds* every pair to it, so that the inherited + :meth:`Info.from_dict` builds one; :func:`info_final` generates no + ``__init__`` for such a class (it has no declared fields); + * :meth:`__setattr__` admits the multi-mapping's own bookkeeping, and + only while one of those two is writing; + * every mutator of the multi-mapping raises + :exc:`~pcapkit.utilities.exceptions.UnsupportedCall`, as assignment to + an :class:`Info` does; + * :meth:`to_dict` exports the plain multi-mapping, so that + ``cls.from_dict(info.to_dict()) == info``; + * :class:`MultiInfo` and :class:`OrderedMultiInfo` bind the mapping + protocol back to their multi-mapping class, which :class:`Info` and + :class:`~collections.abc.Mapping` precede in the MRO. + + A key also reads as an attribute, giving its first value: a :obj:`str` key + by its name, an :class:`~enum.Enum` key by its member name. A name the + class itself defines, such as ``items``, is a method and not a key. + + """ + + def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'Self': # pylint: disable=unused-argument + self = super().__new__(cls) + # NOTE: the multi-mapping class's own ``__init__``, which follows Info + # and Mapping in the MRO, gives the empty storage -- including + # OrderedMultiDict's bucket chain -- for ``__update__`` to add to. + with self.__writing(): # pylint: disable=protected-access + super(Info, self).__init__() # pylint: disable=bad-super-call + return self + + def __init__(self, mapping: 'Optional[Mapping[Any, Any] | Iterable[tuple[Any, Any]]]' = None) -> 'None': # pylint: disable=line-too-long,super-init-not-called + self.__update__(mapping) + self.__post_init__() + + def __update__(self, dict_: 'Optional[Mapping[Any, Any] | Iterable[tuple[Any, Any]]]' = None, + **kwargs: 'Any') -> 'None': + # NOTE: every pair is *added*, as the multi-mapping's own constructor + # and ``update`` do: a mapping's list values count as one pair each, + # and a MultiDict gives every value of each key. + pairs = () if dict_ is None else iter_multi_items(dict_) # type: Iterable[tuple[Any, Any]] + with self.__writing(): + for (key, value) in itertools.chain(pairs, kwargs.items()): + super(Info, self).add(key, value) # type: ignore[misc] # pylint: disable=bad-super-call,no-member + + @contextlib.contextmanager + def __writing(self) -> 'Iterator[None]': + """Admit attribute writes, for the multi-mapping's own bookkeeping.""" + object.__setattr__(self, '_MultiInfo__open', True) + try: + yield + finally: + object.__setattr__(self, '_MultiInfo__open', False) + + def __getattr__(self, name: 'str') -> 'Any': + # NOTE: only reached for a name that is not a real attribute, so a key + # named like a method is read by indexing instead, as for :class:`Info`. + if not (name.startswith('__') and name.endswith('__')): + if dict.__contains__(self, name): + return self[name] + for key in dict.keys(self): # type: ignore[arg-type,var-annotated] + if isinstance(key, enum.Enum) and key.name == name: + return self[key] # type: ignore[index] + raise UnsupportedCall(f'{type(self).__name__!r} object has no attribute {name!r}', + quiet=True) + + def __setattr__(self, name: 'str', value: 'Any') -> 'None': # type: ignore[override] + if not self.__dict__.get('_MultiInfo__open'): + raise UnsupportedCall("can't set attribute") + object.__setattr__(self, name, value) + + def _immutable(self, *args: 'Any', **kwargs: 'Any') -> 'NoReturn': + raise UnsupportedCall(f'{type(self).__name__!r} object is immutable') + + __setitem__ = __delitem__ = __ior__ = _immutable + add = setlist = setdefault = setlistdefault = update = _immutable + pop = popitem = poplist = popitemlist = clear = _immutable + __setstate__ = _immutable + + +class MultiInfo(_MultiInfo[KT, VT], MultiDict[KT, VT]): + """An immutable :class:`~pcapkit.corekit.multidict.MultiDict` that is an :class:`Info`. + + It holds every value of a key, as a + :class:`~pcapkit.corekit.multidict.MultiDict` does, but refuses every + mutation, as :class:`Info` does, with + :exc:`~pcapkit.utilities.exceptions.UnsupportedCall` (see + :class:`_MultiInfo` for how it is both). + + A finalised :class:`Info` holds every + :class:`~pcapkit.corekit.multidict.MultiDict` value as one of these, and + :meth:`to_dict` -- its own and its holder's -- turns it back into a + :class:`~pcapkit.corekit.multidict.MultiDict`. + + Args: + mapping: The initial value, as for + :class:`~pcapkit.corekit.multidict.MultiDict`. + + """ + + # NOTE: :class:`Info` and :class:`~collections.abc.Mapping` precede + # MultiDict in the MRO, so the mapping protocol is bound back to MultiDict's + # own (``__len__``, ``__contains__`` and ``__str__`` are those of + # :obj:`dict` and :obj:`object`). ``__eq__`` and ``__ne__`` are defined + # below instead. + __getitem__ = MultiDict.__getitem__ # type: ignore[assignment] + __iter__ = MultiDict.__iter__ # type: ignore[assignment] + __len__ = MultiDict.__len__ + __contains__ = MultiDict.__contains__ + if hasattr(dict, '__reversed__'): # pragma: no branch -- Python 3.8 and up + __reversed__ = MultiDict.__reversed__ + __hash__ = None # type: ignore[assignment] + __repr__ = MultiDict.__repr__ + __str__ = MultiDict.__str__ + __copy__ = MultiDict.__copy__ # type: ignore[assignment] + keys = MultiDict.keys + values = MultiDict.values + items = MultiDict.items # type: ignore[assignment] + get = MultiDict.get + + def __eq__(self, other: 'object') -> 'bool': + # NOTE: MultiDict's ``__eq__`` is :obj:`dict`'s, which compares the + # internal storage, so against an OrderedMultiDict -- whose values sit + # in buckets -- it answered False where the OrderedMultiDict, asked the + # other way round, answered True. The ordered side decides, as it does + # for a bare MultiDict, which Python asks it first for being its + # subclass (:issue:`1484`). + if isinstance(other, OrderedMultiDict): + return NotImplemented + return MultiDict.__eq__(self, other) + + def __ne__(self, other: 'object') -> 'bool': + # NOTE: the inverse of :meth:`__eq__`, with the ordered side's + # ``__eq__`` asked directly rather than its ``__ne__``: Werkzeug's + # OrderedMultiDict has none, and dict's compares the internal storage. + if isinstance(other, OrderedMultiDict): + result = other.__eq__(self) + else: + result = MultiDict.__eq__(self, other) + if result is NotImplemented: + return result + return not result + + def to_dict(self, flat: 'Optional[bool]' = None) -> 'Any': + """Return the contents as a plain mapping. + + Args: + flat: If omitted, a plain :class:`~pcapkit.corekit.multidict.MultiDict` + of every value, as :meth:`Info.to_dict` writes this object. + Otherwise :meth:`MultiDict.to_dict + `'s :obj:`dict`, of + the first value of each key if :obj:`True`, or of the + :obj:`list` of them if :obj:`False`. + + """ + if flat is None: + return MultiDict(self) + return MultiDict.to_dict(self, flat) # type: ignore[call-overload] + + def listvalues(self) -> 'Iterator[list[VT]]': + """Return an iterator of the :obj:`list` of all values of each key. + + Each :obj:`list` is a copy: :meth:`MultiDict.listvalues + ` hands out the lists it + keeps the values in, through which they could be changed. + + """ + return (values for (_, values) in self.lists()) + + def __reduce_ex__(self, protocol: 'Any') -> 'tuple[type, tuple[list[tuple[KT, VT]]]]': + return type(self), (list(self.items(multi=True)),) + + +class OrderedMultiInfo(_MultiInfo[KT, VT], OrderedMultiDict[KT, VT]): + """An immutable :class:`~pcapkit.corekit.multidict.OrderedMultiDict` that is an :class:`Info`. + + As :class:`MultiInfo`, but keeping the order of every value across keys, as + an :class:`~pcapkit.corekit.multidict.OrderedMultiDict` does -- i.e. wire + order for an option list. + + A finalised :class:`Info` holds every + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` value as one of these, + and :meth:`to_dict` -- its own and its holder's -- turns it back into an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict`. + + Args: + mapping: The initial value, as for + :class:`~pcapkit.corekit.multidict.OrderedMultiDict`. + + """ + + # NOTE: as for :class:`MultiInfo`, the mapping protocol is bound back to + # OrderedMultiDict's own. ``__ne__`` is not: OrderedMultiDict has none, and + # dict's compares the internal buckets, so two equal option lists compared + # ``!=`` as well as ``==``. It is :func:`_ne`, as for the exports + # (:issue:`1484`). + __getitem__ = OrderedMultiDict.__getitem__ # type: ignore[assignment] + __iter__ = OrderedMultiDict.__iter__ # type: ignore[assignment] + __len__ = OrderedMultiDict.__len__ + __contains__ = OrderedMultiDict.__contains__ + if hasattr(dict, '__reversed__'): # pragma: no branch -- Python 3.8 and up + __reversed__ = OrderedMultiDict.__reversed__ + __eq__ = OrderedMultiDict.__eq__ + __ne__ = _ne + __hash__ = None + __repr__ = OrderedMultiDict.__repr__ + __str__ = OrderedMultiDict.__str__ + __copy__ = OrderedMultiDict.__copy__ # type: ignore[assignment] + keys = OrderedMultiDict.keys + values = OrderedMultiDict.values + items = OrderedMultiDict.items # type: ignore[assignment] + get = OrderedMultiDict.get # type: ignore[assignment] + + def to_dict(self, flat: 'Optional[bool]' = None) -> 'Any': + """Return the contents as a plain mapping. + + Args: + flat: If omitted, a plain + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` of every + value, in order, as :meth:`Info.to_dict` writes this object -- + the same :class:`_OrderedMultiDict` it returns. + Otherwise :meth:`MultiDict.to_dict + `'s :obj:`dict`, of + the first value of each key if :obj:`True`, or of the + :obj:`list` of them if :obj:`False`. + + """ + if flat is None: + return _OrderedMultiDict(self) + return OrderedMultiDict.to_dict(self, flat) # type: ignore[call-overload] + + +def _freeze(value: 'Any') -> 'Any': + """Return ``value`` as stored in a finalised :class:`Info`. + + A :class:`~pcapkit.corekit.multidict.OrderedMultiDict` becomes an + :class:`OrderedMultiInfo`, and any other + :class:`~pcapkit.corekit.multidict.MultiDict` a :class:`MultiInfo`. Either + already immutable, and every other value, is returned as it is. + + """ + if isinstance(value, _MultiInfo) or not isinstance(value, MultiDict): + return value + if isinstance(value, OrderedMultiDict): + return OrderedMultiInfo(value) + return MultiInfo(value) + + +def _rebuild(info_cls: 'Type[Info]', value: 'Any') -> 'Any': + """Rebuild ``value`` into ``info_cls``, the class annotated for its key, unless it is one. + + A multi-mapping counts as a plain mapping here, even as the + :class:`OrderedMultiInfo` that :meth:`Info.__update__` stored an exported + :class:`Info` as. + + """ + if issubclass(info_cls, _MultiInfo): + if isinstance(value, MultiDict) and type(value) is not info_cls: # type: ignore[comparison-overlap,unreachable] # pylint: disable=unidiomatic-typecheck + return info_cls(value) + elif isinstance(value, Mapping) and (not isinstance(value, Info) or isinstance(value, _MultiInfo)): + return info_cls.from_dict(value) + return value + + #: Name of the per-class attribute :func:`_nested_info_types` caches its answer in. _NESTED_INFO_CACHE = '__nested_info_types__' +#: Name of the per-class attribute :func:`_nested_types` caches its answer in. +_NESTED_CACHE = '__nested_types__' + #: :pep:`604` union type, i.e. the origin of ``int | None``. It is absent below #: Python 3.10, where only :data:`typing.Optional` spells a union, and is the #: same object as :data:`typing.Union` from 3.14 on. @@ -690,13 +1045,17 @@ def _as_info_type(hint: 'Any') -> 'Optional[Type[Info]]': :data:`~typing.Optional` of a single :class:`Info` subclass counts as that subclass. A parametrised generic such as ``list[Packet]`` does not: the value it describes is a container, and :meth:`Info.from_dict` has nothing - to rebuild it from. + to rebuild it from. A :class:`MultiInfo` or :class:`OrderedMultiInfo` + subclass is an :class:`Info` subclass too, so that :meth:`Info.from_dict` + restores e.g. :class:`~pcapkit.protocols.data.application.httpv2.Settings` + from the :class:`~pcapkit.corekit.multidict.OrderedMultiDict` that + :meth:`Info.to_dict` writes it as. Args: hint: Resolved type annotation. Returns: - The :class:`Info` subclass the annotation names, or :data:`None`. + The class the annotation names, or :data:`None`. Note: Both guards below are load-bearing on Python 3.9 and 3.10, where a @@ -731,6 +1090,32 @@ def _as_info_type(hint: 'Any') -> 'Optional[Type[Info]]': def _nested_info_types(cls: 'Type[Info]') -> 'dict[str, Type[Info]]': """Map the keys of ``cls`` annotated with an :class:`Info` subclass to it. + It is :func:`_nested_types` less the multi-mapping classes + (:class:`_MultiInfo`), and cached alongside. + + Args: + cls: Info class. + + Returns: + Mapping of key names to the :class:`Info` subclass annotated for them. + + """ + cached = cls.__dict__.get(_NESTED_INFO_CACHE) + if cached is not None: + return cached + + nested = {key: value for (key, value) in _nested_types(cls).items() + if not issubclass(value, _MultiInfo)} # type: dict[str, Type[Info]] + setattr(cls, _NESTED_INFO_CACHE, nested) + return nested + + +def _nested_types(cls: 'Type[Info]') -> 'dict[str, Type[Info]]': + """Map the keys of ``cls`` annotated with a class :meth:`Info.from_dict` rebuilds to it. + + Such a class is an :class:`Info`, :class:`MultiInfo` or + :class:`OrderedMultiInfo` subclass, as :func:`_as_info_type` reads it. + Each annotation is resolved on its own, in the namespace of the module and the class that declared it, so one that cannot be resolved at runtime (e.g. a name imported only under :data:`~typing.TYPE_CHECKING`) is skipped rather @@ -740,7 +1125,7 @@ def _nested_info_types(cls: 'Type[Info]') -> 'dict[str, Type[Info]]': cls: Info class. Returns: - Mapping of key names to the :class:`Info` subclass annotated for them. + Mapping of key names to the class annotated for them. Note: The answer is cached in ``cls``'s own ``__dict__``, read back from @@ -761,7 +1146,7 @@ def _nested_info_types(cls: 'Type[Info]') -> 'dict[str, Type[Info]]': not arise in this package costs far more than it saves. """ - cached = cls.__dict__.get(_NESTED_INFO_CACHE) + cached = cls.__dict__.get(_NESTED_CACHE) if cached is not None: return cached @@ -791,5 +1176,5 @@ def _nested_info_types(cls: 'Type[Info]') -> 'dict[str, Type[Info]]': if info_cls is not None: nested[key] = info_cls - setattr(cls, _NESTED_INFO_CACHE, nested) + setattr(cls, _NESTED_CACHE, nested) return nested diff --git a/pcapkit/corekit/multidict.py b/pcapkit/corekit/multidict.py index 1ac586f7e1..464c58b97d 100644 --- a/pcapkit/corekit/multidict.py +++ b/pcapkit/corekit/multidict.py @@ -12,19 +12,17 @@ """ -import collections.abc import copy from typing import TYPE_CHECKING, Generic, TypeVar, cast, overload from pcapkit.utilities.exceptions import MissingKeyError, UnsupportedCall if TYPE_CHECKING: - from typing import (Any, ItemsView, Iterable, Iterator, KeysView, Mapping, NoReturn, Optional, - SupportsIndex, ValuesView) + from typing import Any, Iterable, Iterator, Mapping, NoReturn, Optional, SupportsIndex from typing_extensions import Literal -__all__ = ['MultiDict', 'OrderedMultiDict', 'InfoDict'] +__all__ = ['MultiDict', 'OrderedMultiDict'] ############################################################################### # Type variables @@ -638,73 +636,3 @@ def popitemlist(self) -> 'tuple[_KT, list[_VT]]': bucket.unlink(self) return key, [x.value for x in buckets] - - -class InfoDict(OrderedMultiDict[_KT, _VT]): - """The :class:`OrderedMultiDict` that :meth:`Info.to_dict - ` returns. - - It is the fields of an :class:`~pcapkit.corekit.infoclass.Info` as a mapping: - through the :obj:`dict` interface it holds one value per key, the first, so - :mod:`json`, :mod:`plistlib` and the dumpers see a plain mapping. A field the - :class:`~pcapkit.corekit.infoclass.Info` holds more than once, such as a - repeated IPv6 extension header, keeps every value, in order, for - ``items(multi=True)``. - - That sets it apart from an :class:`OrderedMultiDict` held as a field *value*, - e.g. a list of options, which the dumpers write as one entry per value. - - Note: - Unlike :class:`OrderedMultiDict`, :meth:`keys`, :meth:`values` and - :meth:`items` return views, and it compares like a :obj:`dict`: equal to - a mapping with the same keys -- in any order -- when each key holds the - one value the mapping has, or the same values as another - :class:`MultiDict`. - - """ - - def __eq__(self, other: 'Any') -> 'bool': - if isinstance(other, MultiDict): - if len(self) != len(other): - return False - return all(other.getlist(key) == values for key, values in self.lists()) - if isinstance(other, collections.abc.Mapping): - if len(self) != len(other): - return False - for key, values in self.lists(): - if key not in other or values != [other[key]]: - return False - return True - return NotImplemented - - def __ne__(self, other: 'Any') -> 'bool': - # NOTE: :obj:`dict` defines its own ``__ne__``, which compares the - # internal buckets rather than negating :meth:`__eq__`. - result = self.__eq__(other) - if result is NotImplemented: - return result - return not result - - def __iter__(self) -> 'Iterator[_KT]': - return (key for key, _ in OrderedMultiDict.items(self)) - - def keys(self) -> 'KeysView[_KT]': # type: ignore[override] - """Return a view of the keys, each once, as :meth:`dict.keys` does.""" - return collections.abc.KeysView(self) - - def values(self) -> 'ValuesView[_VT]': # type: ignore[override] - """Return a view of the first value of each key, as :meth:`dict.values` does.""" - return collections.abc.ValuesView(self) - - def items(self, multi: 'bool' = False) -> 'ItemsView[_KT, _VT] | Iterator[tuple[_KT, _VT]]': # type: ignore[override] # pylint: disable=line-too-long - """Return the ``(key, value)`` pairs. - - Args: - multi: If set to :obj:`True`, an iterator with a pair for each value - of each key, in insertion order. Otherwise a view of the pairs - of the first value of each key, as :meth:`dict.items` returns. - - """ - if multi: - return OrderedMultiDict.items(self, multi=True) - return collections.abc.ItemsView(self) diff --git a/pcapkit/dumpkit/common.py b/pcapkit/dumpkit/common.py index 4833c4fb1d..8e8f93decf 100644 --- a/pcapkit/dumpkit/common.py +++ b/pcapkit/dumpkit/common.py @@ -30,7 +30,7 @@ import dictdumper.tree from pcapkit.corekit.infoclass import Info -from pcapkit.corekit.multidict import InfoDict, MultiDict, OrderedMultiDict +from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict from pcapkit.corekit.sentinels import NoValueType, NullType from pcapkit.protocols.schema.schema import Schema from pcapkit.utilities.exceptions import UnsupportedCall @@ -393,9 +393,72 @@ def escape_key(key: 'Any') -> 'Any': return json.dumps(format(key, ''))[1:-1] return key + def fields(info: 'Info | MultiDict[Any, Any]') -> 'dict[Any, Any]': + """Return the fields of ``info`` as the :class:`dict` the writer is handed. + + Args: + info: The fields, as an :class:`~pcapkit.corekit.infoclass.Info` or + as the :class:`~pcapkit.corekit.multidict.OrderedMultiDict` its + :meth:`~pcapkit.corekit.infoclass.Info.to_dict` returns. + + Returns: + One entry per field, in order, with its key escaped as + :func:`escape_key` does. A field held once maps to its value; a field + held more than once, such as a repeated IPv6 extension header, maps + to the :class:`list` of all its values, in order (GitHub issue + :issue:`1484`). The values are left as they are, so each comes back + through :meth:`DictDumper.object_hook` when written. + + """ + pairs = list(info.items(multi=True)) + counts = collections.Counter(key for key, _ in pairs) + result = {} # type: dict[Any, Any] + for key, val in pairs: + if counts[key] > 1: + result.setdefault(escape_key(key), []).append(val) + else: + # NOTE: read by indexing, so that a model overriding it is + # honoured -- the traceflow and reassembly models resolve a + # deferred ``packet`` there. + result[escape_key(key)] = info[key] + return result + + #: Whether the writer is handed the fields of an + #: :class:`~pcapkit.corekit.infoclass.Info` rather than the object itself: + #: every writer but the ones :mod:`pcapkit.dumpkit` defines for internal + #: use, i.e. :class:`~pcapkit.dumpkit.pcap.PCAPIO`, which takes the frame, + #: and :class:`~pcapkit.dumpkit.null.NotImplementedIO`. + takes_fields = not issubclass(output, DumperBase) or issubclass(output, Dumper) + class DictDumper(output): """Customised :class:`~dictdumper.dumper.Dumper` object.""" + if takes_fields: + def __call__(self, value: 'Any', name: 'Optional[str]' = None) -> 'Any': + """Dump a new block. + + Args: + self: Dumper instance. + value: Content to be dumped. + name: Name of the block. + + Returns: + The dumper itself. + + Notes: + An :class:`~pcapkit.corekit.infoclass.Info`, or the + :class:`~pcapkit.corekit.multidict.MultiDict` its + :meth:`~pcapkit.corekit.infoclass.Info.to_dict` returns, is + written as the :class:`dict` :func:`fields` makes of it, so + that a repeated field keeps every value. The writer walks + the block it is handed without calling + :meth:`object_hook` on it, only on what it holds. + + """ + if isinstance(value, (Info, MultiDict)): + value = fields(value) + return super().__call__(value, name) + #: Objects :meth:`_append_fallback` is expanding, by :func:`id` -- the #: one being written and every one enclosing it. _fallback_path = None # type: Optional[set[int]] @@ -427,17 +490,23 @@ def object_hook(self, o: 'Any') -> 'Any': :func:`escape_key` instead. An :class:`~pcapkit.corekit.multidict.OrderedMultiDict` becomes a :class:`list` of single-key :class:`dict` objects, each of - which comes back through the :class:`dict` branch. An - :class:`~pcapkit.corekit.multidict.InfoDict`, the fields of an - :class:`~pcapkit.corekit.infoclass.Info`, takes that branch - directly, so it is written with one value per key. + which comes back through the :class:`dict` branch, and so + does an :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo`, + the option lists an :class:`~pcapkit.corekit.infoclass.Info` + holds. An :class:`~pcapkit.corekit.infoclass.Info` itself + becomes the :class:`dict` of its fields that :func:`fields` + returns, holding every value of a repeated field. """ if isinstance(o, decimal.Decimal): result = str(o) # type: Any elif isinstance(o, datetime.timedelta): result = o.total_seconds() - elif isinstance(o, (Info, Schema)): + elif isinstance(o, Info) and not isinstance(o, MultiDict): + # NOTE: a MultiInfo or OrderedMultiInfo, though an Info, is + # an option list, and is written by the branches below. + result = fields(o) + elif isinstance(o, Schema): result = o.to_dict() elif isinstance(o, (ipaddress.IPv4Address, ipaddress.IPv6Address, ipaddress.IPv4Network, ipaddress.IPv6Network)): @@ -448,7 +517,7 @@ def object_hook(self, o: 'Any') -> 'Any': result = '' elif isinstance(o, NoValueType): result = '' - elif isinstance(o, OrderedMultiDict) and not isinstance(o, InfoDict): + elif isinstance(o, OrderedMultiDict): # NOTE: one single-key mapping per entry, in insertion order -- # i.e. wire order for the option lists -- since a mapping keyed # by name could only gather the repeats of a key under its first @@ -459,7 +528,7 @@ def object_hook(self, o: 'Any') -> 'Any': {render_enum(key) if isinstance(key, (enum.Enum, aenum.Enum)) else key: val} for key, val in o.items(multi=True) ] - elif isinstance(o, MultiDict) and not isinstance(o, InfoDict): + elif isinstance(o, MultiDict): # NOTE: a :class:`MultiDict` keeps no order across keys, only # among the values of each, so grouping by key loses nothing. temp = collections.defaultdict(list) # type: DefaultDict[str, list[Any]] diff --git a/pcapkit/foundation/engines/pcap.py b/pcapkit/foundation/engines/pcap.py index 9ee5974d6c..4ef29d9e81 100644 --- a/pcapkit/foundation/engines/pcap.py +++ b/pcapkit/foundation/engines/pcap.py @@ -133,7 +133,7 @@ def run(self) -> 'None': ofile = ext._ofile # NOTE: a PCAP writer has already written the global header on creation. if ofile.kind != 'pcap': - ofile(self._gbhdr.info.to_dict(), name='Global Header') + ofile(self._gbhdr.info, name='Global Header') ext._offmt = ofile.kind def read_frame(self) -> 'Frame': @@ -168,9 +168,11 @@ def read_frame(self) -> 'Frame': # write plist frnum = f'Frame {ext._frnum}' if not ext._flag_q: - # NOTE: a PCAP writer takes the frame itself, for its octets and - # record header; the other writers take its fields. - info = frame.info if ext._offmt == 'pcap' else frame.info.to_dict() + # NOTE: every writer takes the frame's info: a PCAP writer for its + # octets and record header, the others for its fields, which only + # the info itself tells apart from the option lists it holds + # (GitHub issue #1484). + info = frame.info if ext._flag_f: ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}') ofile(info, name=frnum) diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index 8c9b7b68c9..06f498ab39 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -350,9 +350,9 @@ def _write_file(self, block: 'Data_PCAPNG', *, name: 'str') -> 'None': if ext._flag_f: ofile = ext._ofile(f'{ext._ofnm}/{name}.{ext._fext}') - ofile(block.to_dict(), name=name) + ofile(block, name=name) else: - ext._ofile(block.to_dict(), name=name) + ext._ofile(block, name=name) ofile = ext._ofile ext._offmt = ofile.kind diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index 3131ed46d5..04335ea06c 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -807,12 +807,10 @@ def _read_http_settings(self, schema: 'Schema_SettingsFrame', *, if flag.ACK and header.length != 0: raise ProtocolError(f'HTTP/2: [Type {header.type}] invalid format') - sets = Data_Settings() - # ``add`` rather than item assignment: an identifier may repeat, and + # NOTE: every pair rather than a mapping: an identifier may repeat, and # the values apply in order (:rfc:`9113#section-6.5.3`). ``Settings`` # keeps every entry, and answers an index with the last value. - for setting in schema.settings: - sets.add(setting.id, setting.value) + sets = Data_Settings([(setting.id, setting.value) for setting in schema.settings]) data = Data_SettingsFrame( length=header.length, diff --git a/pcapkit/protocols/data/application/httpv1.py b/pcapkit/protocols/data/application/httpv1.py index 70d6125c91..298fe38ffc 100644 --- a/pcapkit/protocols/data/application/httpv1.py +++ b/pcapkit/protocols/data/application/httpv1.py @@ -12,6 +12,7 @@ from pcapkit.const.http.method import Method as Enum_Method from pcapkit.const.http.status_code import StatusCode as Enum_StatusCode + from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.application.httpv1 import Type as HTTP_Type @@ -30,7 +31,7 @@ class HTTP(Protocol): #: HTTP receipt. receipt: 'Header' #: HTTP header. - header: 'OrderedMultiDict[str, str]' + header: 'OrderedMultiInfo[str, str]' #: HTTP body. body: 'Any' #: HTTP field lines as received, CRLF excluded and ``obs-fold`` diff --git a/pcapkit/protocols/data/application/httpv2.py b/pcapkit/protocols/data/application/httpv2.py index 5295d0d95c..ac050cf9ad 100644 --- a/pcapkit/protocols/data/application/httpv2.py +++ b/pcapkit/protocols/data/application/httpv2.py @@ -3,8 +3,7 @@ from typing import TYPE_CHECKING -from pcapkit.corekit.infoclass import info_final -from pcapkit.corekit.multidict import OrderedMultiDict +from pcapkit.corekit.infoclass import OrderedMultiInfo, info_final from pcapkit.protocols.data.data import Data from pcapkit.protocols.data.protocol import Protocol @@ -166,27 +165,32 @@ class RSTStreamFrame(HTTP): def __init__(self, length: 'int', type: 'Frame', flags: 'Flags', reserved: 'int', sid: 'int', error: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin -class Settings(OrderedMultiDict['Setting', int]): +class Settings(OrderedMultiInfo['Setting', int]): """Settings of an HTTP/2 ``SETTINGS`` frame. An identifier may repeat, and the values apply in order, so the last one wins (:rfc:`9113#section-6.5`). Every entry is kept in wire order, as :meth:`getlist` and ``items(multi=True)`` show and the rebuild writes, but - indexing, :meth:`get`, :meth:`values`, ``items()`` and :meth:`to_dict` + indexing, :meth:`get`, :meth:`values`, ``items()`` and ``to_dict(flat=True)`` answer with the *last* value of each identifier rather than the first. + It is immutable, as an + :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` is, so it is built + from the ``(identifier, value)`` pairs in one go, and ``to_dict()`` exports + every pair as a plain :class:`~pcapkit.corekit.multidict.OrderedMultiDict`. + """ - def __getitem__(self, key: 'Setting') -> 'int': + def __getitem__(self, key: 'Setting') -> 'int': # type: ignore[override] if key in self: return self.getlist(key)[-1] - return super().__getitem__(key) + return super().__getitem__(key) # pylint: disable=no-value-for-parameter def items(self, multi: 'bool' = False) -> 'Iterator[tuple[Setting, int]]': # type: ignore[override] if multi: - yield from super().items(multi=True) + yield from super().items(multi=True) # pylint: disable=no-value-for-parameter return - for key, _ in super().items(): + for key, _ in super().items(): # pylint: disable=no-value-for-parameter yield key, self[key] diff --git a/pcapkit/protocols/data/internet/hip.py b/pcapkit/protocols/data/internet/hip.py index a77a60e556..d6c1227c3b 100644 --- a/pcapkit/protocols/data/internet/hip.py +++ b/pcapkit/protocols/data/internet/hip.py @@ -31,7 +31,7 @@ from pcapkit.const.hip.suite import Suite from pcapkit.const.hip.transport import Transport from pcapkit.const.reg.transtype import TransType - from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.corekit.infoclass import OrderedMultiInfo __all__ = [ 'HIP', 'Control', @@ -124,7 +124,7 @@ class HIP(Protocol): if TYPE_CHECKING: #: HIP parameters. - parameters: 'OrderedMultiDict[Enum_Parameter, Parameter]' + parameters: 'OrderedMultiInfo[Enum_Parameter, Parameter]' def __init__(self, next: 'TransType', length: 'int', type: 'Packet', packet_fixed: 'int', version: 'int', reserved: 'int', version_fixed: 'int', chksum: 'bytes', control: 'Control', shit: 'int', rhit: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long diff --git a/pcapkit/protocols/data/internet/hopopt.py b/pcapkit/protocols/data/internet/hopopt.py index 8ad1fcaf14..4cf1e529f1 100644 --- a/pcapkit/protocols/data/internet/hopopt.py +++ b/pcapkit/protocols/data/internet/hopopt.py @@ -20,6 +20,7 @@ from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode from pcapkit.const.ipv6.tagger_id import TaggerID from pcapkit.const.reg.transtype import TransType + from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict __all__ = [ @@ -59,7 +60,7 @@ class HOPOPT(Protocol): #: Header extension length. length: 'int' #: HOPOPT options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long diff --git a/pcapkit/protocols/data/internet/ipv4.py b/pcapkit/protocols/data/internet/ipv4.py index f05b5d4524..2874afac34 100644 --- a/pcapkit/protocols/data/internet/ipv4.py +++ b/pcapkit/protocols/data/internet/ipv4.py @@ -27,6 +27,7 @@ from pcapkit.const.ipv4.tos_thr import ToSThroughput from pcapkit.const.ipv4.ts_flag import TSFlag from pcapkit.const.reg.transtype import TransType + from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict __all__ = [ @@ -121,7 +122,7 @@ class IPv4(Protocol): if TYPE_CHECKING: #: Options, present only when the header carries any. - options: 'OrderedMultiDict[OptionNumber, Option]' + options: 'OrderedMultiInfo[OptionNumber, Option]' #: Octets after the End of Option List, present along with ``options``. padding: 'bytes' #: Octets captured past the Total Length, present only when there are any. @@ -231,7 +232,7 @@ class TSOption(Option): #: Flag. flag: 'TSFlag' #: Timestamp data, i.e. the slots before the pointer. - timestamp: 'tuple[timedelta | int, ...] | OrderedMultiDict[IPv4Address, timedelta | int]' + timestamp: 'tuple[timedelta | int, ...] | OrderedMultiInfo[IPv4Address, timedelta | int]' #: Slots at or beyond the pointer, as raw 32-bit words. remaining: 'tuple[int, ...]' #: Trailing octets that do not make up a whole 32-bit slot. diff --git a/pcapkit/protocols/data/internet/ipv6_opts.py b/pcapkit/protocols/data/internet/ipv6_opts.py index c064f5fb8c..126a9ee5ea 100644 --- a/pcapkit/protocols/data/internet/ipv6_opts.py +++ b/pcapkit/protocols/data/internet/ipv6_opts.py @@ -20,6 +20,7 @@ from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode from pcapkit.const.ipv6.tagger_id import TaggerID from pcapkit.const.reg.transtype import TransType + from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict __all__ = [ @@ -59,7 +60,7 @@ class IPv6_Opts(Protocol): #: Header extension length. length: 'int' #: IPv6-Opts options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index 64b83b387a..70ca468086 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -64,6 +64,7 @@ from pcapkit.const.mh.upn_reason import \ UpdateNotificationReason as Enum_UpdateNotificationReason from pcapkit.const.reg.transtype import TransType + from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.internet.mh import (FastBindingAcknowledgmentStatus, IPv6AddressPrefixCode, LMAAddressCode, @@ -170,7 +171,7 @@ class BindingRefreshRequestMessage(MH): #: apart from :attr:`MH.reserved`, the reserved octet of the fixed header. msg_reserved: 'int' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -187,7 +188,7 @@ class HomeTestInitMessage(MH): #: Home init cookie. cookie: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -205,7 +206,7 @@ class CareofTestInitMessage(MH): #: Care-of init cookie. cookie: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -224,7 +225,7 @@ class HomeTestMessage(MH): #: Home keygen token. token: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -243,7 +244,7 @@ class CareofTestMessage(MH): #: Care-of keygen token. token: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -286,7 +287,7 @@ class BindingUpdateMessage(MH): #: Lifetime. lifetime: 'timedelta' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -324,7 +325,7 @@ class BindingAcknowledgementMessage(MH): #: Lifetime. lifetime: 'timedelta' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -346,7 +347,7 @@ class BindingErrorMessage(MH): #: Home address. home: 'IPv6Address' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -374,7 +375,7 @@ class FastBindingUpdateMessage(MH): #: the units of 4 seconds the binding messages use. lifetime: 'timedelta' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -403,7 +404,7 @@ class FastBindingAcknowledgmentMessage(MH): #: the units of 4 seconds the binding messages use. lifetime: 'timedelta' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -421,7 +422,7 @@ class FastNeighborAdvertisementMessage(MH): #: from :attr:`MH.reserved`, the reserved octet of the fixed header. msg_reserved: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -459,7 +460,7 @@ class HandoverInitiateMessage(MH): #: Code. code: 'Enum_HandoverInitiateStatus' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -486,7 +487,7 @@ class HandoverAcknowledgeMessage(MH): #: Code. code: 'Enum_HandoverACKStatus' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -509,7 +510,7 @@ class HeartbeatMessage(MH): #: Sequence number. seq: 'int' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -528,7 +529,7 @@ class HomeAgentSwitchMessage(MH): #: run home agent discovery instead. addresses: 'tuple[IPv6Address, ...]' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -563,7 +564,7 @@ class BindingRevocationMessage(MH): #: Reserved bits, i.e. the 13 bits after the ``G`` flag. flags_reserved: 'int' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -586,7 +587,7 @@ class LocalizedRoutingInitiationMessage(MH): #: the units of 4 seconds the binding messages use. lifetime: 'timedelta' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -612,7 +613,7 @@ class LocalizedRoutingAcknowledgmentMessage(MH): #: Lifetime, in seconds. lifetime: 'timedelta' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -636,7 +637,7 @@ class UpdateNotificationMessage(MH): #: Reserved bits, i.e. the 14 bits after the ``D`` flag. flags_reserved: 'int' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -657,7 +658,7 @@ class UpdateNotificationAcknowledgementMessage(MH): #: from :attr:`MH.reserved`, the reserved octet of the fixed header. msg_reserved: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -687,7 +688,7 @@ class FlowBindingMessage(MH): #: Reserved bits, i.e. the seven bits after the ``A`` flag. flags_reserved: 'int' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -707,7 +708,7 @@ class SubscriptionQueryMessage(MH): #: from :attr:`MH.reserved`, the reserved octet of the fixed header. msg_reserved: 'bytes' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -727,7 +728,7 @@ class SubscriptionResponseMessage(MH): #: Reserved bits, i.e. the seven bits after the ``I`` flag. flags_reserved: 'int' #: Mobility options. - options: 'OrderedMultiDict[Enum_Option, Option]' + options: 'OrderedMultiInfo[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', @@ -958,7 +959,7 @@ class CGAParameter(Data): #: Public key (ASN.1 encoded). public_key: 'bytes' #: Extension fields. - extensions: 'OrderedMultiDict[Enum_CGAExtension, CGAExtension]' + extensions: 'OrderedMultiInfo[Enum_CGAExtension, CGAExtension]' if TYPE_CHECKING: def __init__(self, modifier: 'Enum_CGAType', prefix: 'int', collision_count: 'int', public_key: 'bytes', @@ -1574,7 +1575,7 @@ class FlowIdentificationOption(Option): #: Status. status: 'Enum_FlowIDStatus' #: Sub-options. - suboptions: 'OrderedMultiDict[Enum_FlowIDSuboption, FlowIdentificationSuboption]' + suboptions: 'OrderedMultiInfo[Enum_FlowIDSuboption, FlowIdentificationSuboption]' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', fid: 'int', fid_pri: 'int', @@ -1792,7 +1793,7 @@ class AccessNetworkIdentifierOption(Option): """Data model for MH Access Network Identifier option.""" #: Sub-options. - suboptions: 'OrderedMultiDict[Enum_ANISuboption, ANISuboption]' + suboptions: 'OrderedMultiInfo[Enum_ANISuboption, ANISuboption]' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', @@ -1809,7 +1810,7 @@ class IPv4TrafficOffloadSelectorOption(Option): #: Reserved bits, i.e. the 31 bits after the ``M`` flag. reserved: 'int' #: Traffic selector sub-options. - selector: 'OrderedMultiDict[Enum_FlowIDSuboption, FlowIdentificationSuboption]' + selector: 'OrderedMultiInfo[Enum_FlowIDSuboption, FlowIdentificationSuboption]' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', mode: 'bool', reserved: 'int', @@ -2006,7 +2007,7 @@ class QualityOfServiceOption(Option): #: The three reserved octets, as captured. reserved: 'bytes' #: Quality-of-service attributes. - attributes: 'OrderedMultiDict[Enum_QoSAttribute, QoSAttribute]' + attributes: 'OrderedMultiInfo[Enum_QoSAttribute, QoSAttribute]' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', sr_id: 'int', dscp: 'int', @@ -2125,7 +2126,7 @@ class LMAControlledMAGParametersOption(Option): """Data model for MH LMA-Controlled MAG Parameters option.""" #: Sub-options. - suboptions: 'OrderedMultiDict[Enum_LMAControlledMAGSuboption, LMAControlledMAGSuboption]' + suboptions: 'OrderedMultiInfo[Enum_LMAControlledMAGSuboption, LMAControlledMAGSuboption]' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', diff --git a/pcapkit/protocols/data/misc/pcapng.py b/pcapkit/protocols/data/misc/pcapng.py index 5c6e4f367e..5c5b588bd7 100644 --- a/pcapkit/protocols/data/misc/pcapng.py +++ b/pcapkit/protocols/data/misc/pcapng.py @@ -55,6 +55,7 @@ from pcapkit.const.pcapng.secrets_type import SecretsType as Enum_SecretsType from pcapkit.const.pcapng.verdict_type import VerdictType as Enum_VerdictType from pcapkit.const.reg.linktype import LinkType as Enum_LinkType + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.corekit.version import VersionInfo from pcapkit.protocols.misc.pcapng import (PacketDirection, PacketReception, TLSKeyLabel, @@ -155,7 +156,7 @@ class SectionHeaderBlock(PCAPNG): #: Section length. section_length: 'int' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' if TYPE_CHECKING: #: Octets after ``opt_endofopt``, as captured; present only when the @@ -355,7 +356,7 @@ class InterfaceDescriptionBlock(PCAPNG): #: Snap length. snaplen: 'int' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' if TYPE_CHECKING: #: Reserved octets, as captured; present only when they are not all @@ -494,7 +495,7 @@ class EnhancedPacketBlock(PCAPNG): #: Original packet length. original_len: 'int' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' if TYPE_CHECKING: #: Padding of the packet data to 32 bits, as captured; present only when @@ -652,9 +653,9 @@ class NameResolutionBlock(PCAPNG): """Data model for PCAP-NG Name Resolution Block (NRB).""" #: Records. - records: 'OrderedMultiDict[Enum_RecordType, NameResolutionRecord]' + records: 'OrderedMultiInfo[Enum_RecordType, NameResolutionRecord]' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' def __post_init__(self) -> 'None': """Post-initialization handling.""" @@ -677,9 +678,9 @@ def __post_init__(self) -> 'None': #: option area holds any (:issue:`1437`). padding_opts: 'bytes' #: Name resolution mapping (IP address -> name). - mapping: 'MultiDict[IPv4Address | IPv6Address, str]' + mapping: 'MultiInfo[IPv4Address | IPv6Address, str]' #: Name resolution mapping (name -> IP address). - reverse_mapping: 'MultiDict[str, IPv4Address | IPv6Address]' + reverse_mapping: 'MultiInfo[str, IPv4Address | IPv6Address]' def __init__(self, type: 'Enum_BlockType', length: 'int', records: 'OrderedMultiDict[Enum_RecordType, NameResolutionRecord]', @@ -780,7 +781,7 @@ class InterfaceStatisticsBlock(PCAPNG): #: Timestamp as in UNIX epoch (in seconds). timestamp_epoch: 'Decimal' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' if TYPE_CHECKING: #: Octets after ``opt_endofopt``, as captured; present only when the @@ -796,7 +797,7 @@ class SystemdJournalExportBlock(PCAPNG): """Data model for PCAP-NG :manpage:`systemd(1)` Journal Export Block.""" #: Journal entry. - data: 'tuple[OrderedMultiDict[str, str | bytes], ...]' + data: 'tuple[OrderedMultiInfo[str, str | bytes], ...]' if TYPE_CHECKING: def __init__(self, type: 'Enum_BlockType', length: 'int', @@ -823,7 +824,7 @@ class TLSKeyLog(DSBSecrets): """Data model for TLS key log DSB secrets.""" #: TLS key log entries. - entries: 'dict[TLSKeyLabel, OrderedMultiDict[bytes, bytes]]' + entries: 'dict[TLSKeyLabel, OrderedMultiInfo[bytes, bytes]]' if TYPE_CHECKING: #: TLS key log text as the block carries it, comments and line order @@ -838,7 +839,7 @@ class WireGuardKeyLog(DSBSecrets): """Data model for WireGuard key DSB secrets.""" #: WireGuard Key Log entries. - entries: 'OrderedMultiDict[WireGuardKeyLabel, bytes]' + entries: 'OrderedMultiInfo[WireGuardKeyLabel, bytes]' if TYPE_CHECKING: #: WireGuard key log text as the block carries it, comments included. @@ -887,7 +888,7 @@ class DecryptionSecretsBlock(PCAPNG): #: Secrets data. secrets_data: 'DSBSecrets' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' if TYPE_CHECKING: #: Padding of the secrets data to 32 bits, as captured; present only when @@ -995,7 +996,7 @@ class PacketBlock(PCAPNG): #: Original packet length. original_len: 'int' #: Options. - options: 'OrderedMultiDict[Enum_OptionType, Option]' + options: 'OrderedMultiInfo[Enum_OptionType, Option]' if TYPE_CHECKING: #: Padding of the packet data to 32 bits, as captured; present only when diff --git a/pcapkit/protocols/data/transport/sctp.py b/pcapkit/protocols/data/transport/sctp.py index aeacc4aaa6..34a758b361 100644 --- a/pcapkit/protocols/data/transport/sctp.py +++ b/pcapkit/protocols/data/transport/sctp.py @@ -16,6 +16,7 @@ from pcapkit.const.sctp.chunk import Chunk as ChunkType from pcapkit.const.sctp.parameter import Parameter as ParameterType from pcapkit.const.sctp.payload_protocol_identifier import PayloadProtocolIdentifier + from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.multidict import OrderedMultiDict IPAddress = Union[IPv4Address, IPv6Address] @@ -447,7 +448,7 @@ class INITChunk(Chunk): #: Initial transmission sequence number. init_tsn: 'int' #: Optional and variable-length parameters. - parameters: 'OrderedMultiDict[ParameterType, Parameter]' + parameters: 'OrderedMultiInfo[ParameterType, Parameter]' if TYPE_CHECKING: def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', init_tag: 'int', a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int', init_tsn: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -470,7 +471,7 @@ class INITACKChunk(Chunk): #: Initial transmission sequence number. init_tsn: 'int' #: Optional and variable-length parameters. - parameters: 'OrderedMultiDict[ParameterType, Parameter]' + parameters: 'OrderedMultiInfo[ParameterType, Parameter]' if TYPE_CHECKING: def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', init_tag: 'int', a_rwnd: 'int', outbound_streams: 'int', inbound_streams: 'int', init_tsn: 'int', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -506,7 +507,7 @@ class HeartbeatChunk(Chunk): #: Raw chunk flags, kept as received. flags: 'bytes' #: Heartbeat information parameters. - parameters: 'OrderedMultiDict[ParameterType, Parameter]' + parameters: 'OrderedMultiInfo[ParameterType, Parameter]' if TYPE_CHECKING: def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -519,7 +520,7 @@ class HeartbeatACKChunk(Chunk): #: Raw chunk flags, kept as received. flags: 'bytes' #: Heartbeat information parameters. - parameters: 'OrderedMultiDict[ParameterType, Parameter]' + parameters: 'OrderedMultiInfo[ParameterType, Parameter]' if TYPE_CHECKING: def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', parameters: 'OrderedMultiDict[ParameterType, Parameter]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -532,7 +533,7 @@ class AbortChunk(Chunk): #: Chunk flags. flags: 'TBitFlags' #: Zero or more error causes. - error: 'OrderedMultiDict[CauseCode, ErrorCause]' + error: 'OrderedMultiInfo[CauseCode, ErrorCause]' if TYPE_CHECKING: def __init__(self, type: 'ChunkType', length: 'int', flags: 'TBitFlags', error: 'OrderedMultiDict[CauseCode, ErrorCause]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -569,7 +570,7 @@ class ErrorChunk(Chunk): #: Raw chunk flags, kept as received. flags: 'bytes' #: One or more error causes. - error: 'OrderedMultiDict[CauseCode, ErrorCause]' + error: 'OrderedMultiInfo[CauseCode, ErrorCause]' if TYPE_CHECKING: def __init__(self, type: 'ChunkType', length: 'int', flags: 'bytes', error: 'OrderedMultiDict[CauseCode, ErrorCause]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -623,7 +624,7 @@ class SCTP(Protocol): #: Checksum, as a CRC32c over the whole packet with this field zeroed. chksum: 'bytes' #: Chunks. - chunks: 'OrderedMultiDict[ChunkType, Chunk]' + chunks: 'OrderedMultiInfo[ChunkType, Chunk]' if TYPE_CHECKING: #: Octets captured after the last whole chunk, present only when there diff --git a/pcapkit/protocols/data/transport/tcp.py b/pcapkit/protocols/data/transport/tcp.py index 3d20b99911..3f5d043b59 100644 --- a/pcapkit/protocols/data/transport/tcp.py +++ b/pcapkit/protocols/data/transport/tcp.py @@ -17,7 +17,7 @@ from pcapkit.const.tcp.flags import Flags as TCP_Flags from pcapkit.const.tcp.mp_tcp_option import MPTCPOption from pcapkit.const.tcp.option import Option as OptionNumber - from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.corekit.infoclass import OrderedMultiInfo IPAddress = Union[IPv4Address, IPv6Address] @@ -99,7 +99,7 @@ class TCP(Protocol): if TYPE_CHECKING: #: TCP options. - options: 'OrderedMultiDict[OptionNumber, Option]' + options: 'OrderedMultiInfo[OptionNumber, Option]' #: Octets of the option area after the End of Option List option. padding: 'bytes' #: Connection control flags. diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index d071f5d6d9..50ee7bc632 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -40,6 +40,7 @@ from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.corekit.enum import EnumLookup from pcapkit.corekit.fields.strings import DecodedString +from pcapkit.corekit.infoclass import OrderedMultiInfo from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.corekit.version import VersionInfo @@ -2498,7 +2499,9 @@ def _read_block_systemd(self, schema: 'Schema_SystemdJournalExportBlock', *, data = Data_SystemdJournalExportBlock( type=header.type, length=schema.length, - data=() if kept else tuple(entries), # type: ignore[arg-type] + # NOTE: each entry as an OrderedMultiInfo, which ``Info`` does not + # do for a mapping held in a tuple (:issue:`1484`). + data=() if kept else tuple(OrderedMultiInfo(entry) for entry in entries), # type: ignore[arg-type] ) if kept: data.__update__([('entry_raw', entries[0])]) @@ -4110,7 +4113,9 @@ def _read_secrets_tls(self, schema: 'Schema_TLSKeyLog', *, """ secrets = Data_TLSKeyLog( - entries=schema.entries, + # NOTE: each label's entries as an OrderedMultiInfo, which ``Info`` + # does not do for a mapping held in a dict (:issue:`1484`). + entries={label: OrderedMultiInfo(entry) for (label, entry) in schema.entries.items()}, ) # NOTE: the log text is kept as well, since :attr:`entries` holds # neither its comments nor the order of its lines, and a rebuild writes @@ -6520,7 +6525,8 @@ def _make_secrets_tls(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_TL return Schema_TLSKeyLog( data=text, ) - entries = secrets.entries + # NOTE: each an OrderedMultiInfo, which is an OrderedMultiDict. + entries = cast('dict[Enum_TLSKeyLabel, OrderedMultiDict[bytes, bytes]]', secrets.entries) if entries is None: entries = {} diff --git a/tests/corekit/test_from_dict_roundtrip_unit.py b/tests/corekit/test_from_dict_roundtrip_unit.py index 4abbc04dc1..58b26bbf28 100644 --- a/tests/corekit/test_from_dict_roundtrip_unit.py +++ b/tests/corekit/test_from_dict_roundtrip_unit.py @@ -120,7 +120,9 @@ class Out(Info): self.assertIsInstance(rebuilt.inner, In) self.assertIsInstance(rebuilt.maybe, In) self.assertEqual((rebuilt.inner.x, rebuilt.maybe.x, rebuilt.y), (1, 2, 3)) - self.assertEqual(rebuilt.to_dict(), info.to_dict()) + # NOTE: the pairs rather than the two exports: an OrderedMultiDict compares + # its values with ``!=``, which for a nested one compares its buckets. + self.assertEqual(list(rebuilt.to_dict().items(multi=True)), list(info.to_dict().items(multi=True))) def test_string_annotation_in_a_data_module_is_resolved(self) -> None: from pcapkit.protocols.data.application.httpv2 import DataFrame, DataFrameFlags diff --git a/tests/corekit/test_info_roundtrip_unit.py b/tests/corekit/test_info_roundtrip_unit.py index c8d2a433ff..6399f25efe 100644 --- a/tests/corekit/test_info_roundtrip_unit.py +++ b/tests/corekit/test_info_roundtrip_unit.py @@ -18,10 +18,15 @@ :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. +``to_dict()`` must also be a plain +:class:`~pcapkit.corekit.multidict.OrderedMultiDict` at every level, and the +instance must hold no mutable multi-mapping: every +:class:`~pcapkit.corekit.multidict.MultiDict` value is a +:class:`~pcapkit.corekit.infoclass.MultiInfo` or +:class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` (#1484). +: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 @@ -132,24 +137,56 @@ def _structure(info: 'Any') -> 'Any': def _plain_dicts(value: 'Any', path: 'str' = '') -> 'list[str]': - """Paths in a ``to_dict()`` result of every mapping that should be an InfoDict. + """Paths in a ``to_dict()`` result of every mapping that is not a plain multi-mapping. - 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. + A nested :class:`Info` exports as an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict`, and so does an + :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo`; a + :class:`~pcapkit.corekit.infoclass.MultiInfo` exports as a + :class:`~pcapkit.corekit.multidict.MultiDict`. Neither ``*Info`` type, nor a + plain :obj:`dict`, may be left in the export. """ - from pcapkit.corekit.multidict import InfoDict, MultiDict + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo, _OrderedMultiDict + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict - if not isinstance(value, InfoDict): + if type(value) is not _OrderedMultiDict: 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)): + if isinstance(item, (MultiInfo, OrderedMultiInfo)) or (isinstance(item, dict) and not isinstance(item, MultiDict)): + found.append(f'{path}.{key}: {type(item).__module__}.{type(item).__qualname__}') + elif isinstance(item, OrderedMultiDict): found.extend(_plain_dicts(item, f'{path}.{key}')) return found +def _mutable(info: 'Any') -> 'list[str]': + """Keys of ``info`` holding a mutable :class:`~pcapkit.corekit.multidict.MultiDict`.""" + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo + from pcapkit.corekit.multidict import MultiDict + + return [key for key, value in info.items(multi=True) + if isinstance(value, MultiDict) and not isinstance(value, (MultiInfo, OrderedMultiInfo))] + + +def canonical(value: 'Any') -> 'Any': + """``value`` with every multi-mapping in it turned into its class and pairs, recursively. + + The exports are compared through this rather than with ``==``: an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` compares its values + with ``!=``, which for a nested one is the ``__ne__`` of :obj:`dict` and + compares the internal buckets, so two equal exports with a nested + :class:`Info` compare unequal. + + """ + from pcapkit.corekit.multidict import MultiDict + + if isinstance(value, MultiDict): + return (type(value).__qualname__, [(key, canonical(item)) for key, item in value.items(multi=True)]) + return value + + def check(info: 'Any') -> 'Outcome': """``from_dict(to_dict())`` reproduces ``info`` in every respect listed above.""" try: @@ -158,7 +195,10 @@ def check(info: 'Any') -> 'Outcome': return Outcome('TO_DICT', describe(exc)) plain = _plain_dicts(exported) if plain: - return Outcome('TO_DICT', f'not an InfoDict: {plain}') + return Outcome('TO_DICT', f'not a plain OrderedMultiDict: {plain}') + mutable = _mutable(info) + if mutable: + return Outcome('TYPE', f'mutable multi-mapping held under {mutable!r}') try: rebuilt = type(info).from_dict(exported) except Exception as exc: # pylint: disable=broad-except @@ -169,7 +209,7 @@ def check(info: 'Any') -> 'Outcome': 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)): + if canonical(rebuilt.to_dict()) != canonical(exported): return Outcome('TO_DICT', 'to_dict() of the rebuilt instance differs') return OK diff --git a/tests/corekit/test_infoclass.py b/tests/corekit/test_infoclass.py index 77801c6e1c..87ed38da08 100644 --- a/tests/corekit/test_infoclass.py +++ b/tests/corekit/test_infoclass.py @@ -43,7 +43,7 @@ class NamedPoint(BasePoint): name: str named = NamedPoint(3, 'three') - self.assertEqual(named.to_dict(), {'x': 3, 'name': 'three'}) + self.assertEqual(list(named.to_dict().items(multi=True)), [('x', 3), ('name', 'three')]) version = VersionInfo(1, 2) self.assertEqual(version.version, '1.2') @@ -65,7 +65,7 @@ class Bag(Info): self.assertEqual(len(bag), 3) self.assertEqual(str(bag), 'Bag(items=conflict, visible=1, extra=3)') self.assertIn('items=', repr(bag)) - self.assertEqual(bag.to_dict(), {'items': 'conflict', 'visible': 1, 'extra': 3}) + self.assertEqual(list(bag.to_dict().items(multi=True)), [('items', 'conflict'), ('visible', 1), ('extra', 3)]) with self.assertRaises(UnsupportedCall): bag.visible = 9 @@ -94,13 +94,13 @@ class Child(Base): self.assertEqual(outer['inner'], inner) self.assertIn('inner=Info(...)', repr(outer)) converted = outer.to_dict() - self.assertEqual(converted['inner'], {'answer': 42}) + self.assertEqual(list(converted['inner'].items(multi=True)), [('answer', 42)]) self.assertEqual(converted['kept'], 'yes') self.assertNotIn('base_hidden', converted) self.assertNotIn('child_hidden', converted) - self.assertEqual(Info.from_dict({'a': 1}, b=2).to_dict(), {'a': 1, 'b': 2}) - self.assertEqual(Info.from_dict(c=3).to_dict(), {'c': 3}) + self.assertEqual(Info.from_dict({'a': 1}, b=2).to_dict().to_dict(), {'a': 1, 'b': 2}) + self.assertEqual(Info.from_dict(c=3).to_dict().to_dict(), {'c': 3}) if __name__ == '__main__': diff --git a/tests/corekit/test_multiinfo_1484_unit.py b/tests/corekit/test_multiinfo_1484_unit.py new file mode 100644 index 0000000000..ec8ff0ca01 --- /dev/null +++ b/tests/corekit/test_multiinfo_1484_unit.py @@ -0,0 +1,480 @@ +# -*- coding: utf-8 -*- +"""``MultiInfo``, ``OrderedMultiInfo``, and what ``Info`` does with them. + +GitHub issue #1484: + +* :class:`~pcapkit.corekit.infoclass.MultiInfo` and + :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` are immutable, read a key + as an attribute, and keep every value of a repeated key -- the ordered one in + order across keys; +* both are :class:`~pcapkit.corekit.infoclass.Info` subclasses that inherit + ``from_dict`` and the finalisation rules, while the mapping protocol stays the + multi-mapping's own; +* a finalised :class:`~pcapkit.corekit.infoclass.Info` holds every + :class:`~pcapkit.corekit.multidict.MultiDict` value as one of them; +* :meth:`~pcapkit.corekit.infoclass.Info.to_dict` returns a plain + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` at every level, and + :meth:`~pcapkit.corekit.infoclass.Info.from_dict` restores the ``*Info`` types. + +Every case builds its own values and reads no capture. :mod:`pcapkit` is +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import copy +import enum +import itertools +import pickle # nosec: B403 -- round-trips objects built here +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class + +PAIRS = [('a', 1), ('b', 2), ('a', 3)] + + +class Colour(enum.IntEnum): + """An enum key, as the option lists are keyed.""" + + RED = 1 + BLUE = 2 + + +def canonical(value): # type: ignore[no-untyped-def] + """``value`` with every multi-mapping in it turned into its class and pairs, recursively. + + Exports are compared through this: an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` compares its values + with ``!=``, which for a nested one compares its internal buckets. + + """ + from pcapkit.corekit.multidict import MultiDict + + if isinstance(value, MultiDict): + return (type(value).__qualname__, [(key, canonical(item)) for key, item in value.items(multi=True)]) + return value + + +class MultiInfoTests(unittest.TestCase): + """The two classes on their own.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def classes(self): # type: ignore[no-untyped-def] + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo + return MultiInfo, OrderedMultiInfo + + def test_repeated_keys_are_kept_and_ordered(self) -> None: + MultiInfo, OrderedMultiInfo = self.classes() + + ordered = OrderedMultiInfo(PAIRS) + self.assertEqual(list(ordered.items(multi=True)), PAIRS) + self.assertEqual(list(ordered), ['a', 'b']) + + grouped = MultiInfo(PAIRS) + self.assertEqual(list(grouped.items(multi=True)), [('a', 1), ('a', 3), ('b', 2)]) + + for info in (ordered, grouped): + with self.subTest(cls=type(info).__name__): + self.assertEqual(info['a'], 1) + self.assertEqual(info.getlist('a'), [1, 3]) + self.assertEqual(len(info), 2) + + def test_keys_read_as_attributes(self) -> None: + from pcapkit.utilities.exceptions import UnsupportedCall + + for cls in self.classes(): + with self.subTest(cls=cls.__name__): + info = cls(PAIRS + [(Colour.BLUE, 'blue'), ('items', 'shadowed')]) + self.assertEqual((info.a, info.b), (1, 2)) + self.assertEqual(info.BLUE, 'blue') + # A name the class defines is the method, not the key. + self.assertTrue(callable(info.items)) + self.assertEqual(info['items'], 'shadowed') + self.assertFalse(hasattr(info, 'missing')) + with self.assertRaises(AttributeError) as ctx: + info.missing # pylint: disable=pointless-statement + self.assertIsInstance(ctx.exception, UnsupportedCall) + + def test_every_mutation_is_refused(self) -> None: + from pcapkit.utilities.exceptions import UnsupportedCall + + mutations = { + 'setitem': lambda info: info.__setitem__('a', 0), + 'delitem': lambda info: info.__delitem__('a'), + 'ior': lambda info: info.__ior__({'c': 4}), + 'add': lambda info: info.add('c', 4), + 'setlist': lambda info: info.setlist('a', [0]), + 'setdefault': lambda info: info.setdefault('c', 4), + 'setlistdefault': lambda info: info.setlistdefault('c', [4]), + 'update': lambda info: info.update({'c': 4}), + 'pop': lambda info: info.pop('a'), + 'popitem': lambda info: info.popitem(), + 'poplist': lambda info: info.poplist('a'), + 'popitemlist': lambda info: info.popitemlist(), + 'clear': lambda info: info.clear(), + 'setstate': lambda info: info.__setstate__([]), + 'setattr': lambda info: setattr(info, 'a', 0), + 'delattr': lambda info: delattr(info, 'a'), + } + for cls in self.classes(): + for name, mutate in mutations.items(): + with self.subTest(cls=cls.__name__, mutation=name): + info = cls(PAIRS) + before = list(info.items(multi=True)) + with warnings.catch_warnings(), self.assertRaises(UnsupportedCall): + warnings.simplefilter('ignore') + mutate(info) + self.assertEqual(list(info.items(multi=True)), before) + + def test_copies_keep_the_type_and_every_value(self) -> None: + for cls in self.classes(): + info = cls(PAIRS) + for how, clone in (('copy', copy.copy(info)), ('deepcopy', copy.deepcopy(info)), + ('pickle', pickle.loads(pickle.dumps(info)))): # nosec: B301 + with self.subTest(cls=cls.__name__, how=how): + self.assertIs(type(clone), cls) + self.assertEqual(list(clone.items(multi=True)), list(info.items(multi=True))) + + def test_compares_like_the_multi_dict_it_builds_on(self) -> None: + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + + MultiInfo, OrderedMultiInfo = self.classes() + self.assertEqual(OrderedMultiInfo(PAIRS), OrderedMultiDict(PAIRS)) + self.assertNotEqual(OrderedMultiInfo(PAIRS), OrderedMultiDict(list(reversed(PAIRS)))) + self.assertEqual(MultiInfo(PAIRS), MultiDict(PAIRS)) + self.assertIsInstance(OrderedMultiInfo(PAIRS), OrderedMultiDict) + self.assertIsInstance(MultiInfo(PAIRS), MultiDict) + + +class InfoSubclassTests(unittest.TestCase): + """The two classes as :class:`Info` subclasses -- the owner's ruling on #1484.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_both_are_info_subclasses_with_the_multi_dict_protocol(self) -> None: + from pcapkit.corekit.infoclass import Info, MultiInfo, OrderedMultiInfo + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + from pcapkit.protocols.data.application.httpv2 import Settings + + for cls, base in ((MultiInfo, MultiDict), (OrderedMultiInfo, OrderedMultiDict), (Settings, OrderedMultiDict)): + with self.subTest(cls=cls.__name__): + self.assertTrue(issubclass(cls, Info)) + self.assertIsInstance(cls(PAIRS), Info) + self.assertIsInstance(cls(PAIRS), base) + # Info precedes the multi-mapping in the MRO, yet the mapping + # protocol is the multi-mapping's own. + mro = cls.__mro__ + self.assertLess(mro.index(Info), mro.index(base)) + for name in ('__iter__', '__len__', '__contains__', '__repr__', + 'keys', 'values', 'get', 'getlist', 'lists', 'copy', 'deepcopy'): + self.assertIs(getattr(cls, name), getattr(base, name), name) + # ``==`` is the multi-mapping's for the ordered ones, and its own + # for MultiInfo, which defers to an ordered operand; ``!=`` is + # never dict's, which compares the internal storage (#1484). + if cls is MultiInfo: + self.assertIsNot(cls.__eq__, base.__eq__) + else: + self.assertIs(cls.__eq__, base.__eq__) + self.assertIsNot(cls.__ne__, base.__ne__) + if cls is not Settings: # which overrides both, for its last-value-wins reads + self.assertIs(cls.__getitem__, base.__getitem__) + self.assertIs(cls.items, base.items) + self.assertIsNone(cls.__hash__) + + def test_from_dict_is_inherited_and_round_trips(self) -> None: + from pcapkit.corekit.infoclass import Info, MultiInfo, OrderedMultiInfo, _OrderedMultiDict + from pcapkit.corekit.multidict import MultiDict + + for cls, plain in ((MultiInfo, MultiDict), (OrderedMultiInfo, _OrderedMultiDict)): + with self.subTest(cls=cls.__name__): + self.assertIs(cls.from_dict.__func__, Info.from_dict.__func__) # type: ignore[attr-defined] + info = cls.from_dict(PAIRS) + self.assertIs(type(info), cls) + self.assertEqual(list(info.items(multi=True)), list(plain(PAIRS).items(multi=True))) + + exported = info.to_dict() + self.assertIs(type(exported), plain) + rebuilt = cls.from_dict(exported) + self.assertIs(type(rebuilt), cls) + self.assertEqual(list(rebuilt.items(multi=True)), list(info.items(multi=True))) + self.assertEqual(rebuilt, info) + + # As the multi-mapping's own constructor does: keywords and a + # mapping's list values add a pair each. + self.assertEqual(list(cls.from_dict({'a': [1, 3]}, b=2).items(multi=True)), + list(plain([('a', 1), ('a', 3), ('b', 2)]).items(multi=True))) + # MultiDict's flat forms stay reachable. + self.assertEqual(info.to_dict(flat=True), {'a': 1, 'b': 2}) + self.assertEqual(info.to_dict(flat=False), {'a': [1, 3], 'b': [2]}) + + def test_the_finalisation_rules_apply(self) -> None: + from pcapkit.corekit.infoclass import FinalisedState, OrderedMultiInfo, info_final + from pcapkit.utilities.exceptions import InfoError + + @info_final + class Sealed(OrderedMultiInfo): + """A final option list.""" + + # No ``__init__`` is generated: the class keeps the mapping constructor. + self.assertNotIn('__init__', Sealed.__dict__) + self.assertEqual(Sealed.__dict__['__finalised__'], FinalisedState.FINAL) + self.assertEqual(list(Sealed(PAIRS).items(multi=True)), PAIRS) + self.assertIsInstance(OrderedMultiInfo.__additional__, list) + self.assertIsInstance(OrderedMultiInfo.__excluded__, list) + with self.assertRaises(InfoError): + class Derived(Sealed): # pylint: disable=unused-variable + """Refused: its parent is final.""" + + def test_an_info_writes_its_option_lists_in_full(self) -> None: + from pcapkit.corekit.infoclass import Info + from pcapkit.corekit.multidict import OrderedMultiDict + + info = Info.from_dict({'nested': Info(x=1), 'options': OrderedMultiDict(PAIRS)}) + self.assertEqual(repr(info), "") + self.assertEqual(str(info.options), "OrderedMultiInfo([('a', 1), ('b', 2), ('a', 3)])") + + +class ExportComparisonTests(unittest.TestCase): + """``to_dict()`` exports compare as mappings should -- the owner's ``__ne__`` ruling on #1484.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def info(self, last: int = 3): # type: ignore[no-untyped-def] + """An :class:`Info` holding a nested model, both option lists and a nested option list.""" + from pcapkit.corekit.infoclass import Info + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + + return Info.from_dict(OrderedMultiDict([ + ('nested', Info(x=1, options=OrderedMultiDict(PAIRS))), + ('ordered', OrderedMultiDict([('a', 1), ('b', 2), ('a', last)])), + ('grouped', MultiDict(PAIRS)), + ('nested', Info(x=2)), + ])) + + def test_equal_exports_are_equal_and_not_unequal(self) -> None: + info = self.info() + first, second = info.to_dict(), info.to_dict() + self.assertTrue(first == second) + self.assertFalse(first != second) + for ((key, one), (_, two)) in zip(first.items(multi=True), second.items(multi=True)): + with self.subTest(key=key): + self.assertTrue(one == two) + self.assertFalse(one != two) + self.assertFalse(info.ordered.to_dict() != info.ordered.to_dict()) + self.assertFalse(info.nested.options.to_dict() != info.nested.options.to_dict()) + + # A real difference, however deep, still shows. + other = self.info(last=4).to_dict() + self.assertTrue(first != other) + self.assertFalse(first == other) + # As documented: an export is never equal to a plain dict. + self.assertTrue(first['ordered'] != {'a': 1, 'b': 2}) + self.assertFalse(first['ordered'] == {'a': 1, 'b': 2}) + + def test_an_option_list_is_unequal_exactly_when_it_is_not_equal(self) -> None: + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + from pcapkit.protocols.data.application.httpv2 import Settings + + swapped = [('b', 2), ('a', 1), ('a', 3)] + for cls, plain in ((OrderedMultiInfo, OrderedMultiDict), (Settings, OrderedMultiDict), (MultiInfo, MultiDict)): + info = cls(PAIRS) + cases = { + 'an equal copy': (cls(PAIRS), True), + 'another order': (cls(swapped), cls is MultiInfo), # only MultiInfo ignores the order + f'an equal {plain.__name__}': (plain(PAIRS), True), + f'a reordered {plain.__name__}': (plain(swapped), cls is MultiInfo), + 'a dict': ({'a': 1, 'b': 2}, False), + 'its own export': (info.to_dict(), True), + } + for name, (other, equal) in cases.items(): + with self.subTest(cls=cls.__name__, other=name): + for (one, two) in ((info, other), (other, info)): + self.assertIs(one == two, equal) + self.assertIs(one != two, not equal) + + def test_comparisons_are_symmetric_and_negated_across_every_kind(self) -> None: + """``a == b`` is ``b == a``, and ``a != b`` is ``not a == b``, for every pair of kinds.""" + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + from pcapkit.protocols.data.application.httpv2 import Settings + + reordered = [('b', 2), ('a', 1), ('a', 3)] + kinds = { + 'MultiInfo': MultiInfo, + 'OrderedMultiInfo': OrderedMultiInfo, + 'Settings': Settings, + 'OrderedMultiInfo export': lambda pairs: OrderedMultiInfo(pairs).to_dict(), + 'MultiInfo export': lambda pairs: MultiInfo(pairs).to_dict(), + 'MultiDict': MultiDict, + 'OrderedMultiDict': OrderedMultiDict, + 'dict': lambda pairs: MultiDict(pairs).to_dict(flat=True), + } + ordered = {'OrderedMultiInfo', 'Settings', 'OrderedMultiInfo export', 'OrderedMultiDict'} + + def expected(left: str, right: str, same_order: bool) -> bool: + if 'dict' in (left, right): + return left == right + return same_order or not (left in ordered and right in ordered) + + def verbatim_ne(one: object, two: object) -> bool: + # NOTE: Werkzeug's OrderedMultiDict, kept verbatim in + # pcapkit.corekit.multidict (the owner's option 3 on #1484), has no + # ``__ne__``, so ``!=`` is dict's and compares the internal buckets. + # Python asks it first when it is the left operand, and when it is + # the right one of a bare MultiDict, which overrides nothing either; + # no code outside that module is then consulted. + return type(one) is OrderedMultiDict or (type(one) is MultiDict and type(two) is OrderedMultiDict) + + for (left, make_left), (right, make_right) in itertools.product(kinds.items(), repeat=2): + for same_order, pairs in ((True, PAIRS), (False, reordered)): + with self.subTest(left=left, right=right, order='same' if same_order else 'reordered'): + one, two = make_left(PAIRS), make_right(pairs) + self.assertIs(one == two, expected(left, right, same_order)) + self.assertIs(two == one, one == two) + for (a, b) in ((one, two), (two, one)): + if not verbatim_ne(a, b): + self.assertIs(a != b, not a == b) + + def test_copies_of_an_export_keep_its_class(self) -> None: + from pcapkit.corekit.multidict import OrderedMultiDict + + exported = self.info().to_dict() + self.assertIsInstance(exported, OrderedMultiDict) + self.assertTrue(repr(exported).startswith("OrderedMultiDict([('nested', OrderedMultiDict([")) + for how, clone in (('copy()', exported.copy()), ('deepcopy()', exported.deepcopy()), + ('copy.copy', copy.copy(exported)), ('copy.deepcopy', copy.deepcopy(exported)), + ('pickle', pickle.loads(pickle.dumps(exported)))): # nosec: B301 + with self.subTest(how=how): + self.assertIs(type(clone), type(exported)) + self.assertIs(type(clone['ordered']), type(exported['ordered'])) + self.assertFalse(clone != exported) + self.assertTrue(clone == exported) + + def test_listvalues_hands_out_copies(self) -> None: + from pcapkit.corekit.infoclass import MultiInfo, OrderedMultiInfo + + for cls in (MultiInfo, OrderedMultiInfo): + with self.subTest(cls=cls.__name__): + info = cls(PAIRS) + self.assertEqual(list(info.listvalues()), [[1, 3], [2]]) + next(iter(info.listvalues())).append(99) + self.assertEqual(info.getlist('a'), [1, 3]) + self.assertEqual(list(info.items(multi=True)), list(cls(PAIRS).items(multi=True))) + + +class InfoFinalisationTests(unittest.TestCase): + """What :class:`Info` stores, exports and restores.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_a_finalised_info_holds_no_mutable_multi_mapping(self) -> None: + from pcapkit.corekit.infoclass import Info, MultiInfo, OrderedMultiInfo + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + + source = OrderedMultiDict(PAIRS) + info = Info.from_dict({'ordered': source, 'grouped': MultiDict(PAIRS), 'n': 1}) + self.assertIs(type(info.ordered), OrderedMultiInfo) + self.assertIs(type(info.grouped), MultiInfo) + self.assertEqual(list(info.ordered.items(multi=True)), PAIRS) + + # The parser's mapping is copied, so editing it afterwards does not reach the info. + source.add('z', 9) + self.assertEqual(list(info.ordered.items(multi=True)), PAIRS) + + # So is a value added through a MultiDict, i.e. as a repeated key. + info.__update__(OrderedMultiDict([('ordered', OrderedMultiDict([('c', 4)]))])) + self.assertEqual([type(value) for key, value in info.items(multi=True) if key == 'ordered'], + [OrderedMultiInfo, OrderedMultiInfo]) + + def test_to_dict_returns_a_plain_ordered_multi_dict_at_every_level(self) -> None: + from pcapkit.corekit.infoclass import Info, _OrderedMultiDict + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + + info = Info.from_dict(OrderedMultiDict([ + ('nested', Info(x=1)), ('ordered', OrderedMultiDict(PAIRS)), ('grouped', MultiDict(PAIRS)), + ('nested', Info(x=2)), + ])) + exported = info.to_dict() + self.assertIs(type(exported), _OrderedMultiDict) + self.assertIsInstance(exported, OrderedMultiDict) + self.assertEqual([key for key, _ in exported.items(multi=True)], ['nested', 'ordered', 'grouped', 'nested']) + self.assertEqual([type(value) for _, value in exported.items(multi=True)], + [_OrderedMultiDict, _OrderedMultiDict, MultiDict, _OrderedMultiDict]) + self.assertEqual([value['x'] for value in exported.getlist('nested')], [1, 2]) + self.assertEqual(list(exported['ordered'].items(multi=True)), PAIRS) + + # The export is a copy the caller may edit. + exported['ordered'].add('z', 9) + self.assertEqual(list(info.ordered.items(multi=True)), PAIRS) + + def test_from_dict_restores_the_info_types(self) -> None: + from pcapkit.corekit.infoclass import Info, MultiInfo, OrderedMultiInfo, info_final + from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict + + @info_final + class Leaf(Info): + """A nested model.""" + + x: int + + @info_final + class Node(Info): + """A model holding both kinds of option list.""" + + leaf: Leaf + ordered: OrderedMultiInfo + grouped: MultiInfo + + info = Node(leaf=Leaf(x=1), ordered=OrderedMultiDict(PAIRS), grouped=MultiDict(PAIRS)) + rebuilt = Node.from_dict(info.to_dict()) + self.assertEqual(rebuilt, info) + self.assertEqual([type(value) for _, value in rebuilt.items(multi=True)], [Leaf, OrderedMultiInfo, MultiInfo]) + self.assertEqual(canonical(rebuilt.to_dict()), canonical(info.to_dict())) + + def test_from_dict_restores_an_annotated_subclass(self) -> None: + """HTTP/2 ``Settings`` comes back as itself, last value wins and all.""" + from pcapkit.const.http.setting import Setting + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.data.application.httpv2 import Settings, SettingsFrame + + pairs = [(Setting.ENABLE_PUSH, 1), (Setting.ENABLE_PUSH, 0)] + frame = SettingsFrame.from_dict({'settings': OrderedMultiDict(pairs)}) + self.assertIs(type(frame.settings), Settings) + self.assertEqual(frame.settings[Setting.ENABLE_PUSH], 0) + + exported = frame.to_dict() + self.assertIsInstance(exported['settings'], OrderedMultiDict) + self.assertNotIsInstance(exported['settings'], Settings) + rebuilt = SettingsFrame.from_dict(exported) + self.assertIs(type(rebuilt.settings), Settings) + self.assertEqual(list(rebuilt.settings.items(multi=True)), pairs) + + def test_parsed_options_are_frozen_and_rebuild_from_either_form(self) -> None: + from pcapkit.corekit.infoclass import OrderedMultiInfo + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.transport.tcp import TCP + + # MSS 1460, NOP, NOP, SACK-permitted: a 28-octet header. + options = bytes.fromhex('020405b4' '0101' '0402') + header = struct.pack('!HHIIBBHHH', 1, 2, 0, 0, ((20 + len(options)) // 4) << 4, 0x02, 1024, 0, 0) + raw = header + options + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + tcp = TCP(raw, len(raw)) + self.assertIs(type(tcp.info.options), OrderedMultiInfo) + self.assertIsInstance(tcp.info.to_dict()['options'], OrderedMultiDict) + self.assertNotIsInstance(tcp.info.to_dict()['options'], OrderedMultiInfo) + self.assertEqual(TCP.from_data(tcp.info).data, raw) + self.assertEqual(TCP.from_data(tcp.info.to_dict()).data, raw) + self.assertEqual(type(tcp.info).from_dict(tcp.info.to_dict()), tcp.info) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/dumpkit/test_repeated_field_dump_1484_unit.py b/tests/dumpkit/test_repeated_field_dump_1484_unit.py new file mode 100644 index 0000000000..fd5f26d61e --- /dev/null +++ b/tests/dumpkit/test_repeated_field_dump_1484_unit.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +"""Every value of a repeated field reaches the ``json``, ``plist`` and ``tree`` dumps. + +GitHub issue #1484. An IPv6 packet carrying two Destination Options headers +holds both under the one ``opts`` key, and every dump held only the first. +Each dump now writes the field as the list of both headers, in wire order, +while a field held once is written as its value. + +Every case builds its input in a temporary directory, and reads no capture. +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import importlib.util +import json +import os +import plistlib +import struct +import tempfile +import unittest +import warnings + +from tests._support import reimport_once_per_class + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +DST, UDP = 60, 17 +MAC = bytes.fromhex('0123456789ab' 'fedcba987654') +SRC6 = bytes.fromhex('20010db8' + '00' * 11 + '01') +DST6 = bytes.fromhex('20010db8' + '00' * 11 + '02') + +#: Little-endian global header: v2.4, Ethernet. +GLOBAL = bytes.fromhex('d4c3b2a1' '0200' '0400' '00000000' '00000000' 'ffff0000' '01000000') + +#: Last octet of each header's PadN, telling the two apart. +MARKS = (0xa1, 0xb2) + + +def capture() -> bytes: + """A PCAP file of one Ethernet/IPv6 frame: Destination Options twice, then UDP.""" + body = bytes([DST, 0, 1, 4, 0, 0, 0, MARKS[0]]) + bytes([UDP, 0, 1, 4, 0, 0, 0, MARKS[1]]) + body += struct.pack('!HHHH', 40000, 40000, 12, 0) + b'ping' + packet = MAC + b'\x86\xdd' + struct.pack('!IHBB16s16s', 6 << 28, len(body), DST, 64, SRC6, DST6) + body + return GLOBAL + struct.pack(' 'list[str]': + """The PadN octets of each Destination Options header, as the JSON dump spells them.""" + return [next(iter(header['options'][0].values()))['pad']['hex'] for header in headers] + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class RepeatedFieldDumpTests(unittest.TestCase): + """Read each dump back and find both headers in it.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + self.tmpdir = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + self.addCleanup(self.tmpdir.cleanup) + quiet = warnings.catch_warnings() + quiet.__enter__() # pylint: disable=unnecessary-dunder-call + self.addCleanup(quiet.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def dump(self, fmt: 'str') -> 'bytes': + """Extract :func:`capture` as ``fmt``; return the output file.""" + from pcapkit import extract + + src = os.path.join(self.tmpdir.name, 'in.pcap') + with open(src, 'wb') as file: + file.write(capture()) + extractor = extract(fin=src, fout=os.path.join(self.tmpdir.name, f'out-{fmt}'), format=fmt, store=False) + with open(extractor.output, 'rb') as file: + return file.read() + + def test_json_holds_both_headers_in_order(self) -> None: + ipv6 = json.loads(self.dump('json'))['Frame 1']['ethernet']['ipv6'] + self.assertIsInstance(ipv6['opts'], list) + self.assertEqual(pads(ipv6['opts']), [f'000000{mark:02x}' for mark in MARKS]) + # A field held once is still written as its value. + self.assertEqual(ipv6['limit'], 64) + + def test_plist_holds_both_headers_in_order(self) -> None: + ipv6 = plistlib.loads(self.dump('plist'))['Frame 1']['ethernet']['ipv6'] + self.assertIsInstance(ipv6['opts'], list) + self.assertEqual([next(iter(header['options'][0].values()))['pad'] for header in ipv6['opts']], + [b'\x00\x00\x00' + bytes([mark]) for mark in MARKS]) + + def test_tree_holds_both_headers(self) -> None: + text = self.dump('tree').decode('utf-8') + for mark in MARKS: + with self.subTest(mark=mark): + self.assertIn(f'pad -> 00 00 00 {mark:02x}', text) + + def test_the_writer_takes_the_info_or_its_export(self) -> None: + """Handed the info, the writer holds both headers; handed ``to_dict()``, the same top level. + + Below the top level an export cannot tell a nested :class:`Info` from an + option list, both being an ``OrderedMultiDict``, so only the info itself + is written with nested fields as mappings. + + """ + import io + + import dictdumper + + from pcapkit.dumpkit.common import make_dumper + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcap.header import Header + + raw = capture() + frame = Frame(io.BytesIO(raw[24:]), num=1, header=Header(raw[:24]).info) + tops = [] + for index, value in enumerate((frame.info, frame.info.to_dict())): + path = os.path.join(self.tmpdir.name, f'{index}.json') + make_dumper(dictdumper.JSON)(path)(value, name='Frame 1') + with open(path, encoding='utf-8') as file: + tops.append(json.load(file)['Frame 1']) + self.assertEqual(pads(tops[0]['ethernet']['ipv6']['opts']), [f'000000{mark:02x}' for mark in MARKS]) + self.assertEqual(list(tops[1]), list(tops[0])) + self.assertEqual(list(tops[0]), list(frame.info.to_dict())) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/application/test_http2_roundtrip_unit.py b/tests/protocols/application/test_http2_roundtrip_unit.py index ca3b61f117..64811cb613 100644 --- a/tests/protocols/application/test_http2_roundtrip_unit.py +++ b/tests/protocols/application/test_http2_roundtrip_unit.py @@ -120,7 +120,9 @@ def test_repeated_settings_index_to_the_last_value(self) -> None: self.assertEqual(settings[Setting.HEADER_TABLE_SIZE], 3) self.assertEqual(settings.get(Setting.HEADER_TABLE_SIZE), 3) self.assertEqual(dict(settings), {Setting.HEADER_TABLE_SIZE: 3, Setting.ENABLE_PUSH: 0}) - self.assertEqual(settings.to_dict(), {Setting.HEADER_TABLE_SIZE: 3, Setting.ENABLE_PUSH: 0}) + self.assertEqual(settings.to_dict(flat=True), {Setting.HEADER_TABLE_SIZE: 3, Setting.ENABLE_PUSH: 0}) + # The Info export keeps every entry, in wire order (#1484). + self.assertEqual(list(settings.to_dict().items(multi=True)), list(settings.items(multi=True))) self.assertEqual(settings.getlist(Setting.HEADER_TABLE_SIZE), [1, 3]) self.assertEqual(bytes(HTTP.from_data(info)).hex(), wire.hex()) diff --git a/tests/protocols/application/test_ospf_auth_dict_roundtrip_unit.py b/tests/protocols/application/test_ospf_auth_dict_roundtrip_unit.py index e1da893f8c..31847ab72e 100644 --- a/tests/protocols/application/test_ospf_auth_dict_roundtrip_unit.py +++ b/tests/protocols/application/test_ospf_auth_dict_roundtrip_unit.py @@ -71,12 +71,15 @@ def test_from_data_rebuilds_object_form_byte_for_byte(self) -> None: self.assertEqual(OSPF.from_data(parsed.info).data, raw) def test_to_dict_flattens_cryptographic_authentication(self) -> None: - # The premise of the fix: the dict form really does carry a plain dict. + # The premise of the fix: the dict form really does carry a plain + # mapping -- the OrderedMultiDict ``to_dict()`` writes (#1484). + from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.application.ospf import OSPF raw = PACKETS['autype-2-nonzero-reserved'] auth = OSPF(raw, len(raw)).info.to_dict()['auth'] - self.assertEqual(auth, {'reserved': b'\xbe\xef', 'key_id': 255, 'len': 16, 'seq': 0xffffffff}) + self.assertIsInstance(auth, OrderedMultiDict) + self.assertEqual(auth.to_dict(), {'reserved': b'\xbe\xef', 'key_id': 255, 'len': 16, 'seq': 0xffffffff}) def test_make_accepts_dict_auth_data(self) -> None: from pcapkit.protocols.application.ospf import OSPF diff --git a/tests/protocols/internet/test_ipv4_roundtrip_unit.py b/tests/protocols/internet/test_ipv4_roundtrip_unit.py index 130dc9650f..a4aa99ba4a 100644 --- a/tests/protocols/internet/test_ipv4_roundtrip_unit.py +++ b/tests/protocols/internet/test_ipv4_roundtrip_unit.py @@ -188,17 +188,20 @@ def test_edited_options_recompute_the_header_length(self) -> None: rtralt = self.parse(header('94040000')).info.options[OptionNumber.RTRALT] + # NOTE: a parsed header's options are an immutable OrderedMultiInfo + # (#1484), so they are edited in the mutable copy ``to_dict()`` makes. + # one option added to a parsed header grows it - parsed = self.parse(header('88040001')) - parsed.info.options.add(OptionNumber.RTRALT, rtralt) - data = IPv4.from_data(parsed.info).data + edited = self.parse(header('88040001')).info.to_dict() + edited['options'].add(OptionNumber.RTRALT, rtralt) + data = IPv4.from_data(edited).data self.assertEqual(data[0], 0x47) self.assertEqual(data[20:].hex(), '88040001' '94040000') # one option removed shrinks it, rather than zero-filling the old length - parsed = self.parse(header('88040001' '94040000')) - parsed.info.options.pop(OptionNumber.RTRALT) - data = IPv4.from_data(parsed.info).data + edited = self.parse(header('88040001' '94040000')).info.to_dict() + edited['options'].pop(OptionNumber.RTRALT) + data = IPv4.from_data(edited).data self.assertEqual(data[0], 0x46) self.assertEqual(data[20:].hex(), '88040001') diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index c394930b78..78b7abf59a 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -670,7 +670,9 @@ def _assert_option_header_read_make_and_registry_edges(self, protocol_cls: type) reader.assert_called_once_with(6) self.assertEqual(ext_data.next, TransType.UDP) self.assertEqual(ext_data.length, 8) - self.assertIs(ext_data.options, empty_options) + # NOTE: held as the immutable OrderedMultiInfo of it (#1484). + self.assertEqual(type(ext_data.options).__name__, 'OrderedMultiInfo') + self.assertEqual(list(ext_data.options.items(multi=True)), list(empty_options.items(multi=True))) with mock.patch.object(proto, read_options, return_value=empty_options): with mock.patch.object(protocol_cls, '_decode_next_layer', return_value='decoded') as decode: diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index bfead3593c..ba56631f1e 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -1359,11 +1359,18 @@ def test_pcapng_record_and_secrets_readers_cover_dispatch(self) -> None: tls_entries = {TLSKeyLabel.CLIENT_RANDOM: OrderedMultiDict([(b'random', b'secret')])} tls = TLSKeyLog(data='CLIENT_RANDOM 00 11') object.__setattr__(tls, 'entries', tls_entries) - self.assertIs(pcapng._read_secrets_tls(tls, block=block).entries, tls_entries) + # NOTE: held as immutable OrderedMultiInfo copies (#1484). + tls_read = pcapng._read_secrets_tls(tls, block=block).entries + self.assertEqual(list(tls_read), list(tls_entries)) + self.assertEqual({type(entry).__name__ for entry in tls_read.values()}, {'OrderedMultiInfo'}) + self.assertEqual([list(entry.items(multi=True)) for entry in tls_read.values()], + [list(entry.items(multi=True)) for entry in tls_entries.values()]) wg_entries = OrderedMultiDict([(WireGuardKeyLabel.PRESHARED_KEY, b'key')]) wireguard = WireGuardKeyLog(data='PRESHARED_KEY = a2V5') object.__setattr__(wireguard, 'entries', wg_entries) - self.assertIs(pcapng._read_secrets_wireguard(wireguard, block=block).entries, wg_entries) + wg_read = pcapng._read_secrets_wireguard(wireguard, block=block).entries + self.assertEqual(type(wg_read).__name__, 'OrderedMultiInfo') + self.assertEqual(list(wg_read.items(multi=True)), list(wg_entries.items(multi=True))) self.assertEqual(pcapng._read_secrets_zigbee_nwk( ZigBeeNWKKey(key=b'\x01' * 16, panid=0x1234), block=block, ).pan_id, 0x1234) diff --git a/tests/protocols/test_to_dict_compare_1484_runtime.py b/tests/protocols/test_to_dict_compare_1484_runtime.py new file mode 100644 index 0000000000..1af1a23d25 --- /dev/null +++ b/tests/protocols/test_to_dict_compare_1484_runtime.py @@ -0,0 +1,109 @@ +# -*- coding: utf-8 -*- +"""Two exports of the same packet compare ``==`` and not ``!=``. + +GitHub issue #1484. :meth:`Info.to_dict ` +returns an :class:`~pcapkit.corekit.multidict.OrderedMultiDict`, and Werkzeug's +class defines ``__eq__`` but no ``__ne__``: ``!=`` fell back to :obj:`dict`'s, +which compares the internal buckets, so ``info.to_dict() != info.to_dict()`` +held for every frame. The export now negates its ``__eq__``. + +For every frame of the option-heavy captures in :file:`examples/captures/`: + +* two exports of the frame's info are ``==`` and not ``!=``, at the top and at + every nested multi-mapping in them; +* two exports of every option list held by a layer's info are likewise. + +The module reads generated captures, so it belongs to the fixture-dependent tier. + +""" +from __future__ import annotations + +import importlib.util +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import close_extractor, reimport_once_per_class, time_limit +from tests._tiers import SAMPLE_ROOT + +if TYPE_CHECKING: + from typing import Any, Iterator + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Captures whose frames carry option lists, PCAP and PCAP-NG alike. +CAPTURES = ('options-internet.pcap', 'options-ipv4.pcap', 'options-ipv6.pcap', 'options-tcp.pcap', + 'options-transport.pcap', 'test.pcapng', 'dhcp.pcapng') + +#: Whole seconds the sweep may take; it takes about two. +SWEEP_TIMEOUT = 120 + + +def nested_pairs(one: 'Any', two: 'Any', path: 'str') -> 'Iterator[tuple[str, Any, Any]]': + """``(path, a, b)`` for every multi-mapping at the same place in exports ``one`` and ``two``.""" + from pcapkit.corekit.multidict import MultiDict + + if isinstance(one, MultiDict) and isinstance(two, MultiDict): + yield path, one, two + for ((key, a), (_, b)) in zip(one.items(multi=True), two.items(multi=True)): + yield from nested_pairs(a, b, f'{path}/{key}') + + +def option_lists(info: 'Any', path: 'str') -> 'Iterator[tuple[str, Any]]': + """``(path, value)`` for every option list a layer's info holds, however deep.""" + from pcapkit.corekit.infoclass import Info, _MultiInfo + + for key, value in info.items(multi=True): + if isinstance(value, _MultiInfo): + yield f'{path}/{key}', value + elif isinstance(value, Info): + yield from option_lists(value, f'{path}/{key}') + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ExportComparisonTests(unittest.TestCase): + """Compare two exports of every frame and of every option list.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + quiet = warnings.catch_warnings() + quiet.__enter__() # pylint: disable=unnecessary-dunder-call + self.addCleanup(quiet.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def frames(self) -> 'Iterator[tuple[str, Any]]': + """``(case, frame)`` for every frame of :data:`CAPTURES`.""" + from pcapkit import extract + + for name in CAPTURES: + extractor = extract(fin=str(SAMPLE_ROOT / name), store=True, nofile=True) + self.addCleanup(close_extractor, extractor) + for number, frame in enumerate(extractor.frame, start=1): + yield f'{name}#{number}', frame + + def test_two_exports_of_a_frame_are_equal_and_not_unequal(self) -> None: + nested = 0 + with time_limit(SWEEP_TIMEOUT): + for case, frame in self.frames(): + for path, one, two in nested_pairs(frame.info.to_dict(), frame.info.to_dict(), case): + nested += path != case + with self.subTest(case=path): + self.assertTrue(one == two) + self.assertFalse(one != two) + self.assertGreater(nested, 0, 'no nested multi-mapping was compared') + + def test_two_exports_of_an_option_list_are_equal_and_not_unequal(self) -> None: + compared = 0 + with time_limit(SWEEP_TIMEOUT): + for case, frame in self.frames(): + for path, options in option_lists(frame.info, case): + compared += 1 + with self.subTest(case=path): + self.assertTrue(options.to_dict() == options.to_dict()) + self.assertFalse(options.to_dict() != options.to_dict()) + self.assertGreater(compared, 0, 'no option list was compared') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py b/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py index 5a5e4eadcc..d78b69b9c7 100644 --- a/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py +++ b/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py @@ -6,9 +6,10 @@ overwrote the first and the dict rebuild silently came back shorter -- Destination Options, Destination Options, UDP rebuilt 56 of 64 octets. -``to_dict`` now returns an :class:`~pcapkit.corekit.multidict.InfoDict`: the -:obj:`dict` interface, and so every dump, sees the first value of each key, and -``items(multi=True)`` yields every header in wire order for the rebuild. +``to_dict`` returns an :class:`~pcapkit.corekit.multidict.OrderedMultiDict` +(#1484): indexing sees the first value of each key, and ``items(multi=True)`` +yields every header in wire order for the rebuild. The dumpers write every +header too. Every case builds its own octets and reads no capture. Classes are imported inside each test, after :func:`~tests._support.reimport_once_per_class`. @@ -65,12 +66,18 @@ def ipv6(codes: 'list[int]') -> bytes: } -def plain(value): # type: ignore[no-untyped-def] - """``value`` with every :class:`InfoDict` in it turned into the :obj:`dict` of its first values.""" - from pcapkit.corekit.multidict import InfoDict +def canonical(value): # type: ignore[no-untyped-def] + """``value`` with every multi-mapping in it turned into its class and pairs, recursively. - if isinstance(value, InfoDict): - return {key: plain(val) for key, val in value.items()} + Exports are compared through this: an + :class:`~pcapkit.corekit.multidict.OrderedMultiDict` compares its values + with ``!=``, which for a nested one compares its internal buckets. + + """ + from pcapkit.corekit.multidict import MultiDict + + if isinstance(value, MultiDict): + return (type(value).__qualname__, [(key, canonical(val)) for key, val in value.items(multi=True)]) return value @@ -123,26 +130,24 @@ def test_rebuilt_chain_has_the_parsed_headers(self) -> None: [ext.data for ext in ip6.extension_headers.values()]) def test_to_dict_keeps_every_header_in_wire_order(self) -> None: - from pcapkit.corekit.multidict import InfoDict + from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.internet.ipv6 import IPv6 for name in CHAINS: with self.subTest(chain=name): ip6 = self.frame(name)[IPv6] dict_ = ip6.info.to_dict() - self.assertIsInstance(dict_, InfoDict) - self.assertIsInstance(dict_, dict) + self.assertIsInstance(dict_, OrderedMultiDict) - parsed = [ext.info.to_dict() for ext in ip6.extension_headers.values()] - held = [val for _, val in dict_.items(multi=True) if val in parsed] + parsed = [canonical(ext.info.to_dict()) for ext in ip6.extension_headers.values()] + held = [canonical(val) for _, val in dict_.items(multi=True) if canonical(val) in parsed] self.assertEqual(held, parsed) - # The dict interface, and so every dump, sees the first of each. + # Indexing sees the first of each. first = {} # type: dict[str, object] for key, val in dict_.items(multi=True): first.setdefault(key, val) - self.assertEqual(dict(dict_.items()), first) - self.assertEqual(json.dumps(dict_, default=str), json.dumps(plain(dict_), default=str)) + self.assertEqual({key: dict_[key] for key in dict_}, first) def test_info_from_dict_keeps_the_repeats(self) -> None: from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 @@ -155,33 +160,26 @@ def test_info_from_dict_keeps_the_repeats(self) -> None: rebuilt = Data_IPv6.from_dict(dict_) self.assertEqual([key for key, _ in rebuilt.items(multi=True)], [key for key, _ in dict_.items(multi=True)]) - self.assertEqual(rebuilt.to_dict(), dict_) - self.assertEqual(list(rebuilt.to_dict().items(multi=True)), list(dict_.items(multi=True))) + self.assertEqual(canonical(rebuilt.to_dict()), canonical(dict_)) self.assertEqual(rebuilt, info) - def test_info_dict_behaves_like_a_dict(self) -> None: + def test_to_dict_is_a_plain_ordered_multi_dict(self) -> None: from pcapkit.corekit.infoclass import Info - from pcapkit.corekit.multidict import InfoDict, OrderedMultiDict + from pcapkit.corekit.multidict import OrderedMultiDict dict_ = Info(a=1, b=Info(c=2)).to_dict() - self.assertEqual(dict_, {'a': 1, 'b': {'c': 2}}) - self.assertEqual({'b': {'c': 2}, 'a': 1}, dict_) - self.assertFalse(dict_ != {'a': 1, 'b': {'c': 2}}) - self.assertNotEqual(dict_, {'a': 1}) - self.assertEqual(dict(dict_), {'a': 1, 'b': {'c': 2}}) - self.assertEqual({**dict_}, {'a': 1, 'b': {'c': 2}}) - self.assertEqual(dict_.keys() & {'a', 'z'}, {'a'}) - self.assertEqual(list(dict_.values())[0], 1) - self.assertEqual(len(dict_.items()), 2) + self.assertIsInstance(dict_, OrderedMultiDict) + self.assertIsInstance(dict_['b'], OrderedMultiDict) + self.assertEqual(dict_.to_dict(flat=True)['a'], 1) + self.assertEqual(list(dict_), ['a', 'b']) repeated = Info.from_dict(OrderedMultiDict([('a', 1), ('b', 2), ('a', 3)])).to_dict() self.assertEqual(list(repeated.items(multi=True)), [('a', 1), ('b', 2), ('a', 3)]) self.assertEqual(repeated['a'], 1) - self.assertNotEqual(repeated, {'a': 1, 'b': 2}) - self.assertTrue(repeated != {'a': 1, 'b': 2}) + self.assertEqual(repeated.getlist('a'), [1, 3]) for clone in (copy.copy(repeated), copy.deepcopy(repeated), pickle.loads(pickle.dumps(repeated))): # nosec: B301 with self.subTest(clone=clone): - self.assertIs(type(clone), InfoDict) + self.assertIs(type(clone), type(repeated)) self.assertEqual(list(clone.items(multi=True)), [('a', 1), ('b', 2), ('a', 3)]) def test_info_copy_keeps_its_own_repeats(self) -> None: @@ -199,22 +197,24 @@ def test_info_copy_keeps_its_own_repeats(self) -> None: self.assertEqual(len(list(clone.items(multi=True))), len(before) + 1) self.assertEqual(list(info.items(multi=True)), before) - def test_dumper_writes_the_dict_one_value_per_key(self) -> None: + def test_dumper_writes_every_header(self) -> None: import dictdumper.json from pcapkit.dumpkit.common import make_dumper - dict_ = self.frame('dst-route-dst').info.to_dict() + frame = self.frame('dst-route-dst') outputs = [] with tempfile.TemporaryDirectory() as tmp: - for index, value in enumerate((dict_, plain(dict_))): + for index, value in enumerate((frame.info, frame.info.to_dict())): path = os.path.join(tmp, f'{index}.json') make_dumper(dictdumper.json.JSON)(path)(value, name='Frame 1') with open(path, encoding='utf-8') as file: - outputs.append(file.read()) - self.assertEqual(outputs[0], outputs[1]) - self.assertIn('"ipv6": {', outputs[0]) - + outputs.append(json.load(file)) + ipv6 = outputs[0]['Frame 1']['ethernet']['ipv6'] + self.assertEqual(len(ipv6['opts']), 2) + self.assertIsInstance(ipv6['route'], dict) + # Handed the export, the dumper writes the top level the same way. + self.assertEqual(list(outputs[1]['Frame 1']), list(outputs[0]['Frame 1'])) if __name__ == '__main__': unittest.main() diff --git a/tests/test_final_enforcement.py b/tests/test_final_enforcement.py index 61d4affe0e..4fa7beb0ae 100644 --- a/tests/test_final_enforcement.py +++ b/tests/test_final_enforcement.py @@ -145,7 +145,7 @@ class Sealed(Info): self.assertIn('has been finalised', str(emitted[0].message)) self.assertIs(again, Sealed) - self.assertEqual(again(1).to_dict(), {'x': 1}) + self.assertEqual(list(again(1).to_dict().items(multi=True)), [('x', 1)]) def test_a_base_state_info_class_is_still_subclassable(self) -> None: """The line the check must not cross. @@ -171,7 +171,7 @@ class Extended(Intermediate): y: int self.assertIs(info_final(Extended), Extended) - self.assertEqual(Extended(1, 2).to_dict(), {'x': 1, 'y': 2}) + self.assertEqual(list(Extended(1, 2).to_dict().items(multi=True)), [('x', 1), ('y', 2)]) @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') @@ -363,7 +363,7 @@ class Child(Parent): self.assertEqual([str(record.message) for record in caught], []) self.assertIn('__final__', Child.__dict__) - self.assertEqual(Child(1, 2).to_dict(), {'x': 1, 'y': 2}) + self.assertEqual(list(Child(1, 2).to_dict().items(multi=True)), [('x', 1), ('y', 2)]) def test_an_unfinalised_descendant_is_not_mistaken_for_a_mismarked_class(self) -> None: """:meth:`Info.__new__`'s guard must not fire on an inherited marker. @@ -389,7 +389,7 @@ class Child(Parent): self.assertNotIn('__final__', Child.__dict__) # No InfoError: the marker is inherited, not owned by this class. - self.assertEqual(Child(1, 2).to_dict(), {'x': 1, 'y': 2}) + self.assertEqual(list(Child(1, 2).to_dict().items(multi=True)), [('x', 1), ('y', 2)]) def test_a_subclass_that_predates_an_unfinalised_ancestors_bare_final_is_not_blamed_for_it(self) -> None: """The one shape that actually tells ``__dict__`` and ``getattr`` apart @@ -434,7 +434,7 @@ class Child(Parent): # No InfoError: ``Child`` itself was never marked final, only inherits # the marker from ``Parent``, which is not the class this constructs. - self.assertEqual(Child(1, 2).to_dict(), {'x': 1, 'y': 2}) + self.assertEqual(list(Child(1, 2).to_dict().items(multi=True)), [('x', 1), ('y', 2)]) def test_a_final_class_descending_from_a_base_state_ancestor_is_refused(self) -> None: """``__finalised__`` is read from each class's own ``__dict__`` (#1490). @@ -635,7 +635,7 @@ class Sealed(Info): x: int self.assertEqual([str(record.message) for record in caught], []) - self.assert_silently_finalised(Sealed, lambda cls: cls(1).to_dict(), {'x': 1}) + self.assert_silently_finalised(Sealed, lambda cls: list(cls(1).to_dict().items(multi=True)), [('x', 1)]) def test_schema_final_alone_finalises_silently(self) -> None: """The schema-side baseline. @@ -693,7 +693,7 @@ class TopDown(Info): self.assertEqual([str(record.message) for record in caught], []) for cls in (BottomUp, TopDown): with self.subTest(cls=cls.__name__): - self.assert_silently_finalised(cls, lambda c: c(1).to_dict(), {'x': 1}) + self.assert_silently_finalised(cls, lambda c: list(c(1).to_dict().items(multi=True)), [('x', 1)]) def test_schema_final_and_final_agree_in_either_order(self) -> None: """The schema-side half, where the old failure was *silent*.