Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 92 additions & 0 deletions pcapkit/protocols/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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.
Expand Down
150 changes: 150 additions & 0 deletions tests/protocols/test_make_index_str_unit.py
Original file line number Diff line number Diff line change
@@ -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()
Loading