From f345fe84ee8fd5244a6ac36946c3c9197be4cd3b Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 16:05:36 -0400 Subject: [PATCH] fix(protocol): raw items in a make() list build a packet; str enum names resolve per argument - #1312: unpack() read the schema make() returned, so raw bytes items (options, parameters, chunks, MH data) reached read() and crashed with AttributeError. When make() was given raw or Schema items, a made schema holding raw bytes where parsing its packed octets yields a schema is replaced by the parsed one; octets that do not parse raise ProtocolError. A bare MH() now constructs. - #1308: _make_index_namespace() locates the _make_index call in the caller's source and resolves the name against that call's own argument, so two arguments holding the same member name no longer raise "ambiguous member name". The all-arguments check remains the fallback when the source is unavailable or the first argument is not a parameter. test_make_index_str_unit pinned the old ambiguous-name raise and is updated; new tests in tests/protocols/test_protocol_make_unit.py. --- pcapkit/protocols/protocol.py | 183 ++++++++++++++- tests/protocols/test_make_index_str_unit.py | 14 +- tests/protocols/test_protocol_make_unit.py | 236 ++++++++++++++++++++ 3 files changed, 413 insertions(+), 20 deletions(-) create mode 100644 tests/protocols/test_protocol_make_unit.py diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 1520944bd1..fd04e0d45b 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -12,6 +12,7 @@ """ import abc +import ast import collections import contextlib import difflib @@ -30,6 +31,7 @@ import aenum from pcapkit.corekit.context import ContextRegistry +from pcapkit.corekit.fields.misc import ConditionalField, SchemaField, SwitchField from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.protochain import ProtoChain from pcapkit.corekit.sentinels import ABSENT, AbsentType # pylint: disable=unused-import @@ -43,8 +45,9 @@ from pcapkit.utilities.chardet import detect from pcapkit.utilities.compat import cached_property from pcapkit.utilities.decorators import beholder, seekset -from pcapkit.utilities.exceptions import (ProtocolNotFound, ProtocolNotImplemented, RegistryError, - StructError, UnsupportedCall) +from pcapkit.utilities.exceptions import (BaseError, ProtocolError, ProtocolNotFound, + ProtocolNotImplemented, RegistryError, StructError, + UnsupportedCall) from pcapkit.utilities.warnings import RegistryWarning, UnknownFieldWarning, warn if TYPE_CHECKING: @@ -67,6 +70,128 @@ # readable characters' order list readable = [ord(char) for char in filter(lambda char: not char.isspace(), string.printable)] + +@functools.lru_cache(maxsize=None) +def _make_index_calls(func: 'Callable[..., Any]') -> 'tuple[tuple[int, int, str], ...]': + """Locate the ``_make_index`` calls in the source of ``func``. + + Args: + func: method calling ``_make_index`` + + Returns: + One ``(first_line, last_line, argument)`` triple per call whose first + argument is a plain name, with absolute line numbers. Empty if the + source of ``func`` is unavailable. + + """ + try: + lines, start = inspect.getsourcelines(func) + tree = ast.parse(textwrap.dedent(''.join(lines))) + except (OSError, TypeError, SyntaxError): + return () + return tuple( + (node.lineno + start - 1, getattr(node, 'end_lineno', node.lineno) + start - 1, node.args[0].id) + for node in ast.walk(tree) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + and node.func.attr == '_make_index' and node.args and isinstance(node.args[0], ast.Name) + ) + + +def _passes_raw(value: 'Any', nested: 'bool' = False) -> 'bool': + """Tell if ``make()`` keywords may carry a raw :obj:`bytes` item. + + Args: + value: keyword arguments of ``make()``, or a value within them + nested: if ``value`` sits inside a container + + Returns: + :data:`True` if :obj:`bytes` sits anywhere inside a :obj:`list`, + :obj:`tuple` or :obj:`dict` among ``value``, or a + :class:`~pcapkit.protocols.schema.schema.Schema` does anywhere. A + top-level :obj:`bytes` keyword (e.g. a payload) does not count. + + """ + if isinstance(value, bytes): + return nested + if isinstance(value, Schema): # a ready-made item, which ``make()`` may pack into raw octets + return True + if isinstance(value, dict): + return any(_passes_raw(item, nested) for key, item in value.items() if key != '__packet__') + if isinstance(value, (list, tuple)): + return any(_passes_raw(item, True) for item in value) + return False + + +@functools.lru_cache(maxsize=None) +def _schema_slots(schema: 'Type[Schema]') -> 'tuple[str, ...]': + """Names of the fields of ``schema`` that hold a schema. + + Args: + schema: schema class + + Returns: + Names of the :class:`~pcapkit.corekit.fields.misc.SchemaField` and + :class:`~pcapkit.corekit.fields.misc.SwitchField` fields, including + those wrapped in a :class:`~pcapkit.corekit.fields.misc.ConditionalField`. + + """ + slots = [] + for name, field in schema.__fields__.items(): + while isinstance(field, ConditionalField): + field = field.field + if isinstance(field, (SchemaField, SwitchField)): + slots.append(name) + return tuple(slots) + + +def _holds_raw(value: 'Any') -> 'bool': + """Tell if a made schema may hold raw :obj:`bytes` in place of a schema. + + Args: + value: made schema, or a value within it + + Returns: + :data:`True` if ``value`` holds :obj:`bytes` as a list item, or as + the value of a field that holds a schema + (:class:`~pcapkit.corekit.fields.misc.SchemaField` or + :class:`~pcapkit.corekit.fields.misc.SwitchField`). + + """ + if isinstance(value, (list, tuple)): + return any(isinstance(item, bytes) or _holds_raw(item) for item in value) + if isinstance(value, Schema): + for name, field in value.__fields__.items(): + item = value.get(name) + while isinstance(field, ConditionalField): + field = field.field + if isinstance(item, bytes) and isinstance(field, (SchemaField, SwitchField)): + return True + if _holds_raw(item): + return True + return False + + +def _raw_where_parsed(made: 'Any', parsed: 'Any') -> 'bool': + """Tell if ``made`` holds raw :obj:`bytes` where ``parsed`` holds a schema. + + Args: + made: made schema, or a value within it + parsed: the same position in the schema parsed from its packed octets + + Returns: + :data:`True` if, at some position both share, ``made`` holds + :obj:`bytes` and ``parsed`` a :class:`~pcapkit.protocols.schema.schema.Schema`. + + """ + if isinstance(made, bytes): + return isinstance(parsed, Schema) + if isinstance(made, (list, tuple)) and isinstance(parsed, (list, tuple)): + return any(_raw_where_parsed(item, other) for item, other in zip(made, parsed)) + if isinstance(made, Schema) and isinstance(parsed, Schema): + return any(_raw_where_parsed(made.get(name), parsed.get(name)) + for name in made.__fields__ if name in parsed.__fields__) + return False + #: Keywords that configure the construction rather than naming a field, and are #: therefore consumed by :meth:`ProtocolBase.__init__ #: ` or by the schema layer @@ -572,6 +697,9 @@ def pack(self, **kwargs: 'Any') -> 'bytes': """ self.__header__ = self.make(**kwargs) + #: bool: If the made header may hold raw items (see :meth:`unpack`). + self._make_raw = _passes_raw(kwargs) or any( + isinstance(self.__header__.get(name), bytes) for name in _schema_slots(type(self.__header__))) packet = kwargs.get('__packet__', {}) # packet data return self.__header__.pack(packet) @@ -590,10 +718,28 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': global packet data to underlying methods. This is useful when the packet data is not available in the current instance. + When :meth:`pack` or :meth:`from_schema` has already set + :attr:`self.__header__ `, that schema is + kept, unless it still holds raw :obj:`bytes` where parsing its + packed octets yields a schema -- e.g. a raw option given to + ``make()`` -- which :meth:`read` cannot consume. The header is + then replaced by the parsed one (:issue:`1312`). + + Raises: + ProtocolError: If the made header holds raw items and its packed + octets do not parse. + """ + packet = kwargs.get('__packet__', {}) # packet data if cast('Optional[_ST]', self.__header__) is None: - packet = kwargs.get('__packet__', {}) # packet data self.__header__ = cast('_ST', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc] + elif self._make_raw and _holds_raw(self.__header__): + try: + parsed = cast('_ST', type(self.__header__).unpack(io.BytesIO(self._data), length, packet)) # type: ignore[call-arg,misc] + except (BaseError, struct.error) as error: # pylint: disable=no-member + raise ProtocolError(f'{type(self).__name__}: malformed raw item given to make()') from error + if _raw_where_parsed(self.__header__, parsed): + self.__header__ = parsed return self.read(length, **kwargs) @staticmethod @@ -857,6 +1003,9 @@ def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'Self': # pylint: disable=un # reference between instances. self.__cached__ = {} self.__header__ = None # type: ignore[assignment] + #: bool: If a header set before :meth:`unpack` may hold raw items; + #: :meth:`pack` narrows it down for a made one. + self._make_raw = True return self @@ -1603,12 +1752,8 @@ def _make_index(cls, name: 'str | int | StdlibEnum | AenumEnum', default: 'Optio 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`). + that ``name`` was passed as, found from the call in the + caller's source (see :meth:`_make_index_namespace`). """ if isinstance(name, (enum.Enum, aenum.Enum)): @@ -1659,6 +1804,15 @@ def _make_index_namespace(cls, name: 'str', member. The candidate namespace of a match is the class of that default value. + Only the argument the call passes as ``name`` is considered: the + caller's source is parsed (by ``_make_index_calls``) for the + :meth:`_make_index` call spanning the current line, and its first + argument names the method argument, so two arguments holding the + same member name each resolve against their own enumeration. Every + argument is considered when that call cannot be located (e.g. the + source is unavailable) or its first argument is not an argument of + the method (e.g. ``value = arg; self._make_index(value)``). + Args: name: member name given to :meth:`_make_index` frame: frame of :meth:`_make_index` @@ -1671,9 +1825,9 @@ def _make_index_namespace(cls, name: 'str', 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. + ProtocolNotImplemented: If every argument is considered, 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 @@ -1696,8 +1850,13 @@ def _make_index_namespace(cls, name: 'str', return None params = inspect.signature(func).parameters + lineno = caller.f_lineno + located = {arg for first, last, arg in _make_index_calls(func) if first <= lineno <= last} found = {} # type: dict[str, Type[StdlibEnum] | Type[AenumEnum]] for arg, param in params.items(): + # once the call is located, only its own argument is a candidate + if len(located) == 1 and located <= params.keys() and arg not in located: + continue value = values.get(arg) if f'{arg}_namespace' not in params or not isinstance(value, str) or value != name: continue diff --git a/tests/protocols/test_make_index_str_unit.py b/tests/protocols/test_make_index_str_unit.py index cd4beb70d0..ea8ccae827 100644 --- a/tests/protocols/test_make_index_str_unit.py +++ b/tests/protocols/test_make_index_str_unit.py @@ -108,7 +108,8 @@ def test_shared_name_with_one_meaning_resolves(self) -> None: 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: + def test_shared_name_with_two_meanings_resolves_per_argument(self) -> None: + # the per-argument resolution of #1308; see test_protocol_make_unit enum = importlib.import_module('enum') exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') @@ -121,16 +122,13 @@ 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'), 1) 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 + # a real make(): ``LOW`` is a ToSDelay member but not a ToSThroughput one, + # so ``tos_thr`` alone fails to resolve it 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"): + with self.assertRaisesRegex(exc, r"protocol 'LOW' not implemented"): ipv4(tos_del='LOW', tos_thr='LOW') def test_without_inferable_enum_still_raises(self) -> None: diff --git a/tests/protocols/test_protocol_make_unit.py b/tests/protocols/test_protocol_make_unit.py new file mode 100644 index 0000000000..d157318620 --- /dev/null +++ b/tests/protocols/test_protocol_make_unit.py @@ -0,0 +1,236 @@ +# -*- coding: utf-8 -*- +"""Construction through ``make()`` in :class:`~pcapkit.protocols.protocol.Protocol`. + +GitHub issue #1312: :meth:`~pcapkit.protocols.protocol.ProtocolBase.unpack` +reused the schema :meth:`~pcapkit.protocols.protocol.ProtocolBase.pack` had +made instead of parsing the packed octets, so ``read`` met the raw +:obj:`bytes` (or :class:`~pcapkit.protocols.schema.schema.Schema`) items +the option makers accept, and construction crashed with +:exc:`AttributeError`. A bare ``MH()`` crashed the same way on its default +``data=b'\\x00\\x00'``. A made schema holding raw octets where parsing its +packed form yields a schema is now replaced by the parsed one. + +GitHub issue #1308: a :obj:`str` member name given to two enum arguments +of one ``make()`` raised ``ambiguous member name`` when only one of the +enumerations had it. Each ``_make_index`` call now resolves against the +argument it was passed. + +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 +import unittest.mock +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) + + +#: One raw item per protocol whose ``make()`` accepts raw octets in a list: +#: (protocol class, construction keywords, packed octets). +RAW_ITEMS = { + 'HIP parameters': ('pcapkit.protocols.internet.hip.HIP', + {'extension': True, 'parameters': [b'\x00\x41\x00\x0c' + bytes(12)]}, + '1106206100000000' + '00' * 32 + '0041000c' + '00' * 12), + 'IPv4 options': ('pcapkit.protocols.internet.ipv4.IPv4', + {'options': [b'\x94\x04\x00\x00'], 'protocol': 253}, + '460000180000000000fd00007f0000010000000094040000'), + 'TCP options': ('pcapkit.protocols.transport.tcp.TCP', + {'options': [b'\x02\x04\x05\xb4']}, + '0000000000000000000000006000ffff00000000020405b4'), + 'HOPOPT options': ('pcapkit.protocols.internet.hopopt.HOPOPT', + {'extension': True, 'options': [b'\x05\x02\x00\x00']}, + '1100050200000000'), + 'IPv6_Opts options': ('pcapkit.protocols.internet.ipv6_opts.IPv6_Opts', + {'extension': True, 'options': [b'\x05\x02\x00\x00']}, + '1100050200000000'), + 'MH options': ('pcapkit.protocols.internet.mh.MH', + {'data': {'options': [b'\x01\x02\x00\x00']}}, + '11010000000000000102000001020000'), + 'SCTP chunks': ('pcapkit.protocols.transport.sctp.SCTP', + {'chunks': [b'\x0e\x00\x00\x04']}, + '0000000000000000617cebb70e000004'), + 'IPv6_Route data': ('pcapkit.protocols.internet.ipv6_route.IPv6_Route', + {'extension': True, 'data': bytes(4)}, + '1100000000000000'), +} + + +class TestMakeRawItems(unittest.TestCase): + """Raw and :class:`Schema` items given to ``make()`` build a packet (#1312).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_raw_item_builds_and_parses(self) -> None: + for case, (klass, kwargs, packed) in RAW_ITEMS.items(): + with self.subTest(case=case): + cls = _attr(klass) + proto = cls(**kwargs) + self.assertEqual(proto.data.hex(), packed) + # the made packet reads as the same octets parsed afresh + extra = {'extension': True} if kwargs.get('extension') else {} + self.assertEqual(proto.info, cls(proto.data, **extra).info) + + def test_schema_item_builds_as_raw_item(self) -> None: + for case in ('HIP parameters', 'IPv4 options', 'TCP options'): + with self.subTest(case=case): + klass, kwargs, packed = RAW_ITEMS[case] + cls = _attr(klass) + key = next(key for key in ('parameters', 'options') if key in kwargs) + extra = {'extension': True} if kwargs.get('extension') else {} + header = cls(bytes.fromhex(packed), **extra).__header__ + item = getattr(header, 'param' if key == 'parameters' else 'options')[0] + self.assertNotIsInstance(item, bytes) + proto = cls(**{**kwargs, key: [item]}) + self.assertEqual(proto.data.hex(), packed) + + def test_bare_mh_constructs(self) -> None: + mh = _attr('pcapkit.protocols.internet.mh.MH') + proto = mh() + self.assertEqual(proto.data.hex(), '1100000000000000') + self.assertEqual(proto.info, mh(proto.data).info) + + def test_made_schema_without_raw_items_is_kept(self) -> None: + module = importlib.import_module('pcapkit.protocols.protocol') + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + option = _attr('pcapkit.const.ipv4.option_number.OptionNumber') + + # an option given as a tuple is made into a schema, so nothing is re-parsed + made = ipv4.make(ipv4.__new__(ipv4), options=[(option.RTRALT, {'alert': 0})], protocol=253) + self.assertFalse(module._holds_raw(made)) + raw = ipv4.make(ipv4.__new__(ipv4), options=[b'\x94\x04\x00\x00'], protocol=253) + self.assertTrue(module._holds_raw(raw)) + + parsed = ipv4(raw.pack()).__header__ + self.assertTrue(module._raw_where_parsed(raw, parsed)) + self.assertFalse(module._raw_where_parsed(parsed, parsed)) + # raw octets matched by raw octets are not a stand-in for a schema + self.assertFalse(module._raw_where_parsed([b'\x01'], [b'\x01'])) + + def test_unparsable_raw_item_raises_protocol_error(self) -> None: + exc = _attr('pcapkit.utilities.exceptions.ProtocolError') + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + # a lone length-less option makes the packed octets unparsable + with self.assertRaisesRegex(exc, r'malformed raw item given to make\(\)') as caught: + ipv4(options=[b'\x02', b'\x94\x04\x00\x00'], protocol=253) + self.assertIsInstance(caught.exception.__cause__, exc) + + def test_random_raw_items_raise_library_errors_only(self) -> None: + random = importlib.import_module('random') + base = _attr('pcapkit.utilities.exceptions.BaseError') + rng = random.Random(1352) + + def items() -> 'list[bytes]': + return [bytes(rng.randrange(256) for _ in range(rng.randrange(1, 13))) + for _ in range(rng.randrange(1, 3))] + + for case, (klass, kwargs, _) in RAW_ITEMS.items(): + key = next(iter(kw for kw in ('parameters', 'options', 'chunks', 'data') if kw in kwargs)) + cls = _attr(klass) + with self.subTest(case=case): + for _ in range(200): + value = items() # type: Any + if case == 'MH options': + value = {'options': value} + elif case == 'IPv6_Route data': + value = b''.join(value) + try: + with importlib.import_module('warnings').catch_warnings(): + importlib.import_module('warnings').simplefilter('ignore') + cls(**{**kwargs, key: value}) + except base: + pass + + def test_malformed_raw_item_raises_protocol_error(self) -> None: + # ESP_INFO needs a 12-octet body; an 8-octet one is a parse error, + # reported as such rather than as an AttributeError on raw bytes + hip = _attr('pcapkit.protocols.internet.hip.HIP') + exc = _attr('pcapkit.utilities.exceptions.ProtocolError') + with self.assertRaisesRegex(exc, r'\[ParamNo 65\] invalid format'): + hip(extension=True, parameters=[b'\x00\x41\x00\x08' + bytes(8)]) + + +class TestMakeIndexPerArgument(unittest.TestCase): + """A member name resolves against its own ``make()`` argument (#1308).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _pair(self) -> 'tuple[Any, Any, Any]': + enum = importlib.import_module('enum') + protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') + + left = enum.IntEnum('Left', {'Shared': 1}) + right = enum.IntEnum('Right', {'Shared': 2, 'Only': 3}) + + 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) -> 'tuple[int, int]': + return (self._make_index(a, namespace=a_namespace), + self._make_index(b, 0, namespace=b_namespace)) + + return Pair, left, right + + def test_shared_name_resolves_per_argument(self) -> None: + pair, _, _ = self._pair() + self.assertEqual(pair.make_pair(pair, a='Shared', b='Shared'), (1, 2)) + # a name only one argument holds resolves as it did before + self.assertEqual(pair.make_pair(pair, a='Shared', b='Only'), (1, 3)) + + def test_ipv4_shared_name_falls_back_per_argument(self) -> None: + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + tos_del = _attr('pcapkit.const.ipv4.tos_del.ToSDelay') + + expected = ipv4(tos_del=tos_del.LOW, tos_thr=1).data + # ``LOW`` is a ToSDelay member but not a ToSThroughput one + self.assertEqual(ipv4(tos_del='LOW', tos_thr='LOW', tos_thr_default=1).data, expected) + self.assertEqual(ipv4(tos_thr='LOW', tos_thr_default=1).data, + ipv4(tos_thr=1).data) + + def test_non_parameter_first_argument_matches_every_argument(self) -> None: + enum = importlib.import_module('enum') + protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') + level = enum.IntEnum('Level', {'HIGH': 7}) + + class Alias(protocol): # type: ignore[misc,valid-type] + def make_alias(self, b: 'Any' = level.HIGH, b_namespace: 'Any' = None) -> 'int': + value = b + return self._make_index(value) + + # ``value`` is a local, so ``b`` is found by its value as before + self.assertEqual(Alias.make_alias(Alias, b='HIGH'), 7) + + def test_unlocated_call_matches_every_argument(self) -> None: + # without the caller's source the call cannot be located, so every + # argument holding the name is a candidate and they must agree + module = importlib.import_module('pcapkit.protocols.protocol') + exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') + pair, _, right = self._pair() + + with unittest.mock.patch.object(module, '_make_index_calls', lambda func: ()): + 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') + self.assertEqual(pair.make_pair(pair, a='Shared', b='Shared', b_namespace=right), (1, 2)) + + def test_source_unavailable_locates_nothing(self) -> None: + module = importlib.import_module('pcapkit.protocols.protocol') + # a builtin has no Python source + self.assertEqual(module._make_index_calls(len), ()) + + +if __name__ == '__main__': + unittest.main()