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()