From dcb1306962ea98d727d6d54111a8c111d52f130f Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 14:45:27 -0400 Subject: [PATCH] fix(corekit): from_dict(to_dict()) loses nested values, and Info.__len__ counts hidden keys ``SchemaField.pack`` now rebuilds a mapping into its declared schema, so a schema round-tripped through ``to_dict``/``from_dict`` packs instead of raising ``AttributeError: 'dict' object has no attribute 'pack'`` (#1290). ``Info.from_dict`` does the same for keys whose annotation resolves at runtime to an ``Info`` subclass, and ``to_dict`` now documents that the conversion stays one-way for every other key. ``Info.__len__`` counts only the keys ``__iter__`` yields, so ``len(info) == len(list(info))`` (#1296). Closes #1290 Closes #1296 --- pcapkit/corekit/fields/misc.py | 10 + pcapkit/corekit/infoclass.py | 157 ++++++++- .../corekit/test_from_dict_roundtrip_unit.py | 302 ++++++++++++++++++ tests/corekit/test_infoclass.py | 5 +- 4 files changed, 472 insertions(+), 2 deletions(-) create mode 100644 tests/corekit/test_from_dict_roundtrip_unit.py diff --git a/pcapkit/corekit/fields/misc.py b/pcapkit/corekit/fields/misc.py index d66dcdd68e..e54528ad73 100644 --- a/pcapkit/corekit/fields/misc.py +++ b/pcapkit/corekit/fields/misc.py @@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, TypeVar, cast from pcapkit.corekit.fields.field import NO_VALUE, FieldBase +from pcapkit.utilities.compat import Mapping from pcapkit.utilities.exceptions import FieldError, NoDefaultValue from pcapkit.utilities.warnings import RegistryWarning, warn @@ -717,6 +718,15 @@ def pack(self, value: 'Optional[_TS | bytes]', packet: 'dict[str, Any]') -> 'byt if isinstance(value, bytes): return value + # NOTE: :meth:`Schema.to_dict ` + # flattens a nested schema into a plain mapping, so a schema rebuilt + # through ``from_dict`` hands that mapping back here; turn it into the + # declared schema before packing it. + from pcapkit.protocols.schema.schema import \ + Schema # pylint: disable=import-outside-top-level + if isinstance(value, Mapping) and not isinstance(value, Schema): + value = self._schema.from_dict(value) + packet.update(self._packet) return value.pack(nested_packet_context(packet)) diff --git a/pcapkit/corekit/infoclass.py b/pcapkit/corekit/infoclass.py index cfa5010af3..7a2ca99b2d 100644 --- a/pcapkit/corekit/infoclass.py +++ b/pcapkit/corekit/infoclass.py @@ -14,6 +14,9 @@ import collections.abc import enum import itertools +import sys +import types +import typing from typing import TYPE_CHECKING, Generic, TypeVar from pcapkit.corekit.enum import EnumLookup @@ -507,7 +510,9 @@ def __repr__(self) -> 'str': return f'<{type(self).__name__} {args}>' def __len__(self) -> 'int': - return len(self.__dict__) + # NOTE: count exactly the keys :meth:`__iter__` yields, so the + # bookkeeping attributes in ``__excluded__`` are not counted. + return sum(1 for key in self.__dict__ if key not in self.__excluded__) def __iter__(self) -> 'Iterator[str]': for key in self.__dict__: @@ -537,6 +542,19 @@ def from_dict(cls, dict_: 'Optional[Mapping[str, VT] | Iterable[tuple[str, VT]]] * If ``dict_`` is not present, then does: ``for k, v in kwargs.items(): self[k] = v``. + 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 + rebuilt into that class, so ``cls.from_dict(info.to_dict())`` restores + such nested values. A key without such an annotation keeps the mapping + as is -- see :meth:`to_dict`. + + 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 + :meth:`from_dict` would. Validating it here would reject data that + ``Info(**kwargs)`` has always accepted, so the shape of the mapping + stays the caller's responsibility. + Args: dict\_: Source data. **kwargs: Arbitrary keyword arguments. @@ -544,6 +562,12 @@ 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(): + 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) return self def to_dict(self) -> 'dict[str, VT]': @@ -556,6 +580,14 @@ def to_dict(self) -> 'dict[str, VT]': types, such as :obj:`list`, :obj:`tuple`, :obj:`set`, etc., we shall not convert them into :obj:`dict` and remain them intact. + Note: + :meth:`from_dict` reverses the flattening only for keys whose type + 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`. + """ dict_ = {} # type: dict[str, Any] for (key, value) in self.__dict__.items(): @@ -578,3 +610,126 @@ def to_dict(self) -> 'dict[str, VT]': else: dict_[out_key] = value return dict_ + + +#: Name of the per-class attribute :func:`_nested_info_types` caches its answer in. +_NESTED_INFO_CACHE = '__nested_info_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. +_UNION_TYPE = getattr(types, 'UnionType', None) + +#: Type of :data:`None`, i.e. the member ``Optional`` adds to a union. +_NONE_TYPE = type(None) + + +def _as_info_type(hint: 'Any') -> 'Optional[Type[Info]]': + """Return the :class:`Info` subclass a resolved annotation names, if any. + + :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. + + Args: + hint: Resolved type annotation. + + Returns: + The :class:`Info` subclass the annotation names, or :data:`None`. + + Note: + Both guards below are load-bearing on Python 3.9 and 3.10, where a + :class:`types.GenericAlias` such as ``list[Packet]`` satisfies + ``isinstance(hint, type)`` while :func:`issubclass` still refuses it + with :exc:`TypeError` -- which would otherwise escape + :meth:`Info.from_dict` for the 26 of 487 :class:`Info` classes that + carry such an annotation. From 3.11 on the ``isinstance`` is already + :data:`False`, so only the older interpreters reach the + :exc:`TypeError`; ``NestedTypeResolverTests`` in + :file:`tests/corekit/test_from_dict_roundtrip_unit.py` pins the shape + on every version by handing this function an object that mimics it. + + """ + origin = typing.get_origin(hint) + if origin is not None or not isinstance(hint, type): + # NOTE: a union of exactly one non-``None`` member is ``Optional`` of + # that member, so it unwraps to it; anything else is not an ``Info``. + args = [arg for arg in getattr(hint, '__args__', ()) if arg is not _NONE_TYPE] + is_union = origin is typing.Union or (_UNION_TYPE is not None and origin is _UNION_TYPE) + hint = args[0] if is_union and len(args) == 1 else None + + if not isinstance(hint, type): + return None + + try: + return hint if issubclass(hint, Info) else None + except TypeError: + return None + + +def _nested_info_types(cls: 'Type[Info]') -> 'dict[str, Type[Info]]': + """Map the keys of ``cls`` annotated with an :class:`Info` subclass to 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 + than spoiling the rest. + + Args: + cls: Info class. + + Returns: + Mapping of key names to the :class:`Info` subclass annotated for them. + + Note: + The answer is cached in ``cls``'s own ``__dict__``, read back from + there rather than through :func:`getattr` so a subclass never inherits + its parent's map. The cache is deliberately kept on the class rather + than in a module-level :class:`weakref.WeakKeyDictionary`: such a map + would pin every self-referential class forever, since the cached value + strongly references the class the weak key points at, whereas the + reference cycle a class attribute creates is collectable. + + An unresolvable annotation is cached as "skip" alongside the rest, so a + forward reference whose target is defined only *after* the first + :meth:`Info.from_dict` call stays one-way for the life of the process. + That is accepted rather than fixed: :meth:`Info.from_dict` runs per + parsed packet (see :meth:`IPv4._read_ipv4_options + `), and + re-resolving the annotations on every call to catch a case that does + not arise in this package costs far more than it saves. + + """ + cached = cls.__dict__.get(_NESTED_INFO_CACHE) + if cached is not None: + return cached + + nested = {} # type: dict[str, Type[Info]] + seen = set() # type: set[str] + for cls_ in cls.mro(): # pragma: no branch + # NOTE: same walk as :func:`info_final`, which builds ``__init__`` + # from these annotations; the most derived declaration of a key wins. + if cls_ is Info: + break + + module = sys.modules.get(cls_.__module__) + globalns = dict(vars(module)) if module is not None else {} + localns = dict(vars(cls_)) + for (key, ann) in cls_.__dict__.get('__annotations__', getattr(cls_, '__annotations__', {})).items(): + if key in seen: + continue + seen.add(key) + + try: + hint = typing.get_type_hints(types.SimpleNamespace(__annotations__={key: ann}), + globalns, localns)[key] + except Exception: # pylint: disable=broad-except + continue + + info_cls = _as_info_type(hint) + if info_cls is not None: + nested[key] = info_cls + + setattr(cls, _NESTED_INFO_CACHE, nested) + return nested diff --git a/tests/corekit/test_from_dict_roundtrip_unit.py b/tests/corekit/test_from_dict_roundtrip_unit.py new file mode 100644 index 0000000000..4abbc04dc1 --- /dev/null +++ b/tests/corekit/test_from_dict_roundtrip_unit.py @@ -0,0 +1,302 @@ +# -*- coding: utf-8 -*- +"""``from_dict(to_dict())`` rebuilds nested values, and ``len`` matches iteration. + +GitHub issue #1290: :meth:`Schema.to_dict +` flattens a nested schema +into a :obj:`dict`, and :meth:`SchemaField.pack +` then called ``.pack`` on that +dict. :meth:`Info.from_dict ` kept +the flattened dict too, so ``info.inner.x`` raised. + +GitHub issue #1296: :meth:`Info.__len__ +` counted the ``__excluded__`` +bookkeeping keys that iteration skips. + +The PCAP-NG cases build their blocks as octets in memory rather than reading a +capture: :file:`examples/captures/` is generated by +:file:`examples/generators/make_samples.py` and is not tracked, so the unit +tier has no captures to read. + +Every class is declared inside its test, after +:func:`~tests._support.reimport_once_per_class`, so it binds the live +:mod:`pcapkit` import rather than one an earlier module has purged. + +""" + +import gc +import unittest + +from tests._support import reimport_once_per_class + +#: PCAP-NG blocks, little-endian, one per block schema that claims a type code. +#: Each nests its body in :attr:`PCAPNG.block +#: `, which is the +#: :class:`~pcapkit.corekit.fields.misc.SchemaField` issue #1290 is about. +PCAPNG_BLOCKS = { + 'shb': '0a0d0d0a' '1c000000' '4d3c2b1a' '0100' '0000' 'ffffffffffffffff' '1c000000', + 'shb-options': ('0a0d0d0a' '28000000' '4d3c2b1a' '0100' '0000' 'ffffffffffffffff' + '0100' '0400' '61626364' '0000' '0000' '28000000'), + 'idb': '01000000' '14000000' '0100' '0000' '00000400' '14000000', + 'idb-options': ('01000000' '20000000' '0100' '0000' '00000400' + '0900' '0100' '06000000' '0000' '0000' '20000000'), + 'epb': ('06000000' '24000000' '00000000' '00000000' '0a000000' + '04000000' '04000000' '61626364' '24000000'), + 'spb': '03000000' '14000000' '04000000' '61626364' '14000000', + 'nrb': '04000000' '10000000' '0000' '0000' '10000000', + 'isb': '05000000' '18000000' '00000000' '00000000' '0a000000' '18000000', + 'dsb': '0a000000' '18000000' '544c534b' '04000000' '61626364' '18000000', + 'cb': 'ad0b0000' '14000000' 'ad0b0000' '61626364' '14000000', + 'sjeb': '09000000' '10000000' '61626364' '10000000', + 'unknown-type': 'ff7f0000' '10000000' '61626364' '10000000', +} + + +class TestSchemaFromDictRoundTrip(unittest.TestCase): + """GitHub issue #1290, the :class:`Schema` half.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _schemas(self): # type: ignore[no-untyped-def] + from pcapkit.corekit.fields.misc import SchemaField + from pcapkit.corekit.fields.numbers import UInt8Field + from pcapkit.protocols.schema.schema import Schema, schema_final + + @schema_final + class In(Schema): + x: int = UInt8Field() + + @schema_final + class Out(Schema): + inner: In = SchemaField(length=1, schema=In) + tail: int = UInt8Field() + + return In, Out + + def test_from_dict_of_to_dict_packs_nested_schema(self) -> None: + _, Out = self._schemas() + raw = b'\x07\x09' + parsed = Out.unpack(raw, len(raw), None) + self.assertEqual(Out.from_dict(parsed.to_dict()).pack(), raw) + + def test_schema_value_still_packs_as_is(self) -> None: + In, Out = self._schemas() + self.assertEqual(Out.from_dict(inner=In(x=3), tail=4).pack(), b'\x03\x04') + + def test_pcapng_block_schemas_round_trip(self) -> None: + from pcapkit.protocols.schema.misc.pcapng import PCAPNG + + for name, hexed in PCAPNG_BLOCKS.items(): + raw = bytes.fromhex(hexed) + with self.subTest(block=name): + schema = PCAPNG.unpack(raw, len(raw), None) + self.assertEqual(schema.pack(), raw) + self.assertEqual(type(schema).from_dict(schema.to_dict()).pack(), raw) + + +class TestInfoFromDictRoundTrip(unittest.TestCase): + """GitHub issue #1290, the :class:`Info` half.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_annotated_nested_info_is_rebuilt(self) -> None: + from typing import Optional + + from pcapkit.corekit.infoclass import Info, info_final + + @info_final + class In(Info): + x: int + + @info_final + class Out(Info): + inner: In + maybe: Optional[In] + y: int + + info = Out(inner=In(x=1), maybe=In(x=2), y=3) + rebuilt = Out.from_dict(info.to_dict()) + 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()) + + def test_string_annotation_in_a_data_module_is_resolved(self) -> None: + from pcapkit.protocols.data.application.httpv2 import DataFrame, DataFrameFlags + + frame = DataFrame.from_dict({'flags': {'END_STREAM': True, 'PADDED': False}}) + self.assertIsInstance(frame.flags, DataFrameFlags) + self.assertTrue(frame.flags.END_STREAM) + + def test_unannotated_or_unresolvable_key_stays_a_dict(self) -> None: + from pcapkit.corekit.infoclass import Info, info_final + + @info_final + class Out(Info): + inner: 'NoSuchType' # type: ignore[name-defined] # noqa: F821 + + info = Out.from_dict({'inner': {'x': 1}, 'extra': {'z': 2}}) + self.assertEqual(info.inner, {'x': 1}) + self.assertEqual(info.extra, {'z': 2}) + + def test_container_annotation_does_not_raise(self) -> None: + """A ``list[Info]`` annotation is skipped, not fed to :func:`issubclass`. + + Below Python 3.11 such an annotation satisfies ``isinstance(hint, + type)``, so an unguarded :func:`issubclass` raised :exc:`TypeError` out + of :meth:`~pcapkit.corekit.infoclass.Info.from_dict` -- measured on + 3.10 for 26 of 487 :class:`~pcapkit.corekit.infoclass.Info` classes, + including :class:`~pcapkit.foundation.reassembly.data.ip.Packet`. + + """ + from typing import Dict, List + + from pcapkit.corekit.infoclass import Info, info_final + + @info_final + class In(Info): + x: int + + @info_final + class Out(Info): + entries: List[In] + mapped: Dict[str, In] + builtin: 'list[In]' + + info = Out.from_dict({'entries': [{'x': 1}], 'mapped': {'a': {'x': 2}}, + 'builtin': [In(x=3)]}) + self.assertEqual(info.entries, [{'x': 1}]) + self.assertEqual(info.mapped, {'a': {'x': 2}}) + self.assertIsInstance(info.builtin[0], In) + + def test_reassembly_packet_from_dict_still_works(self) -> None: + from pcapkit.foundation.reassembly.data.ip import Packet + + self.assertEqual(Packet.from_dict({'bufid': 1}).bufid, 1) + + +class NestedTypeResolverTests(unittest.TestCase): + """The annotation resolver behind :meth:`Info.from_dict`.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_a_generic_alias_shape_is_skipped_rather_than_raising(self) -> None: + """Pin the Python 3.9/3.10 :class:`types.GenericAlias` shape. + + There, ``isinstance(list[In], type)`` is :data:`True` while + :func:`issubclass` refuses the same object with :exc:`TypeError`. From + 3.11 on the ``isinstance`` is already :data:`False`, so the object below + mimics the older shape -- its ``__class__`` reports :class:`type`, which + is what makes :func:`isinstance` agree and :func:`issubclass` object. + + """ + from pcapkit.corekit.infoclass import _as_info_type + + class FakeGenericAlias: + """An object :func:`isinstance` reads as a class and :func:`issubclass` does not.""" + + @property + def __class__(self): # type: ignore[override] + return type + + alias = FakeGenericAlias() + self.assertIsInstance(alias, type) + with self.assertRaises(TypeError): + issubclass(alias, int) # type: ignore[arg-type] + self.assertIsNone(_as_info_type(alias)) + + def test_resolver_answers(self) -> None: + from typing import List, Optional, Union + + from pcapkit.corekit.infoclass import Info, _as_info_type, info_final + + @info_final + class In(Info): + x: int + + self.assertIs(_as_info_type(In), In) + self.assertIs(_as_info_type(Optional[In]), In) + for hint in (int, None, 'In', List[In], Union[In, int]): + with self.subTest(hint=hint): + self.assertIsNone(_as_info_type(hint)) + + def test_cache_is_per_class_and_collectable(self) -> None: + """A self-referential class is still collected. + + The map is cached on the class, not in a module-level + :class:`weakref.WeakKeyDictionary`: there the cached value strongly + references the class the weak key points at, so a class annotated with + itself could never be collected. + + """ + import weakref + + from pcapkit.corekit.infoclass import Info, _nested_info_types, info_final + + @info_final + class Node(Info): + nxt: int + + # NOTE: the annotation is rebound to the class object itself, because a + # locally declared class is not resolvable by name from its module's + # globals, and ``Optional[Node]`` would be pinned by :mod:`typing`'s own + # argument cache and so never collected whatever this code does. + Node.__annotations__['nxt'] = Node + + self.assertIn('nxt', _nested_info_types(Node)) + self.assertIn('__nested_info_types__', Node.__dict__) + + ref = weakref.ref(Node) + del Node + gc.collect() + self.assertIsNone(ref()) + + def test_a_subclass_does_not_inherit_its_parents_map(self) -> None: + from pcapkit.corekit.infoclass import Info, _nested_info_types, info_final + + @info_final + class In(Info): + x: int + + class Base(Info): + inner: In + + @info_final + class Derived(Base): + other: int + + self.assertEqual(_nested_info_types(Base), {'inner': In}) + self.assertEqual(_nested_info_types(Derived), {'inner': In}) + self.assertNotIn('__nested_info_types__', Base.__dict__.get('__annotations__', {})) + + +class TestInfoLen(unittest.TestCase): + """GitHub issue #1296.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_len_matches_iteration(self) -> None: + from pcapkit.corekit.infoclass import Info, info_final + + @info_final + class I(Info): + a: int + + info = I(a=1) + self.assertEqual(list(info), ['a']) + self.assertEqual(len(info), 1) + self.assertEqual(len(Info()), 0) + + def test_len_counts_renamed_builtin_keys(self) -> None: + from pcapkit.corekit.infoclass import Info + + info = Info.from_dict({'keys': 1, 'b': 2}) + self.assertEqual(len(info), len(list(info))) + self.assertEqual(len(info), 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/corekit/test_infoclass.py b/tests/corekit/test_infoclass.py index 8825e7ebdd..77801c6e1c 100644 --- a/tests/corekit/test_infoclass.py +++ b/tests/corekit/test_infoclass.py @@ -59,7 +59,10 @@ class Bag(Info): bag = Bag({'items': 'conflict', 'visible': 1, 'hidden': 2}, extra=3) self.assertEqual(bag['items'], 'conflict') self.assertEqual(list(bag), ['items', 'visible', 'extra']) - self.assertEqual(len(bag), 6) + # NOTE: GitHub issue #1296 -- ``len`` counts exactly the keys + # iteration yields, so the ``__excluded__`` ``hidden`` key and the + # internal bookkeeping attributes are not counted. + 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})