From d7c505ffdd61928eaf4fa855f1d8fc2c9634cd2f Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 13:41:12 -0400 Subject: [PATCH] fix(protocols): make() rejects an enum member name given as str unless a namespace is passed Protocol._make_index now infers the namespace for a str member name when none is given: it finds the calling make() method on the class, picks the arguments whose value equals the name, whose *_namespace companion is None, and whose default is an enum member, and uses that default's enum class. If several arguments match, every candidate enum must have the name with the same value; otherwise ProtocolNotImplemented asks for an explicit *_namespace. int, enum members and explicit namespaces take the same path as before. Arguments whose default is not an enum member (None defaults and loop variables over list arguments) keep the old behaviour. Closes #1210 --- pcapkit/protocols/protocol.py | 92 ++++++++++++ tests/protocols/test_make_index_str_unit.py | 150 ++++++++++++++++++++ 2 files changed, 242 insertions(+) create mode 100644 tests/protocols/test_make_index_str_unit.py diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 722ebd229e..1520944bd1 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -49,6 +49,7 @@ if TYPE_CHECKING: from enum import IntEnum as StdlibEnum + from types import FrameType from typing import IO, Any, Callable, DefaultDict, Optional, Type from aenum import IntEnum as AenumEnum @@ -1598,12 +1599,25 @@ def _make_index(cls, name: 'str | int | StdlibEnum | AenumEnum', default: 'Optio ProtocolNotImplemented: If ``name`` is **NOT** in ``namespace`` and ``default`` is :data:`None`. + Note: + If ``name`` is a :obj:`str` and ``namespace`` is :data:`None`, + the namespace is inferred from the calling ``make`` method: it + is the enumeration class of the default value of the argument + that ``name`` was passed as. If several arguments were given the + same name, every one of their enumerations must have the name + and map it to the same value; otherwise + :exc:`~pcapkit.utilities.exceptions.ProtocolNotImplemented` is + raised, asking for an explicit ``*_namespace`` (see + :meth:`_make_index_namespace`). + """ if isinstance(name, (enum.Enum, aenum.Enum)): index = cast('int', name.value) elif isinstance(name, int): index = name else: # name is str + if namespace is None: + namespace = cls._make_index_namespace(name, inspect.currentframe()) try: if isinstance(namespace, type) and issubclass(namespace, (enum.IntEnum, aenum.IntEnum)): index = cast('int', namespace[name].value) @@ -1631,6 +1645,84 @@ def _make_index(cls, name: 'str | int | StdlibEnum | AenumEnum', default: 'Optio return cls._make_pack(index, size=size, signed=signed, lilendian=lilendian) return index + @classmethod + def _make_index_namespace(cls, name: 'str', + frame: 'Optional[FrameType]') -> 'Optional[Type[StdlibEnum] | Type[AenumEnum]]': + """Infer the enumeration namespace of ``name`` for :meth:`_make_index`. + + The caller of :meth:`_make_index` (``frame.f_back``) is looked up + among the methods of ``cls``. An argument ``arg`` of that method + matches if it is accompanied by an ``arg_namespace`` argument that + is :data:`None`, its current value is a :obj:`str` equal to + ``name``, and its default + value is an :class:`~enum.IntEnum` (or :class:`aenum.IntEnum`) + member. The candidate namespace of a match is the class of that + default value. + + Args: + name: member name given to :meth:`_make_index` + frame: frame of :meth:`_make_index` + + Returns: + The candidate namespace if exactly one argument matches. If + several match, any candidate, provided every candidate has + ``name`` as a member and maps it to the same value. + :data:`None` if the caller is not found, nothing matches, or no + candidate has ``name`` as a member. + + Raises: + ProtocolNotImplemented: If several arguments match and their + candidates disagree on ``name``: some lack it, or they map + it to different values. + + """ + caller = frame.f_back if frame is not None else None + if caller is None: + return None + try: + code = caller.f_code + values = caller.f_locals + + func = None + for klass in cls.__mro__: + for attr in vars(klass).values(): + attr = getattr(attr, '__func__', attr) + if getattr(attr, '__code__', None) is code: + func = attr + break + if func is not None: + break + if func is None: + return None + + params = inspect.signature(func).parameters + found = {} # type: dict[str, Type[StdlibEnum] | Type[AenumEnum]] + for arg, param in params.items(): + value = values.get(arg) + if f'{arg}_namespace' not in params or not isinstance(value, str) or value != name: + continue + # an explicit namespace resolves that argument by itself + if values.get(f'{arg}_namespace') is not None: + continue + if isinstance(param.default, (enum.IntEnum, aenum.IntEnum)): + found[arg] = type(param.default) + finally: + del caller, frame + + if len(found) == 1: + return next(iter(found.values())) + + known = {arg: int(space[name]) for arg, space in found.items() if name in space.__members__} + if not known: + return None + if len(known) < len(found) or len(set(known.values())) > 1: + matches = ', '.join(f'{space.__name__}.{name} = {known[arg]}' if arg in known + else f'{space.__name__} has no {name!r}' for arg, space in found.items()) + hint = ' or '.join(f'{arg}_namespace=' for arg in found) + raise ProtocolNotImplemented(f'{cls.__name__}: ambiguous member name {name!r} given to ' + f'{", ".join(found)} ({matches}); pass {hint} explicitly') + return next(iter(found.values())) + @classmethod def _make_data(cls, data: 'Data') -> 'dict[str, Any]': """Create key-value pairs from ``data`` for protocol construction. diff --git a/tests/protocols/test_make_index_str_unit.py b/tests/protocols/test_make_index_str_unit.py new file mode 100644 index 0000000000..cd4beb70d0 --- /dev/null +++ b/tests/protocols/test_make_index_str_unit.py @@ -0,0 +1,150 @@ +# -*- coding: utf-8 -*- +"""``make()`` resolves an enum member name given as :obj:`str`. + +GitHub issue #1210: the ``make()`` signatures declare enum arguments as +``Enum | StdlibEnum | AenumEnum | str | int``, but +:meth:`~pcapkit.protocols.protocol.ProtocolBase._make_index` raised +:exc:`~pcapkit.utilities.exceptions.ProtocolNotImplemented` for a :obj:`str` +unless a ``*_namespace`` was passed. A member name now resolves against the +enumeration class of the argument's default value, and builds the same bytes +as the member itself and as its integer value. + +Every case builds its own packet in memory and reads no capture. Classes are +imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, so they belong to the live +:mod:`pcapkit` import. + +""" + +import importlib +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + + +def _attr(path: 'str') -> 'Any': + module, _, name = path.rpartition('.') + return getattr(importlib.import_module(module), name) + + +def _tcp_alternate_checksum(**kwargs: 'Any') -> 'Any': + option = _attr('pcapkit.const.tcp.option.Option') + return _attr('pcapkit.protocols.transport.tcp.TCP')( + options=[(option.TCP_Alternate_Checksum_Request, kwargs)], + ) + + +#: One ``make()`` per protocol family: +#: (protocol class or builder, argument, enum class, non-default member). +CASES = { + 'link/ethernet': ('pcapkit.protocols.link.ethernet.Ethernet', 'type', + 'pcapkit.const.reg.ethertype.EtherType', 'Internet_Protocol_version_6'), + 'link/l2tpv2': ('pcapkit.protocols.link.l2tpv2.L2TPv2', 'type', + 'pcapkit.const.l2tp.type.Type', 'Control'), + 'internet/ipv4': ('pcapkit.protocols.internet.ipv4.IPv4', 'protocol', + 'pcapkit.const.reg.transtype.TransType', 'TCP'), + 'transport/tcp': (_tcp_alternate_checksum, 'algorithm', + 'pcapkit.const.tcp.checksum.Checksum', 'Checksum_16_bit_Fletcher_s_algorithm'), + 'application/ospf': ('pcapkit.protocols.application.ospf.OSPF', 'type', + 'pcapkit.const.ospf.packet.Packet', 'Database_Description'), + 'misc/pcap': ('pcapkit.protocols.misc.pcap.header.Header', 'network', + 'pcapkit.const.reg.linktype.LinkType', 'RAW'), +} + + +class TestMakeIndexStr(unittest.TestCase): + """Pin member-name resolution in ``_make_index`` across families.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_member_name_matches_member_and_int(self) -> None: + for case, (maker, arg, enum_path, member) in CASES.items(): + with self.subTest(case=case): + build = _attr(maker) if isinstance(maker, str) else maker + value = getattr(_attr(enum_path), member) + + expected = build(**{arg: value}).data + self.assertEqual(build(**{arg: member}).data, expected) + self.assertEqual(build(**{arg: int(value)}).data, expected) + # the member is not the default, so the name was not ignored + self.assertNotEqual(build().data, expected) + + def test_unknown_name_still_raises(self) -> None: + exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') + ethernet = _attr('pcapkit.protocols.link.ethernet.Ethernet') + with self.assertRaises(exc): + ethernet(type='No_Such_EtherType') + + def test_unknown_name_falls_back_to_default(self) -> None: + ethernet = _attr('pcapkit.protocols.link.ethernet.Ethernet') + proto = ethernet(type='No_Such_EtherType', type_default=0x1234) + self.assertEqual(proto.data[12:14], b'\x12\x34') + + def test_explicit_namespace_wins(self) -> None: + ethernet = _attr('pcapkit.protocols.link.ethernet.Ethernet') + # the name is also an EtherType member, but the namespace decides + proto = ethernet(type='Internet_Protocol_version_6', + type_namespace={'Internet_Protocol_version_6': 0x1234}, + type_reversed=True) + self.assertEqual(proto.data[12:14], b'\x12\x34') + + def test_shared_name_with_one_meaning_resolves(self) -> None: + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + tos_del = _attr('pcapkit.const.ipv4.tos_del.ToSDelay') + tos_thr = _attr('pcapkit.const.ipv4.tos_thr.ToSThroughput') + tos_rel = _attr('pcapkit.const.ipv4.tos_rel.ToSReliability') + + expected = ipv4(tos_del=tos_del.NORMAL, tos_thr=tos_thr.NORMAL).data + self.assertEqual(ipv4(tos_del='NORMAL', tos_thr='NORMAL').data, expected) + # an equal but distinct string object, so the match is not by identity + self.assertEqual(ipv4(tos_del='NORMAL', tos_thr=''.join(['NOR', 'MAL'])).data, expected) + + expected = ipv4(tos_thr=tos_thr.HIGH, tos_rel=tos_rel.HIGH).data + self.assertEqual(ipv4(tos_thr='HIGH', tos_rel='HIGH').data, expected) + self.assertNotEqual(ipv4().data, expected) + + def test_shared_name_with_two_meanings_raises(self) -> None: + enum = importlib.import_module('enum') + exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') + protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') + + left = enum.IntEnum('Left', {'Shared': 1}) + right = enum.IntEnum('Right', {'Shared': 2}) + + class Pair(protocol): # type: ignore[misc,valid-type] + def make_pair(self, a: 'Any' = left.Shared, a_namespace: 'Any' = None, + b: 'Any' = right.Shared, b_namespace: 'Any' = None) -> 'int': + return self._make_index(a, namespace=a_namespace) + + with self.assertRaisesRegex(exc, r"ambiguous member name 'Shared'.*" + r"pass a_namespace= or b_namespace= explicitly"): + Pair.make_pair(Pair, a='Shared', b='Shared') + # an explicit namespace on the other argument removes the ambiguity + self.assertEqual(Pair.make_pair(Pair, a='Shared', b='Shared', b_namespace=right), 1) + + # a real make(): ``LOW`` is a ToSDelay member but not a ToSThroughput one + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + with self.assertRaisesRegex(exc, r"ambiguous member name 'LOW'.*" + r"pass tos_del_namespace= or tos_thr_namespace= explicitly"): + ipv4(tos_del='LOW', tos_thr='LOW') + + def test_without_inferable_enum_still_raises(self) -> None: + exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') + protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') + # called outside any ``make()``, so there is no argument to infer from + with self.assertRaises(exc): + protocol._make_index('Internet_Protocol_version_6') + + # ``status`` defaults to ``None``, so its enum cannot be inferred + http = _attr('pcapkit.protocols.application.httpv1.HTTP') + with self.assertRaises(exc): + http(status='OK') + + +if __name__ == '__main__': + unittest.main()