diff --git a/docs/source/pcapkit/corekit/multidict.rst b/docs/source/pcapkit/corekit/multidict.rst index 67ae8aab1..ea18f2e6b 100644 --- a/docs/source/pcapkit/corekit/multidict.rst +++ b/docs/source/pcapkit/corekit/multidict.rst @@ -35,6 +35,15 @@ 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/corekit/infoclass.py b/pcapkit/corekit/infoclass.py index 7a2ca99b2..1ad0629e3 100644 --- a/pcapkit/corekit/infoclass.py +++ b/pcapkit/corekit/infoclass.py @@ -20,12 +20,13 @@ from typing import TYPE_CHECKING, Generic, TypeVar from pcapkit.corekit.enum import EnumLookup +from pcapkit.corekit.multidict import InfoDict, MultiDict from pcapkit.utilities.compat import Mapping, final from pcapkit.utilities.exceptions import InfoError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import InfoWarning, warn if TYPE_CHECKING: - from typing import Any, Iterable, Iterator, NoReturn, Optional, Type + from typing import Any, ItemsView, Iterable, Iterator, NoReturn, Optional, Type from typing_extensions import Self @@ -137,7 +138,7 @@ def info_final(cls: 'ST', *, _finalised: 'bool' = True) -> 'ST': if cls is Info and cls.__dict__.get('__base_ready__'): return cls - temp = ['__map__', '__map_reverse__', '__builtin__', '__finalised__'] + temp = ['__map__', '__map_reverse__', '__multi__', '__builtin__', '__finalised__'] temp.extend(cls.__additional__) for obj in cls.mro(): temp.extend(dir(obj)) @@ -306,6 +307,10 @@ class Info(Mapping[str, VT], Generic[VT], metaclass=InfoMeta): #: Mapping of name conflicts with builtin methods (transformed names to #: original names). __map_reverse__: 'dict[str, str]' + #: Values of keys held more than once, i.e. every value but the first, + #: keyed by the (transformed) key that was last in :attr:`__dict__` when + #: each was added, so :meth:`to_dict` writes it right after that key. + __multi__: 'dict[str, list[tuple[str, Any]]]' #: List of builtin methods. __builtin__: 'set[str]' @@ -439,9 +444,20 @@ def __new__(cls, *args: 'VT', **kwargs: 'VT') -> 'Self': # pylint: disable=unus # and thus being shared by all instances. super().__setattr__(self, '__map__', {}) super().__setattr__(self, '__map_reverse__', {}) + super().__setattr__(self, '__multi__', {}) return self + def __copy__(self) -> 'Self': + # NOTE: The bookkeeping mappings are per instance, so a copy gets its + # own: a later :meth:`__update__` on it must not reach the original. + new = type(self).__new__(type(self)) + new.__dict__.update(self.__dict__) + new.__dict__['__map__'] = dict(self.__map__) + new.__dict__['__map_reverse__'] = dict(self.__map_reverse__) + new.__dict__['__multi__'] = {key: list(values) for (key, values) in self.__multi__.items()} + return new + def __post_init__(self) -> 'None': """Customisation method to be called after initialisation.""" @@ -452,16 +468,39 @@ def __update__(self, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT] # implicitly and internally. Such mapping information will be stored # within the :attr:`__map__` attribute. + # NOTE: A :class:`MultiDict` *adds* its values to the ones already held, + # as :meth:`MultiDict.update` does: a key already present keeps its + # value and records the new one in :attr:`__multi__`. Anything else + # replaces the values held. + __name__ = type(self).__name__ # pylint: disable=redefined-builtin + multi_iter = () # type: Iterable[tuple[str, Any]] if dict_ is None: data_iter = kwargs.items() # type: Iterable[tuple[str, Any]] + elif isinstance(dict_, MultiDict): + multi_iter = dict_.items(multi=True) + data_iter = kwargs.items() elif isinstance(dict_, (dict, collections.abc.Mapping)) or hasattr(dict_, 'items'): data_iter = itertools.chain(dict_.items(), kwargs.items()) else: data_iter = itertools.chain(dict_, kwargs.items()) + for (key, value) in multi_iter: + 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)) + else: + self.__update__({key: value}) + for (key, value) in data_iter: + if self.__multi__: + # NOTE: A key replaced drops the values it held besides the first. + new_key = f'_{__name__}{key}' if key in self.__builtin__ else key + for values in self.__multi__.values(): + values[:] = [item for item in values if item[0] != new_key] + if key in self.__builtin__: new_key = f'_{__name__}{key}' @@ -524,6 +563,27 @@ def __getitem__(self, name: 'str') -> 'VT': key = self.__map__.get(name, name) return self.__dict__[key] + def items(self, multi: 'bool' = False) -> 'ItemsView[str, VT] | Iterator[tuple[str, VT]]': # type: ignore[override] + """Return the ``(key, value)`` pairs. + + Args: + multi: If set to :obj:`True`, an iterator with a pair for each value + of each key, as :meth:`to_dict` writes them. Otherwise a view of + the pairs of the first value of each key, as + :meth:`Mapping.items ` returns. + + """ + if not multi: + return super().items() + return ((self.__map_reverse__.get(key, key), value) for (key, value) in self.__items_multi()) + + def __items_multi(self) -> 'Iterator[tuple[str, VT]]': + """Every ``(key, value)`` pair, with the transformed keys, in order.""" + for (key, value) in self.__dict__.items(): + if key not in self.__excluded__: + yield key, value + yield from self.__multi__.get(key, ()) + def __setattr__(self, name: 'str', value: 'VT') -> 'NoReturn': raise UnsupportedCall("can't set attribute") @@ -541,6 +601,10 @@ def from_dict(cls, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT]]] ``for k, v in dict_: self[k] = v``. * If ``dict_`` is not present, then does: ``for k, v in kwargs.items(): self[k] = v``. + * If ``dict_`` is a :class:`~pcapkit.corekit.multidict.MultiDict`, a + key it holds more than once keeps every value, as :meth:`to_dict` + returns them: the first is the value of the key, and + ``items(multi=True)`` yields all of them. A mapping given for a key whose type annotation resolves at runtime to an :class:`Info` subclass (or to :data:`~typing.Optional` of one) is @@ -568,11 +632,21 @@ def from_dict(cls, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT]]] value = self.__dict__.get(key) if isinstance(value, Mapping) and not isinstance(value, Info): self.__dict__[key] = info_cls.from_dict(value) + 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)) return self - def to_dict(self) -> 'dict[str, VT]': + def to_dict(self) -> 'InfoDict[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 + :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. + Important: We only convert nested :class:`Info` objects into :obj:`dict` if they are the direct value of the :class:`Info` object's attribute. @@ -585,18 +659,15 @@ def to_dict(self) -> 'dict[str, VT]': annotation resolves at runtime to an :class:`Info` subclass. For 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 a - plain :obj:`dict`. + the conversion is one-way, and the nested value comes back as the + :class:`~pcapkit.corekit.multidict.InfoDict` it was written as. """ - dict_ = {} # type: dict[str, Any] - for (key, value) in self.__dict__.items(): - if key in self.__excluded__: - continue - + dict_ = InfoDict() # type: InfoDict[str, Any] + for (key, value) in self.__items_multi(): out_key = self.__map_reverse__.get(key, key) if isinstance(value, Info): - dict_[out_key] = value.to_dict() + dict_.add(out_key, value.to_dict()) #elif isinstance(value, (tuple, list, set, frozenset)): # temp = [] # type: list[Any] @@ -608,7 +679,7 @@ def to_dict(self) -> 'dict[str, VT]': # dict_[out_key] = value.__class__(temp) else: - dict_[out_key] = value + dict_.add(out_key, value) return dict_ diff --git a/pcapkit/corekit/multidict.py b/pcapkit/corekit/multidict.py index 464c58b97..1ac586f7e 100644 --- a/pcapkit/corekit/multidict.py +++ b/pcapkit/corekit/multidict.py @@ -12,17 +12,19 @@ """ +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, Iterable, Iterator, Mapping, NoReturn, Optional, SupportsIndex + from typing import (Any, ItemsView, Iterable, Iterator, KeysView, Mapping, NoReturn, Optional, + SupportsIndex, ValuesView) from typing_extensions import Literal -__all__ = ['MultiDict', 'OrderedMultiDict'] +__all__ = ['MultiDict', 'OrderedMultiDict', 'InfoDict'] ############################################################################### # Type variables @@ -636,3 +638,73 @@ 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 95edee840..4833c4fb1 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 MultiDict, OrderedMultiDict +from pcapkit.corekit.multidict import InfoDict, MultiDict, OrderedMultiDict from pcapkit.corekit.sentinels import NoValueType, NullType from pcapkit.protocols.schema.schema import Schema from pcapkit.utilities.exceptions import UnsupportedCall @@ -427,7 +427,10 @@ 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. + 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. """ if isinstance(o, decimal.Decimal): @@ -445,7 +448,7 @@ def object_hook(self, o: 'Any') -> 'Any': result = '' elif isinstance(o, NoValueType): result = '' - elif isinstance(o, OrderedMultiDict): + elif isinstance(o, OrderedMultiDict) and not isinstance(o, InfoDict): # 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 @@ -456,7 +459,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): + elif isinstance(o, MultiDict) and not isinstance(o, InfoDict): # 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/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 9241e4171..475906526 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -31,7 +31,8 @@ from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader from pcapkit.const.reg.transtype import TransType as Enum_TransType -from pcapkit.corekit.multidict import OrderedMultiDict +from pcapkit.corekit.infoclass import Info +from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict from pcapkit.corekit.protochain import ProtoChain from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 from pcapkit.protocols.internet.ip import IP @@ -407,15 +408,18 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl The keys the extension headers are under, and the chain in the shape of ``__exthdr__``, i.e. the parser class and info of each. - Note: - A header repeated in the chain is written under the key of its - predecessor, so only the last of them is left in ``data``. + A header repeated in the chain is held under the same key as its + predecessor, and ``data.items(multi=True)`` yields each of them in turn. """ names = [] # type: list[str] chain = [] # type: list[tuple[Type[ProtocolBase], Any]] proto = data.get('next') - for key in cls._lookup_nested(data): + nested = cls._lookup_nested(data) + pairs = data.items(multi=True) if isinstance(data, (Info, MultiDict)) else data.items() + for (key, info) in pairs: + if key not in nested: + continue try: ex_proto = Enum_ExtensionHeader(proto) except ValueError: @@ -423,7 +427,6 @@ def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tupl # NOTE: As in :meth:`_decode_next_layer`, a code with no parser # carrying ``next`` ends the chain, and so does a fragment header. - info = data[key] if 'next' not in info: break names.append(key) @@ -521,10 +524,12 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, if not hasattr(info, 'next'): break + # NOTE: Added through a :class:`MultiDict`, so a repeated header + # keeps its predecessor's info rather than overwriting it, and + # :meth:`Info.to_dict ` + # writes every one of them, in wire order (:issue:`1453`). name = next_.alias.lstrip('IPv6-').lower() - ipv6.__update__({ - name: info, - }) + ipv6.__update__(OrderedMultiDict([(name, info)])) # record protocol name # self._protos = ProtoChain(name, chain, alias) @@ -561,9 +566,8 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # update next header 'protocol': proto, - # extension header chain in wire order, for ``from_data``; the - # per-header keys above cannot carry it, since a repeated header - # overwrites its predecessor's key + # extension header chain in wire order, with the parser of each, + # for ``from_data`` '__exthdr__': tuple(_exthdr), }) diff --git a/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py b/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py new file mode 100644 index 000000000..5a5e4eadc --- /dev/null +++ b/tests/protocols/test_to_dict_repeated_exthdr_1453_unit.py @@ -0,0 +1,220 @@ +# -*- coding: utf-8 -*- +"""``from_data(info.to_dict())`` keeps a repeated IPv6 extension header. + +GitHub issue #1453: :meth:`Info.to_dict ` +keyed the extension headers by name, so a second header of the same type +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. + +Every case builds its own octets and reads no capture. Classes are imported +inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import copy +import io +import json +import os +import pickle # nosec: B403 -- round-trips objects built here +import struct +import tempfile +import unittest +import warnings + +from tests._support import reimport_once_per_class + +HOP, DST, ROUTE, UDP = 0, 60, 43, 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') + + +def exthdr(code: int, nxt: int, mark: int) -> bytes: + """An 8-octet extension header of type ``code``; ``mark`` tells repeats apart.""" + if code == ROUTE: + # An unassigned routing type, with ``mark`` in its type-specific data. + return bytes([nxt, 0, 253, 0, 0, 0, 0, mark]) + # Options: a PadN carrying ``mark``. + return bytes([nxt, 0, 1, 4, 0, 0, 0, mark]) + + +def ipv6(codes: 'list[int]') -> bytes: + """An IPv6 packet carrying the extension headers ``codes``, then UDP.""" + body = b'' + for index, code in enumerate(codes): + body += exthdr(code, codes[index + 1] if index + 1 < len(codes) else UDP, index) + body += struct.pack('!HHHH', 40000, 40000, 12, 0) + b'ping' + return struct.pack('!IHBB16s16s', 6 << 28, len(body), codes[0], 64, SRC6, DST6) + body + + +#: Chains repeating a header type, back to back or not. +CHAINS = { + 'dst-dst': [DST, DST], + 'hop-dst-dst': [HOP, DST, DST], + 'dst-dst-dst': [DST, DST, DST], + 'route-route': [ROUTE, ROUTE], + 'dst-route-dst': [DST, ROUTE, DST], + 'hop-dst-route-dst': [HOP, DST, ROUTE, DST], +} + + +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 + + if isinstance(value, InfoDict): + return {key: plain(val) for key, val in value.items()} + return value + + +class TestToDictRepeatedExtensionHeader(unittest.TestCase): + """Pin the dict rebuild of a repeated extension header chain.""" + + 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 frame(self, name: str): # type: ignore[no-untyped-def] + """Parse ``CHAINS[name]`` over Ethernet as the first record of a PCAP file.""" + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcap.header import Header + + packet = MAC + b'\x86\xdd' + ipv6(CHAINS[name]) + record = struct.pack(' None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.pcap.frame import Frame + + for name in CHAINS: + frame = self.frame(name) + with self.subTest(chain=name, layer='IPv6'): + ip6 = frame[IPv6] + self.assertEqual(len(ip6.data), 40 + 8 * len(CHAINS[name]) + 12) + self.assertEqual(IPv6.from_data(ip6.info.to_dict()).data, ip6.data) + with self.subTest(chain=name, layer='Ethernet'): + eth = frame[Ethernet] + self.assertEqual(Ethernet.from_data(eth.info.to_dict()).data, eth.data) + with self.subTest(chain=name, layer='Frame'): + rebuilt = Frame.from_data(frame.info.to_dict(), num=1, header=self.header) + self.assertEqual(rebuilt.data, frame.data) + + def test_rebuilt_chain_has_the_parsed_headers(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + for name in CHAINS: + with self.subTest(chain=name): + ip6 = self.frame(name)[IPv6] + rebuilt = IPv6.from_data(ip6.info.to_dict()) + self.assertEqual([ext.data for ext in rebuilt.extension_headers.values()], + [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.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) + + 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] + self.assertEqual(held, parsed) + + # The dict interface, and so every dump, 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)) + + def test_info_from_dict_keeps_the_repeats(self) -> None: + from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 + from pcapkit.protocols.internet.ipv6 import IPv6 + + for name in CHAINS: + with self.subTest(chain=name): + info = self.frame(name)[IPv6].info + dict_ = info.to_dict() + 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(rebuilt, info) + + def test_info_dict_behaves_like_a_dict(self) -> None: + from pcapkit.corekit.infoclass import Info + from pcapkit.corekit.multidict import InfoDict, 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) + + 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}) + 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.assertEqual(list(clone.items(multi=True)), [('a', 1), ('b', 2), ('a', 3)]) + + def test_info_copy_keeps_its_own_repeats(self) -> None: + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.internet.ipv6 import IPv6 + + info = self.frame('dst-dst')[IPv6].info + before = list(info.items(multi=True)) + extra = info['opts'] + for clone in (copy.copy(info), copy.deepcopy(info)): + with self.subTest(clone=type(clone).__name__): + self.assertIsNot(clone.__multi__, info.__multi__) + self.assertIsNot(clone.__map__, info.__map__) + clone.__update__(OrderedMultiDict([('opts', extra)])) + 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: + import dictdumper.json + + from pcapkit.dumpkit.common import make_dumper + + dict_ = self.frame('dst-route-dst').info.to_dict() + outputs = [] + with tempfile.TemporaryDirectory() as tmp: + for index, value in enumerate((dict_, plain(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]) + + +if __name__ == '__main__': + unittest.main()