From e549c91623ddd847562c8ec3d5e963d5cea5a9f5 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 18:54:03 -0400 Subject: [PATCH] fix(protocols): restoring a registration puts back the shipped ModuleDescriptor (#1504) - ProtocolBase: snapshot each class's own __proto__ at class creation as _shipped_proto, and add _next_layer_entry, which decides what a register stores: None for the stored entry (#1364), the shipped descriptor object itself, unresolved, for a descriptor equal to the one the code shipped with, else the resolved and checked class. - ProtocolBase, Link, Internet, Transport, SCTP and Frame register use it and keep their own warning, so undoing an override leaves the entry a descriptor and imports nothing. - PCAPNG.register (misc/pcapng.py) is not changed here; the Gap(1504) row narrows to register_linktype and register_pcapng. - New per-site tests for the round trip, the restore importing nothing, and dispatch resolving the restored entry when it runs. Part of #1504 --- pcapkit/protocols/internet/internet.py | 21 +- pcapkit/protocols/link/link.py | 21 +- pcapkit/protocols/misc/pcap/frame.py | 22 +- pcapkit/protocols/protocol.py | 79 ++++++- pcapkit/protocols/transport/sctp.py | 22 +- pcapkit/protocols/transport/transport.py | 22 +- .../registry/test_registry_symmetry_unit.py | 17 +- ...r_restores_shipped_descriptor_1504_unit.py | 197 ++++++++++++++++++ 8 files changed, 316 insertions(+), 85 deletions(-) create mode 100644 tests/protocols/test_register_restores_shipped_descriptor_1504_unit.py diff --git a/pcapkit/protocols/internet/internet.py b/pcapkit/protocols/internet/internet.py index 7dd4da0355..7cb51dbe0e 100644 --- a/pcapkit/protocols/internet/internet.py +++ b/pcapkit/protocols/internet/internet.py @@ -21,7 +21,6 @@ from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read from pcapkit.utilities.decorators import beholder -from pcapkit.utilities.exceptions import RegistryError from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: @@ -156,24 +155,18 @@ def register(cls, code: 'Enum_TransType', protocol: 'ModuleDescriptor[ProtocolBa replacement so a caller can tell *what* was lost. Fires only when the incumbent differs from the replacement; see :meth:`ProtocolBase.register ` for the - guard this shares with ``register_protocol``. + guard this shares with ``register_protocol``, and for the + shipped descriptor a restore puts back (:issue:`1504`). """ incumbent = cls.__proto__.get(code) - if incumbent is not None and (incumbent is protocol or ( - isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) - and incumbent == protocol)): + entry = cls._next_layer_entry(code, protocol) + if entry is None: return - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - if not isinstance(protocol, type): - raise RegistryError(f'protocol must be a class, not {protocol!r}') - if not issubclass(protocol, ProtocolBase): - raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - if incumbent is not None and incumbent is not protocol: + if incumbent is not None and incumbent is not entry: warn(f'protocol {code} already registered, overwriting ' - f'{incumbent!r} with {protocol!r}', RegistryWarning) - cls.__proto__[code] = protocol + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry @classmethod def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': diff --git a/pcapkit/protocols/link/link.py b/pcapkit/protocols/link/link.py index 805e35e674..2010f550d0 100644 --- a/pcapkit/protocols/link/link.py +++ b/pcapkit/protocols/link/link.py @@ -22,7 +22,6 @@ from pcapkit.corekit.module import ModuleDescriptor from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read -from pcapkit.utilities.exceptions import RegistryError from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: @@ -136,24 +135,18 @@ def register(cls, code: 'Enum_EtherType', protocol: 'ModuleDescriptor[ProtocolBa when the incumbent differs from the replacement -- see :meth:`ProtocolBase.register ` for the - guard this shares with ``register_protocol``. + guard this shares with ``register_protocol``, and for the + shipped descriptor a restore puts back (:issue:`1504`). """ incumbent = cls.__proto__.get(code) - if incumbent is not None and (incumbent is protocol or ( - isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) - and incumbent == protocol)): + entry = cls._next_layer_entry(code, protocol) + if entry is None: return - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - if not isinstance(protocol, type): - raise RegistryError(f'protocol must be a class, not {protocol!r}') - if not issubclass(protocol, ProtocolBase): - raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - if incumbent is not None and incumbent is not protocol: + if incumbent is not None and incumbent is not entry: warn(f'protocol {code} already registered, overwriting ' - f'{incumbent!r} with {protocol!r}', RegistryWarning) - cls.__proto__[code] = protocol + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry @classmethod def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index 86f7041fbf..f92d1ef397 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -37,7 +37,7 @@ from pcapkit.protocols.protocol import ProtocolBase from pcapkit.protocols.schema.misc.pcap.frame import Frame as Schema_Frame from pcapkit.utilities.compat import localcontext -from pcapkit.utilities.exceptions import RegistryError, StreamEOFError, UnsupportedCall, stacklevel +from pcapkit.utilities.exceptions import StreamEOFError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import ProtocolWarning, RegistryWarning, warn if TYPE_CHECKING: @@ -143,24 +143,18 @@ def register(cls, code: 'Enum_LinkType', protocol: 'ModuleDescriptor[ProtocolBas Fires only when the incumbent differs from the replacement -- see :meth:`ProtocolBase.register ` for the - guard this shares with ``register_protocol``. + guard this shares with ``register_protocol``, and for the + shipped descriptor a restore puts back (:issue:`1504`). """ incumbent = cls.__proto__.get(code) - if incumbent is not None and (incumbent is protocol or ( - isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) - and incumbent == protocol)): + entry = cls._next_layer_entry(code, protocol) + if entry is None: return - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - if not isinstance(protocol, type): - raise RegistryError(f'protocol must be a class, not {protocol!r}') - if not issubclass(protocol, ProtocolBase): - raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - if incumbent is not None and incumbent is not protocol: + if incumbent is not None and incumbent is not entry: warn(f'protocol {code} already registered, overwriting ' - f'{incumbent!r} with {protocol!r}', RegistryWarning) - cls.__proto__[code] = protocol + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry def index(self, name: 'str | ProtocolBase | Type[ProtocolBase]') -> 'int': """Call :meth:`ProtoChain.index `. diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 9ce4902d41..9ea07a74d9 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -26,6 +26,7 @@ import string import struct import textwrap +import types import urllib.parse from typing import TYPE_CHECKING, Any, Generic, Optional, Type, TypeVar, cast, overload @@ -456,6 +457,13 @@ class ProtocolBase(Generic[_PT, _ST], metaclass=ProtocolMeta): lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), ) + #: Read-only copy of :attr:`__proto__` as the class defining it shipped it, + #: taken by :meth:`__init_subclass__` before anything can register into it. + #: :meth:`_next_layer_entry` reads it to tell a descriptor handed back from + #: a new one without resolving it (:issue:`1504`). + _shipped_proto: 'types.MappingProxyType[Any, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]' = \ + types.MappingProxyType({}) + #: Construction keywords this protocol consumes out of ``**kwargs`` instead #: of declaring as a parameter, e.g. with ``kwargs.get('spam')`` in #: :meth:`read` -- as :meth:`ESP.read ` @@ -914,22 +922,73 @@ class object, or a :class:`~pcapkit.corekit.module.ModuleDescriptor` deferred, and an incumbent left as an unresolved descriptor still reports as different from the class it names. + A descriptor equal to the one ``code`` shipped with restores that + shipped object, unresolved, so undoing an override leaves the entry + as shipped (:issue:`1504`); see :meth:`_next_layer_entry`, which + every layer's ``register`` shares. + + """ + incumbent = cls.__proto__.get(code) + entry = cls._next_layer_entry(code, protocol) + if entry is None: + return + if incumbent is not None and incumbent is not entry: + warn(f'protocol {code} already registered, overwriting ' + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry + + @classmethod + def _next_layer_entry(cls, code: 'Any', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]' + ) -> 'Optional[ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]': + """Work out what a next-layer ``register`` stores under ``code``. + + :meth:`register` and the :class:`~pcapkit.protocols.link.link.Link`, + :class:`~pcapkit.protocols.internet.internet.Internet`, + :class:`~pcapkit.protocols.transport.transport.Transport`, + :class:`~pcapkit.protocols.transport.sctp.SCTP` and + :class:`~pcapkit.protocols.misc.pcap.frame.Frame` overrides share this; + each still warns and stores for itself. + + Arguments: + code: registry key + protocol: module descriptor or a :class:`ProtocolBase` subclass + + Returns: + :obj:`None` if ``protocol`` is the entry stored under ``code`` + already -- the same object, or a descriptor equal to a stored one + (:issue:`1364`). Otherwise the entry to store: if ``protocol`` is a + descriptor equal to the one ``code`` shipped with, per + :attr:`_shipped_proto`, the shipped object itself, unresolved; + else ``protocol`` resolved to its class. + + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + class, or not a :class:`ProtocolBase` subclass. + + Note: + The shipped descriptor is matched before it is resolved, so an + override undone by handing back the shipped entry leaves the + registry as it shipped: the entry is still a descriptor, nothing is + imported, and the class is looked up when a frame dispatches to it + (:issue:`1504`). + """ incumbent = cls.__proto__.get(code) if incumbent is not None and (incumbent is protocol or ( isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) and incumbent == protocol)): - return + return None + shipped = cls._shipped_proto.get(code) + if (isinstance(protocol, ModuleDescriptor) and isinstance(shipped, ModuleDescriptor) + and protocol == shipped): + return shipped if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not isinstance(protocol, type): raise RegistryError(f'protocol must be a class, not {protocol!r}') if not issubclass(protocol, ProtocolBase): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - if incumbent is not None and incumbent is not protocol: - warn(f'protocol {code} already registered, overwriting ' - f'{incumbent!r} with {protocol!r}', RegistryWarning) - cls.__proto__[code] = protocol + return protocol @classmethod def from_schema(cls, schema: '_ST | dict[str, Any]') -> 'Self': @@ -1304,6 +1363,16 @@ def __init_subclass__(cls, /, schema: 'Optional[Type[_ST]]' = None, cls.__schema__ = schema cls.__data__ = data + # NOTE: a class defining its own ``__proto__`` ships what its body put + # there; copy it now, before anything can register into it, so that + # ``_next_layer_entry`` can restore a shipped descriptor without + # resolving it (:issue:`1504`). A class that inherits ``__proto__`` + # inherits this copy with it. + if '__proto__' in cls.__dict__: + table = cls.__dict__['__proto__'] + cls._shipped_proto = types.MappingProxyType( + dict(table) if isinstance(table, collections.abc.Mapping) else {}) + if code is not None: from pcapkit.foundation.registry.protocols import \ register_protocol_code # pylint: disable=import-outside-toplevel diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 322c250dae..ed19dc8b96 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -156,7 +156,7 @@ from pcapkit.protocols.schema.transport.sctp import \ UserInitiatedAbortCause as Schema_UserInitiatedAbortCause from pcapkit.protocols.transport.transport import Transport -from pcapkit.utilities.exceptions import ProtocolError, RegistryError +from pcapkit.utilities.exceptions import ProtocolError from pcapkit.utilities.warnings import RegistryWarning, warn if TYPE_CHECKING: @@ -736,24 +736,18 @@ def register(cls, code: 'Enum_PayloadProtocolIdentifier | int', protocol: 'Modul :class:`~pcapkit.corekit.module.ModuleDescriptor` incumbent counts as different from the class it names; handing back the stored entry, or a descriptor equal to it, is a silent no-op - that keeps it. + that keeps it, and handing back the descriptor a PPID shipped + with restores that object, unresolved (:issue:`1504`). """ incumbent = cls.__proto__.get(code) - if incumbent is not None and (incumbent is protocol or ( - isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) - and incumbent == protocol)): + entry = cls._next_layer_entry(code, protocol) + if entry is None: return - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - if not isinstance(protocol, type): - raise RegistryError(f'protocol must be a class, not {protocol!r}') - if not issubclass(protocol, ProtocolBase): - raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - if incumbent is not None and incumbent is not protocol: + if incumbent is not None and incumbent is not entry: warn(f'payload protocol identifier {code} already registered, overwriting ' - f'{incumbent!r} with {protocol!r}', RegistryWarning) - cls.__proto__[code] = protocol + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry @classmethod def register_chunk(cls, code: 'Enum_Chunk', meth: 'str | tuple[ChunkParser, ChunkConstructor]') -> 'None': diff --git a/pcapkit/protocols/transport/transport.py b/pcapkit/protocols/transport/transport.py index 299bb4d76d..cfef40e52e 100644 --- a/pcapkit/protocols/transport/transport.py +++ b/pcapkit/protocols/transport/transport.py @@ -19,7 +19,7 @@ from pcapkit.corekit.module import ModuleDescriptor from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase from pcapkit.protocols.schema.schema import keep_short_read, replay_short_read -from pcapkit.utilities.exceptions import RegistryError, StructError, UnsupportedCall, stacklevel +from pcapkit.utilities.exceptions import StructError, UnsupportedCall, stacklevel from pcapkit.utilities.logging import DEVMODE, get_logger from pcapkit.utilities.warnings import RegistryWarning, warn @@ -100,7 +100,8 @@ def register(cls, code: 'int', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ different from the class it names. See :meth:`ProtocolBase.register ` for the - guard this shares with ``register_protocol``. + guard this shares with ``register_protocol``, and for the + shipped descriptor a restore puts back (:issue:`1504`). Note: :class:`~pcapkit.protocols.transport.tcp.TCP` and @@ -114,20 +115,13 @@ def register(cls, code: 'int', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ raise UnsupportedCall(f'{cls.__name__} is an abstract class') incumbent = cls.__proto__.get(code) - if incumbent is not None and (incumbent is protocol or ( - isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) - and incumbent == protocol)): + entry = cls._next_layer_entry(code, protocol) + if entry is None: return - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - if not isinstance(protocol, type): - raise RegistryError(f'protocol must be a class, not {protocol!r}') - if not issubclass(protocol, ProtocolBase): - raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - if incumbent is not None and incumbent is not protocol: + if incumbent is not None and incumbent is not entry: warn(f'port {code} already registered, overwriting ' - f'{incumbent!r} with {protocol!r}', RegistryWarning) - cls.__proto__[code] = protocol + f'{incumbent!r} with {entry!r}', RegistryWarning) + cls.__proto__[code] = entry @classmethod def analyze(cls, ports: 'tuple[int, int]', payload: 'bytes', **kwargs: 'Any') -> 'ProtocolBase': # type: ignore[override] # pylint: disable=arguments-renamed diff --git a/tests/foundation/registry/test_registry_symmetry_unit.py b/tests/foundation/registry/test_registry_symmetry_unit.py index d4b885daf2..4a1b9901d1 100644 --- a/tests/foundation/registry/test_registry_symmetry_unit.py +++ b/tests/foundation/registry/test_registry_symmetry_unit.py @@ -405,17 +405,14 @@ class TestRegistrySymmetry(harness.RoundTripBase): STATUSES = ('RAISED', 'CHANGED') KNOWN_FAILURES = ( - Gap(1504, 'restoring a shipped dispatch entry over an override stores the class it names, ' - 'not the shipped ModuleDescriptor: each register resolves a descriptor unequal ' - 'to the incumbent (pcapkit/protocols/protocol.py:924, link/link.py:148, ' - 'internet/internet.py:168, transport/transport.py:122, misc/pcap/frame.py:155, ' - 'misc/pcapng.py:1027, transport/sctp.py:748), where the foundation registrars ' - 'keep the shipped object (#1363, #1364)', - 'CHANGED', ("ModuleDescriptor(module='pcapkit.protocols.",), + Gap(1504, 'restoring a shipped PCAPNG.__proto__ entry over an override stores the class ' + 'it names, not the shipped ModuleDescriptor: PCAPNG.register resolves a ' + 'descriptor unequal to the incumbent (pcapkit/protocols/misc/pcapng.py:1027) ' + 'instead of sharing ProtocolBase._next_layer_entry, as the other layers do', + 'CHANGED', ("PCAPNG.__proto__[] " + "ModuleDescriptor(module='pcapkit.protocols.",), tuple(f'{name}/{variant}/dispatch' - for name in ('register_linktype', 'register_pcap', 'register_pcapng', - 'register_ethertype', 'register_transtype', 'register_tcp', - 'register_udp', 'register_apptype', 'register_sctp') + for name in ('register_linktype', 'register_pcapng') for variant in ('override', 'override-same-name', 'restore-by-name'))), Gap(1505, 'a protocol registrar adds the class it registers to pcapkit.protocols.__proto__ ' 'by name (pcapkit/foundation/registry/protocols.py:235, reached from e.g. :416), ' diff --git a/tests/protocols/test_register_restores_shipped_descriptor_1504_unit.py b/tests/protocols/test_register_restores_shipped_descriptor_1504_unit.py new file mode 100644 index 0000000000..17957c81e6 --- /dev/null +++ b/tests/protocols/test_register_restores_shipped_descriptor_1504_unit.py @@ -0,0 +1,197 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1504: restoring a shipped dispatch entry puts back its descriptor. + +:meth:`ProtocolBase.register ` +and its ``Link``, ``Internet``, ``Transport`` (``TCP``, ``UDP``), ``SCTP`` and +``Frame`` overrides resolved a :class:`~pcapkit.corekit.module.ModuleDescriptor` +argument to its class before storing it. The #1364 guard returns early only when +the argument equals the *incumbent*, so after an override, handing back the +shipped descriptor stored the class it names: the round trip did not close, and +the restore imported the module the entry was meant to defer. A descriptor equal +to the one a code shipped with is now matched before it is resolved, and the +shipped object itself is stored. + +``PCAPNG.register`` has the same defect and is not covered here; it stays +recorded under ``Gap(1504, ...)`` in +:file:`tests/foundation/registry/test_registry_symmetry_unit.py`. + +Every registry is process-wide, so each case runs under +:func:`unittest.mock.patch.dict`. Classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import collections +import contextlib +import importlib +import sys +import types +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class + +#: ``(module, class, code)`` per site family, ``code`` given as a +#: ``module:Enum.member`` path or a literal; each code ships a descriptor. +#: ``ProtocolBase.register`` has no populated registry of its own, so +#: :meth:`TestRestoreShippedDescriptor._sites` adds a scratch subclass for it. +SITES = ( + ('pcapkit.protocols.link.link', 'Link', + 'pcapkit.const.reg.ethertype:EtherType.Address_Resolution_Protocol'), + ('pcapkit.protocols.internet.internet', 'Internet', 'pcapkit.const.reg.transtype:TransType.TCP'), + ('pcapkit.protocols.transport.tcp', 'TCP', 80), + ('pcapkit.protocols.transport.udp', 'UDP', 1701), + ('pcapkit.protocols.transport.sctp', 'SCTP', + 'pcapkit.const.sctp.payload_protocol_identifier:PayloadProtocolIdentifier.' + 'PayloadProtocolIdentifier_3GPP_NG_Application_Protocol'), + ('pcapkit.protocols.misc.pcap.frame', 'Frame', 'pcapkit.const.reg.linktype:LinkType.ETHERNET'), +) + +#: A code no site ships an entry under. +UNSHIPPED = 0xFFFE + + +def _code(spec): # type: ignore[no-untyped-def] + if isinstance(spec, str): + module, _, rest = spec.partition(':') + enum, _, member = rest.partition('.') + return getattr(getattr(importlib.import_module(module), enum), member) + return spec + + +class _Blocker: + """A :data:`sys.meta_path` finder that records and refuses one import.""" + + def __init__(self, name: 'str') -> None: + self.name = name + self.attempts = [] # type: list[str] + + def find_spec(self, name, path=None, target=None): # type: ignore[no-untyped-def] + if name == self.name: + self.attempts.append(name) + raise ModuleNotFoundError(f'blocked by the #1504 test: {name}') + return None + + +@contextlib.contextmanager +def _unimported(name: 'str'): # type: ignore[no-untyped-def] + """Run the block with ``name`` absent from :data:`sys.modules` and unimportable.""" + saved = sys.modules.pop(name) + blocker = _Blocker(name) + sys.meta_path.insert(0, blocker) + try: + yield blocker + finally: + sys.meta_path.remove(blocker) + sys.modules[name] = saved + + +@contextlib.contextmanager +def _stand_in(name: 'str', **attrs): # type: ignore[no-untyped-def] + """Run the block with ``name`` in :data:`sys.modules` replaced by a stub module.""" + saved = sys.modules[name] + stub = types.ModuleType(name) + stub.__dict__.update(attrs) + sys.modules[name] = stub + try: + yield stub + finally: + sys.modules[name] = saved + + +class TestRestoreShippedDescriptor(unittest.TestCase): + """Override, then restore the shipped descriptor, at every site family.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _sites(self): # type: ignore[no-untyped-def] + """Yield ``(label, cls, code)``, the scratch ``ProtocolBase`` site last.""" + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.protocols.misc.raw import Raw + + for module, name, code in SITES: + yield name, getattr(importlib.import_module(module), name), _code(code) + table = collections.defaultdict(lambda: ModuleDescriptor('pcapkit.protocols.misc.raw', 'Raw'), + {1: ModuleDescriptor('pcapkit.protocols.misc.null', 'NoPayload')}) + yield 'ProtocolBase', type('W1504Scratch', (Raw,), {'__module__': __name__, '__proto__': table}), 1 + + def _each_site(self, check): # type: ignore[no-untyped-def] + """Run ``check(cls, code, shipped, custom)`` per site, each under its own ``subTest``.""" + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.protocols.misc.raw import Raw + + custom = type('W1504Custom', (Raw,), {'__module__': __name__}) + for label, cls, code in self._sites(): + registry = cls.__proto__ + shipped = registry[code] + self.assertIsInstance(shipped, ModuleDescriptor, label) + with self.subTest(site=label), mock.patch.dict(registry), warnings.catch_warnings(): + warnings.simplefilter('ignore') + check(cls, code, shipped, custom) + + def test_restoring_the_shipped_descriptor_stores_that_object(self) -> None: + def check(cls, code, shipped, custom): # type: ignore[no-untyped-def] + cls.register(code, custom) + self.assertIs(cls.__proto__[code], custom) + cls.register(code, shipped) + self.assertIs(cls.__proto__[code], shipped) + self._each_site(check) + + def test_restoring_by_an_equal_descriptor_stores_the_shipped_object(self) -> None: + from pcapkit.corekit.module import ModuleDescriptor + + def check(cls, code, shipped, custom): # type: ignore[no-untyped-def] + cls.register(code, custom) + cls.register(code, ModuleDescriptor(shipped.module, shipped.name)) + self.assertIs(cls.__proto__[code], shipped) + self._each_site(check) + + def test_restoring_warns_once_naming_the_override(self) -> None: + from pcapkit.utilities.warnings import RegistryWarning + + def check(cls, code, shipped, custom): # type: ignore[no-untyped-def] + cls.register(code, custom) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + cls.register(code, shipped) + messages = [str(item.message) for item in caught if issubclass(item.category, RegistryWarning)] + self.assertEqual(len(messages), 1, messages) + self.assertIn('W1504Custom', messages[0]) + self._each_site(check) + + def test_restoring_imports_nothing(self) -> None: + """The restore leaves the shipped module unimported, as shipped.""" + def check(cls, code, shipped, custom): # type: ignore[no-untyped-def] + cls.register(code, custom) + with _unimported(shipped.module) as blocker: + cls.register(code, shipped) + self.assertNotIn(shipped.module, sys.modules) + self.assertEqual(blocker.attempts, []) + self.assertIs(cls.__proto__[code], shipped) + self._each_site(check) + + def test_dispatch_resolves_the_restored_entry_when_it_runs(self) -> None: + """After the round trip the class is looked up at dispatch, not at the restore.""" + from pcapkit.protocols.misc.raw import Raw + + def check(cls, code, shipped, custom): # type: ignore[no-untyped-def] + cls.register(code, custom) + cls.register(code, shipped) + marker = type('W1504Marker', (Raw,), {'__module__': __name__}) + with _stand_in(shipped.module, **{shipped.name: marker}): + self.assertIs(cls._lookup_next_layer(cls.__proto__, code), marker) + self._each_site(check) + + def test_a_shipped_descriptor_under_another_code_is_still_resolved(self) -> None: + """Only the code a descriptor shipped under restores it unresolved.""" + def check(cls, code, shipped, custom): # type: ignore[no-untyped-def] + self.assertNotIn(UNSHIPPED, cls.__proto__) + cls.register(UNSHIPPED, shipped) + self.assertIs(cls.__proto__[UNSHIPPED], shipped.klass) + self._each_site(check) + + +if __name__ == '__main__': + unittest.main()