From f4588115ee2ff874a847142bc56312b40b66cb0e Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 22 Sep 2026 23:26:29 -0400 Subject: [PATCH] fix(registry): report the silent schema overwrites, and name what the code-keyed registrars displaced (#692) Generalises #681, per the ask to "apply what #681 added to other registry as well". - `EnumSchema.register` assigned bare, so the schema half of 14 public registrars silently displaced a built-in while the parser half of the very same call warned. Guard it on presence, naming both schemas. - `EnumSchema.__init_subclass__` reaches the same registry without calling `register`, so `class MyOption(Option, code=...)` stayed silent too. Guard it as well, folding its two branches into one loop so the guard is written once. - The seven code-keyed registrars now name the displaced entry and its replacement. Their presence-only condition is deliberately unchanged: their key is caller-supplied and independent of the value, so #681's "present and a different class" has nothing to fix here, and the `ModuleDescriptor` incumbents these tables ship with would make it undecidable without resolving the descriptor -- forcing the import it exists to defer, just to decide whether to warn. - `ContextRegistry.register` already raises on a duplicate, and the reassembly and ESP registrars are unkeyed lists, so none of those three takes a guard. `import pcapkit` holds at 1 warning and 0 RegistryWarning; mypy 112 errors and pylint 364 messages both unchanged; schema.py coverage 99% with its 5 new statements covered and misses flat at 1. Fixes #692 --- pcapkit/protocols/internet/internet.py | 15 +- pcapkit/protocols/link/link.py | 15 +- pcapkit/protocols/misc/pcap/frame.py | 15 +- pcapkit/protocols/misc/pcapng.py | 15 +- pcapkit/protocols/protocol.py | 32 +- pcapkit/protocols/schema/schema.py | 72 ++++- pcapkit/protocols/transport/sctp.py | 16 +- pcapkit/protocols/transport/transport.py | 23 +- tests/foundation/registry/test_protocols.py | 80 +++++ .../schema/test_enum_schema_registry_unit.py | 281 ++++++++++++++++++ tests/protocols/schema/test_schema_unit.py | 19 +- 11 files changed, 568 insertions(+), 15 deletions(-) diff --git a/pcapkit/protocols/internet/internet.py b/pcapkit/protocols/internet/internet.py index 6e080b4c3..fe6af9279 100644 --- a/pcapkit/protocols/internet/internet.py +++ b/pcapkit/protocols/internet/internet.py @@ -145,13 +145,26 @@ def register(cls, code: 'Enum_TransType', protocol: 'ModuleDescriptor[Protocol] protocol: module descriptor or a :class:`~pcapkit.protocols.protocol.Protocol` subclass + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.Protocol` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this transport-layer + protocol number is already registered, naming the displaced + entry and its replacement so a caller can tell *what* was lost. + Fires on presence alone -- see :meth:`ProtocolBase.register + ` for why that + differs from ``register_protocol``. + """ if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not issubclass(protocol, Protocol): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'protocol {code} already registered, overwriting', RegistryWarning) + warn(f'protocol {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol ########################################################################## diff --git a/pcapkit/protocols/link/link.py b/pcapkit/protocols/link/link.py index 0aa3735a1..70bec0e84 100644 --- a/pcapkit/protocols/link/link.py +++ b/pcapkit/protocols/link/link.py @@ -125,13 +125,26 @@ def register(cls, code: 'Enum_EtherType', protocol: 'ModuleDescriptor[Protocol] protocol: module descriptor or a :class:`~pcapkit.protocols.protocol.Protocol` subclass + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.Protocol` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this EtherType is + already registered, naming the displaced entry and its + replacement so a caller can tell *what* was lost. Fires on + presence alone -- see :meth:`ProtocolBase.register + ` for why that + differs from ``register_protocol``. + """ if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not issubclass(protocol, Protocol): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'protocol {code} already registered, overwriting', RegistryWarning) + warn(f'protocol {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol ########################################################################## diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index 03a4fdfe2..2a05f037a 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -132,13 +132,26 @@ def register(cls, code: 'Enum_LinkType', protocol: 'ModuleDescriptor[Protocol] | protocol: module descriptor or a :class:`~pcapkit.protocols.protocol.Protocol` subclass + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.Protocol` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this link type is + already registered against PCAP frames, naming the displaced + entry and its replacement so a caller can tell *what* was lost. + Fires on presence alone -- see :meth:`ProtocolBase.register + ` for why that + differs from ``register_protocol``. + """ if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not issubclass(protocol, Protocol): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'protocol {code} already registered, overwriting', RegistryWarning) + warn(f'protocol {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol def index(self, name: 'str | Protocol | Type[Protocol]') -> 'int': diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index e69bfedd6..a046e3075 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -866,13 +866,26 @@ def register(cls, code: 'Enum_LinkType', protocol: 'ModuleDescriptor[Protocol] | protocol: module descriptor or a :class:`~pcapkit.protocols.protocol.Protocol` subclass + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.Protocol` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this link type is + already registered against PCAP-NG blocks, naming the displaced + entry and its replacement so a caller can tell *what* was lost. + Note this registry is separate from the PCAP one, so + :func:`~pcapkit.foundation.registry.protocols.register_linktype` + writing to both cannot make either warn about the other. + """ if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not issubclass(protocol, Protocol): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'protocol {code} already registered, overwriting', RegistryWarning) + warn(f'protocol {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol @classmethod diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index bab56764e..c71cafee9 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -767,13 +767,43 @@ def register(cls, code: 'int', protocol: 'ModuleDescriptor | Type[ProtocolBase]' protocol: module descriptor or a :class:`~pcapkit.protocols.protocol.Protocol` subclass + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If ``code`` is already + registered. The warning names the displaced entry and its + replacement, so a caller can tell *what* was lost rather than + only that something was. + + Note: + The guard fires on the mere presence of ``code``, including when the + incumbent and the replacement denote the same protocol. That is + deliberate, and differs from :func:`register_protocol + `, which + additionally requires the incumbent to be a *different* class. Two + things separate them. This + registry is keyed on a ``code`` the caller supplies, independently of + the value, so registering one class under two codes yields two keys + and never reaches the same key twice -- the spurious-warning case + that motivated the narrower guard cannot arise here, while a repeat + call for one code is a caller mistake worth reporting even when the + value is unchanged. And the incumbent may still be an unresolved + :class:`~pcapkit.corekit.module.ModuleDescriptor` while the + replacement is the very class it names, so "a different class" is not + decidable here without resolving the descriptor -- forcing the import + that the descriptor exists to defer, purely to decide whether to + warn. + """ if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not issubclass(protocol, ProtocolBase): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'protocol {code} already registered, overwriting', RegistryWarning) + warn(f'protocol {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol @classmethod diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index 7b3b63c56..183dc9082 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -16,7 +16,8 @@ from pcapkit.utilities.compat import Mapping from pcapkit.utilities.decorators import prepare from pcapkit.utilities.exceptions import NoDefaultValue, ProtocolUnbound, SchemaError, stacklevel -from pcapkit.utilities.warnings import SchemaWarning, UnknownFieldWarning, warn +from pcapkit.utilities.warnings import (RegistryWarning, SchemaWarning, UnknownFieldWarning, + warn) if TYPE_CHECKING: from collections import OrderedDict @@ -1074,6 +1075,16 @@ def __init_subclass__(cls, /, code: 'Optional[_ET | Iterable[_ET]]' = None, *arg :attr:`registry` mapping with the given ``code``. If ``code`` is not given, the subclass will not be registered. + Warns: + pcapkit.utilities.warnings.RegistryWarning: If any of ``code`` is + already registered, naming the displaced schema and its + replacement. This is the same guard :meth:`register` applies, + and it is here as well because a class declaration is the + *other* way into :attr:`__enum__` -- ``class MyOption(Option, + code=...)`` writes the registry without any call to + :meth:`register`, so guarding only the method would leave the + declaration path silently displacing a built-in schema. + Notes: If :attr:`__enum__` is not yet defined at function call, it will automatically be defined as a :class:`_EnumRegistry` @@ -1104,11 +1115,18 @@ def __init_subclass__(cls, /, code: 'Optional[_ET | Iterable[_ET]]' = None, *arg cls.__enum__ = _EnumRegistry(getattr(manual, 'default_factory', None), manual) if code is not None: - if isinstance(code, collections.abc.Iterable): - for _code in code: - cls.__enum__[_code] = (cls) # type: ignore[index] - else: - cls.__enum__[code] = (cls) # type: ignore[index] + # One loop over both shapes, so the overwrite guard below is written + # once rather than once per branch. ``register`` cannot be delegated + # to here: :class:`pcapkit.protocols.schema.misc.pcapng.Option` + # overrides it with an incompatible signature, so ``cls.register`` + # does not mean the same thing for every subclass. + codes = code if isinstance(code, collections.abc.Iterable) else (code,) + for _code in codes: + if _code in cls.__enum__: + incumbent = cls.__enum__[_code] # type: ignore[index] + warn(f'schema {_code} already registered, overwriting ' + f'{incumbent!r} with {cls!r}', RegistryWarning) + cls.__enum__[_code] = (cls) # type: ignore[index] super().__init_subclass__() @classmethod @@ -1119,5 +1137,47 @@ def register(cls, code: '_ET', schema: 'Type[Self]') -> 'None': code: Enumetaion code. schema: Enumetaion schema. + Warns: + pcapkit.utilities.warnings.RegistryWarning: If ``code`` is already + registered, naming the displaced schema and its replacement. + + Note: + Every public registrar in + :mod:`pcapkit.foundation.registry.protocols` that accepts a + ``schema`` registers two halves of one binding -- a parser class + through e.g. :meth:`IPv4.register_option + `, and a + schema class through this method. The parser half has warned on an + overwrite for as long as it has existed; this half assigned bare, so + one ``register_ipv4_option`` call replacing a built-in reported the + parser it displaced and said nothing about the schema. The guard + here closes that asymmetry. + + It fires on the mere presence of ``code``, as the code-keyed parser + registries do and unlike :func:`register_protocol + `, whose key + is derived from the value it stores. ``code`` here is supplied by the + caller and is independent of ``schema``, so a repeat is a caller + mistake worth reporting even when the value is unchanged. + + Presence is a faithful "was this really registered" test only because + :class:`_EnumRegistry` returns a miss without recording it. A plain + :class:`collections.defaultdict` would have inserted + :attr:`__default__` the first time any unregistered ``code`` was + looked up, so parsing a single packet carrying an unknown code would + have made the next legitimate registration for that code warn about + an entry no caller ever asked for -- the defect fixed for this layer + in #555, and for the parser-layer ``__proto__`` family in #421 and + #425/#428. That fix is what makes this guard safe to add. + + :class:`pcapkit.protocols.schema.misc.pcapng.Option` overrides this + method with a namespaced registry of its own and does not delegate + here, so it is guarded separately. + """ + if code in cls.__enum__: + incumbent = cls.__enum__[code] # type: ignore[index] + warn(f'schema {code} already registered, overwriting ' + f'{incumbent!r} with {schema!r}', RegistryWarning) + cls.__enum__[code] = schema # type: ignore[index] diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index b70e9b181..d2ec503b7 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -609,14 +609,26 @@ def register(cls, code: 'Enum_PayloadProtocolIdentifier | int', protocol: 'Modul its :attr:`self.__proto__ ` registry is keyed by PPID rather than by port number. + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.ProtocolBase` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this PPID is already + registered, naming the displaced entry and its replacement so a + caller can tell *what* was lost. Fires on presence alone, as the + port-keyed :meth:`Transport.register + ` it + overrides does. + """ if isinstance(protocol, ModuleDescriptor): protocol = protocol.klass if not issubclass(protocol, ProtocolBase): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'payload protocol identifier {code} already registered, overwriting', - RegistryWarning) + warn(f'payload protocol identifier {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol @classmethod diff --git a/pcapkit/protocols/transport/transport.py b/pcapkit/protocols/transport/transport.py index 6f347bf5f..5f5009906 100644 --- a/pcapkit/protocols/transport/transport.py +++ b/pcapkit/protocols/transport/transport.py @@ -85,6 +85,26 @@ def register(cls, code: 'int', protocol: 'ModuleDescriptor[Protocol] | Type[Prot protocol map should be associated directly with specific transport layer protocol type. + Raises: + pcapkit.utilities.exceptions.UnsupportedCall: If called on + :class:`Transport` itself. + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + :class:`~pcapkit.protocols.protocol.Protocol` subclass. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If this port is already + registered, naming the displaced entry and its replacement so a + caller can tell *what* was lost. Fires on presence alone -- see + :meth:`ProtocolBase.register + ` for why that + differs from ``register_protocol``. + + Note: + ``cls.__proto__`` belongs to the concrete protocol, not to + :class:`Transport`, so ``register_apptype`` reaching this method + twice for one call -- once as ``TCP``, once as ``UDP`` -- inspects + two different registries and cannot warn spuriously. + """ if cls is Transport: raise UnsupportedCall(f'{cls.__name__} is an abstract class') @@ -94,7 +114,8 @@ def register(cls, code: 'int', protocol: 'ModuleDescriptor[Protocol] | Type[Prot if not issubclass(protocol, Protocol): raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') if code in cls.__proto__: - warn(f'port {code} already registered, overwriting', RegistryWarning) + warn(f'port {code} already registered, overwriting ' + f'{cls.__proto__[code]!r} with {protocol!r}', RegistryWarning) cls.__proto__[code] = protocol @classmethod diff --git a/tests/foundation/registry/test_protocols.py b/tests/foundation/registry/test_protocols.py index d03c20a41..b732fb804 100644 --- a/tests/foundation/registry/test_protocols.py +++ b/tests/foundation/registry/test_protocols.py @@ -268,6 +268,86 @@ def test_sibling_registries_still_warn_on_an_identical_re_registration(self) -> self.assertEqual( len(self._registry_warnings(caught, RegistryWarning)), 1) + def test_sibling_registries_name_what_they_displaced(self) -> None: + """Every code-keyed registrar says *what* it overwrote, not just that it did. + + The presence-only condition is deliberately unchanged -- see the test + above -- but the message was ``'protocol {code} already registered, + overwriting'`` and stopped there, so a caller learned that something had + been displaced and never which class it was. That is the same diagnostic + gap #675 closed for :func:`~pcapkit.foundation.registry.protocols.\ + register_protocol`, and it is the part of #681 that does generalise. + + All seven code-keyed registrars are covered, including the two that reach + :meth:`Transport.register + ` through + :class:`~pcapkit.protocols.transport.tcp.TCP` and + :class:`~pcapkit.protocols.transport.udp.UDP` rather than overriding it, + and :meth:`ProtocolBase.register + ` itself, which the + other six shadow. + + """ + from pcapkit.const.reg.ethertype import EtherType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.const.reg.transtype import TransType + from pcapkit.const.sctp.payload_protocol_identifier import PayloadProtocolIdentifier + from pcapkit.protocols.link.link import Link + from pcapkit.protocols.internet.internet import Internet + from pcapkit.protocols.misc.null import NoPayload + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.protocols.transport.sctp import SCTP + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.protocols.transport.udp import UDP + from pcapkit.utilities.warnings import RegistryWarning + + UnitProtocol = self._unit_protocol() + + # The premise: two different classes, both acceptable to every gate. + self.assertIsNot(Raw, NoPayload) + + cases = ( + ('protocol-base', UnitProtocol, ProtocolBase.__dict__['__proto__'], 1), + ('link-ethertype', Link, Link.__dict__['__proto__'], + EtherType.Internet_Protocol_version_4), + ('internet-transtype', Internet, Internet.__dict__['__proto__'], + TransType.TCP), + ('pcap-frame-linktype', Frame, Frame.__dict__['__proto__'], + LinkType.ETHERNET), + ('pcapng-linktype', PCAPNG, PCAPNG.__dict__['__proto__'], + LinkType.ETHERNET), + ('tcp-port', TCP, TCP.__dict__['__proto__'], 80), + ('udp-port', UDP, UDP.__dict__['__proto__'], 53), + ('sctp-ppid', SCTP, SCTP.__dict__['__proto__'], + PayloadProtocolIdentifier.WebRTC_DCEP), + ) + for label, owner, registry, code in cases: + with self.subTest(registry=label): + self._guard_registry(registry, code) + + # Seed a known incumbent, so the assertion is about this test's + # own value rather than whatever the built-in table happens to + # hold -- several of these codes ship pre-seeded with an + # unresolved ``ModuleDescriptor``, whose ``repr`` is not the + # class's. + registry[code] = Raw + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + owner.register(code, NoPayload) + + self.assertIs(registry[code], NoPayload) + + messages = self._registry_warnings(caught, RegistryWarning) + self.assertEqual(len(messages), 1) + self.assertIn(repr(Raw), messages[0]) + self.assertIn(repr(NoPayload), messages[0]) + self.assertLess(messages[0].index(repr(Raw)), + messages[0].index(repr(NoPayload))) + def test_register_protocol_validates_and_updates_registry(self) -> None: from pcapkit.foundation.registry import protocols as registry from pcapkit.utilities.exceptions import RegistryError diff --git a/tests/protocols/schema/test_enum_schema_registry_unit.py b/tests/protocols/schema/test_enum_schema_registry_unit.py index 66b1d88f5..f180af154 100644 --- a/tests/protocols/schema/test_enum_schema_registry_unit.py +++ b/tests/protocols/schema/test_enum_schema_registry_unit.py @@ -18,6 +18,7 @@ import enum import importlib.util import unittest +import warnings from tests._support import purge_modules @@ -156,3 +157,283 @@ def test_real_tcp_option_schema_registry_does_not_leak_on_miss(self) -> None: self.assertIs(schema, UnassignedOption) self.assertNotIn(probe, Option.registry) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class EnumSchemaRegistryOverwriteTests(unittest.TestCase): + """The schema half of a registration reports an overwrite. + + Every public registrar in :mod:`pcapkit.foundation.registry.protocols` that + takes a ``schema`` registers two halves of one binding: a parser class, + through e.g. :meth:`~pcapkit.protocols.internet.ipv4.IPv4.register_option`, + and a schema class, through :meth:`EnumSchema.register`. The parser half has + warned on an overwrite for as long as it has existed; the schema half + assigned bare. So one ``register_ipv4_option`` call replacing a built-in + named the parser it displaced and said nothing about the schema -- half a + report for one call. + + The declaration path is covered as well, because ``class MyOption(Option, + code=...)`` reaches :attr:`EnumSchema.__enum__` without any call to + :meth:`register`; guarding only the method would leave it silent. + + Generalises the guard GitHub issue #675 added to ``register_protocol`` in + #681. The condition here is presence alone, as the code-keyed parser + registries use, rather than #681's "present *and* a different class" -- that + narrower form is licensed by a key *derived* from the value, which this + registry's caller-supplied ``code`` is not. + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def _guard_registry(self, registry, key) -> None: + """Restore ``key`` in ``registry`` on teardown, absence included. + + Mirrors the helper of the same name in + ``tests/foundation/registry/test_protocols.py``. Every registry in this + package is process-global and order-dependent, so a test that writes into + one has to put back exactly what it found -- and a key that was *absent* + has no value to restore, so writing one back would leave a stray entry + for the next test in the process to inherit. That is the class of defect + #674/#686 was about, hence the sentinel and the ``pop``. + + ``tests/_support.py`` deliberately offers nothing for this: its own + ``restore_modules`` docstring notes that it restores which module object a + name refers to, not the *contents* of one, so a registry mutated in place + rebinds nothing and outlives the test unless undone here. + + """ + missing = object() + previous = registry.get(key, missing) + + def restore() -> None: + if previous is missing: + registry.pop(key, None) + else: + registry[key] = previous + + self.addCleanup(restore) + + @staticmethod + def _registry_warnings(caught, category) -> 'list[str]': + """The messages of the captured warnings that are of ``category``. + + Filtering by category matters: the parse and import paths raise other + warning types, and a bare count of everything captured would make an + assertion about "no warning" pass or fail for unrelated reasons. + + """ + return [str(item.message) for item in caught + if issubclass(item.category, category)] + + @staticmethod + def _base_schema(): + """A fresh ``EnumSchema`` hierarchy with its own registry. + + Locally declared, so nothing here touches a process-global registry -- + :meth:`EnumSchema.__init_subclass__` builds a new ``__enum__`` for the + first subclass in a chain. + + """ + from pcapkit.protocols.schema.schema import EnumSchema + + class Code(enum.IntEnum): + one = 1 + two = 2 + three = 3 + + class DefaultSchema: + """Stand-in for the fallback schema.""" + + class BaseSchema(EnumSchema[Code]): + __default__ = lambda: DefaultSchema # noqa: E731 + + return Code, BaseSchema + + def test_register_warns_when_a_code_is_already_taken(self) -> None: + """The schema half now reports what it displaced. + + Before this change the body was a bare ``cls.__enum__[code] = schema``: + the replacement happened, the read-side fell back to the new class, and + nothing connected that to the registration which caused it. + + """ + from pcapkit.utilities.warnings import RegistryWarning + + Code, BaseSchema = self._base_schema() + + class Incumbent(BaseSchema, code=Code.one): + pass + + class Replacement(BaseSchema): + pass + + # The premise: two *different* schemas, and the code really is taken. + self.assertIsNot(Incumbent, Replacement) + self.assertIs(BaseSchema.registry[Code.one], Incumbent) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + BaseSchema.register(Code.one, Replacement) + + # The overwrite still happens. This reports the collision, it does not + # refuse it -- refusing would break the documented ability to replace a + # built-in schema. + self.assertIs(BaseSchema.registry[Code.one], Replacement) + + messages = self._registry_warnings(caught, RegistryWarning) + self.assertEqual(len(messages), 1) + + # Naming both is the point: 'schema 1 already registered' on its own does + # not say which schema was lost. Ordering asserted too, so the message + # cannot name them the wrong way round. + self.assertIn(repr(Incumbent), messages[0]) + self.assertIn(repr(Replacement), messages[0]) + self.assertLess(messages[0].index(repr(Incumbent)), + messages[0].index(repr(Replacement))) + + def test_register_stays_quiet_for_a_free_code(self) -> None: + """A first registration displaces nothing and must not warn. + + Pins the other half of the guard. A registrar that warned here would + make every legitimate ``register_*`` call noisy, and the wholesale + ``RegistryWarning`` filter that invites is what would then hide a real + collision. + + """ + from pcapkit.utilities.warnings import RegistryWarning + + Code, BaseSchema = self._base_schema() + + class Replacement(BaseSchema): + pass + + self.assertNotIn(Code.two, BaseSchema.registry) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + BaseSchema.register(Code.two, Replacement) + + self.assertIs(BaseSchema.registry[Code.two], Replacement) + self.assertEqual(self._registry_warnings(caught, RegistryWarning), []) + + def test_a_lookup_miss_still_does_not_make_a_later_register_warn(self) -> None: + """The #555 retention fix is what makes a presence-only guard safe here. + + On a plain :class:`collections.defaultdict` a bare ``registry[code]`` for + an unregistered code inserted the default, so parsing one packet carrying + an unknown code would make the next legitimate registration for that code + warn about an entry no caller ever asked for. This asserts the two fixes + compose: read a miss, then register that code, and stay silent. + + """ + from pcapkit.utilities.warnings import RegistryWarning + + Code, BaseSchema = self._base_schema() + + class Replacement(BaseSchema): + pass + + # The miss, exactly as every schema-layer call site performs it. + BaseSchema.registry[Code.three] + self.assertNotIn(Code.three, BaseSchema.registry) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + BaseSchema.register(Code.three, Replacement) + + self.assertEqual(self._registry_warnings(caught, RegistryWarning), []) + + def test_declaring_a_subclass_over_a_taken_code_warns(self) -> None: + """The declaration path reaches the registry without calling ``register``. + + :meth:`EnumSchema.__init_subclass__` assigns ``cls.__enum__[code]`` + directly, so a guard on :meth:`register` alone would leave + ``class MyOption(Option, code=...)`` -- the documented way to add a schema + -- silently displacing a built-in. + + """ + from pcapkit.utilities.warnings import RegistryWarning + + Code, BaseSchema = self._base_schema() + + class Incumbent(BaseSchema, code=Code.one): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + class Replacement(BaseSchema, code=Code.one): + pass + + self.assertIs(BaseSchema.registry[Code.one], Replacement) + + messages = self._registry_warnings(caught, RegistryWarning) + self.assertEqual(len(messages), 1) + self.assertIn(repr(Incumbent), messages[0]) + self.assertIn(repr(Replacement), messages[0]) + + def test_declaring_a_subclass_with_fresh_codes_stays_quiet(self) -> None: + """Every ordinary schema declaration must stay silent. + + This is the case that governs whether ``import pcapkit`` is noisy: 326 + registry writes happen during import, and a guard that warned on a + first-time declaration would fire on the great majority of them. Covers + the iterable form of ``code`` as well, which is the branch that shares + the guard. + + """ + from pcapkit.utilities.warnings import RegistryWarning + + Code, BaseSchema = self._base_schema() + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + class Single(BaseSchema, code=Code.one): + pass + + class Several(BaseSchema, code=(Code.two, Code.three)): + pass + + self.assertEqual(self._registry_warnings(caught, RegistryWarning), []) + + # The iterable branch was restructured into a single loop to carry the + # guard once; assert it still registers every code it was given. + self.assertIs(BaseSchema.registry[Code.one], Single) + self.assertIs(BaseSchema.registry[Code.two], Several) + self.assertIs(BaseSchema.registry[Code.three], Several) + + def test_the_guard_holds_on_a_real_shipped_registry(self) -> None: + """The same thing on :class:`...schema.transport.tcp.Option`. + + The tests above build a local hierarchy, which proves the mechanism but + not that it is reachable on a registry the package actually ships. This + one displaces a real built-in schema and puts it back. + + """ + from pcapkit.const.tcp.option import Option as OptionNumber + from pcapkit.protocols.schema.transport.tcp import Option + from pcapkit.utilities.warnings import RegistryWarning + + code = OptionNumber.Maximum_Segment_Size + incumbent = Option.registry[code] + + # Asserted rather than assumed: if this code were somehow unregistered, + # the warning below would not fire and the test would be vacuous. + self.assertIn(code, Option.registry) + self._guard_registry(Option.registry, code) + + class Replacement(Option): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + Option.register(code, Replacement) + + self.assertIs(Option.registry[code], Replacement) + + messages = self._registry_warnings(caught, RegistryWarning) + self.assertEqual(len(messages), 1) + self.assertIn(repr(incumbent), messages[0]) + self.assertIn(repr(Replacement), messages[0]) diff --git a/tests/protocols/schema/test_schema_unit.py b/tests/protocols/schema/test_schema_unit.py index c00d12f24..bd4d897c1 100644 --- a/tests/protocols/schema/test_schema_unit.py +++ b/tests/protocols/schema/test_schema_unit.py @@ -5,6 +5,7 @@ import importlib.util import io import unittest +import warnings from unittest import mock from tests._support import purge_modules, time_limit @@ -260,7 +261,23 @@ class ManySchema(BaseEnumSchema, code=[Code.two, Code.three]): self.assertIs(BaseEnumSchema.registry[Code.two], ManySchema) self.assertIs(BaseEnumSchema.registry[Code.three], ManySchema) - BaseEnumSchema.register(Code.two, OneSchema) + # ``Code.two`` is held by ``ManySchema``, so this is a genuine overwrite + # and now reports one. Captured and asserted rather than left to escape: + # an unasserted warning is noise in every later run of the suite, and the + # capture is what stops this line from quietly becoming a second, silent + # copy of the behaviour ``EnumSchemaRegistryOverwriteTests`` pins. + from pcapkit.utilities.warnings import RegistryWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + BaseEnumSchema.register(Code.two, OneSchema) + + overwrites = [str(item.message) for item in caught + if issubclass(item.category, RegistryWarning)] + self.assertEqual(len(overwrites), 1) + self.assertIn(repr(ManySchema), overwrites[0]) + self.assertIn(repr(OneSchema), overwrites[0]) + self.assertIs(BaseEnumSchema.registry[Code.two], OneSchema) self.assertIs(BaseEnumSchema.from_dict().registry, BaseEnumSchema.registry)