diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index e0a4dee839..4fbfa860bd 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -471,8 +471,18 @@ def register_frame(cls, code: 'Enum_Frame', meth: 'str | tuple[FrameParser, Fram code: HTTP frame type code. meth: Method name or callable to parse and/or construct the frame. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__frame__: + incumbent = cls.__frame__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'HTTP/2: [Type {code}] frame already registered', RegistryWarning) cls.__frame__[code] = meth diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 79dca4159e..8935ebd50e 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -694,8 +694,18 @@ def register_parameter(cls, code: 'Enum_Parameter', meth: 'str | tuple[Parameter code: HIP parameter code. meth: Method name or callable to parse and/or construct the parameter. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__parameter__: + incumbent = cls.__parameter__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'parameter {code} already registered, overwriting', RegistryWarning) cls.__parameter__[code] = meth diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index b9cc1e6911..88c11284c2 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -383,8 +383,18 @@ def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, O code: HOPOPT option code. meth: Method name or callable to parse and/or construct the option. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__option__: + incumbent = cls.__option__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index e1ece1e837..9687002b40 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -510,8 +510,18 @@ def register_option(cls, code: 'Enum_OptionNumber', meth: 'str | tuple[OptionPar code: IPv4 option code. meth: Method name or callable to parse and/or construct the option. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__option__: + incumbent = cls.__option__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 998db90b02..4375fc76bd 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -375,8 +375,18 @@ def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, O code: IPv6-Opts option code. meth: Method name or callable to parse and/or construct the option. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__option__: + incumbent = cls.__option__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index 095be49ffa..e2235bd5cb 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -353,8 +353,18 @@ def register_routing(cls, code: 'Enum_Routing', meth: 'str | tuple[TypeParser, T code: IPv6-Route data type code. meth: Method name or callable to parse and/or construct the data. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__routing__: + incumbent = cls.__routing__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'routing {code} already registered, overwriting', RegistryWarning) cls.__routing__[code] = meth diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index ec868ba929..90379becbd 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1520,8 +1520,18 @@ def register_message(cls, code: 'Enum_Packet', meth: 'str | tuple[PacketParser, code: MH message type code. meth: Method name or callable to parse and/or construct the message. - """ - if code in cls.__message__: + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + + """ + incumbent = cls.__message__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'message type {code} already registered, overwriting', RegistryWarning) cls.__message__[code] = meth @@ -1533,8 +1543,18 @@ def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, O code: MH option code. meth: Method name or callable to parse and/or construct the option. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__option__: + incumbent = cls.__option__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth @@ -1546,8 +1566,18 @@ def register_extension(cls, code: 'Enum_CGAExtension', meth: 'str | tuple[Extens code: CGA extension code. meth: Method name or callable to parse and/or construct the extension. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__extension__: + incumbent = cls.__extension__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'extension {code} already registered, overwriting', RegistryWarning) cls.__extension__[code] = meth diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index fc65f8f1d9..fa0d3f075d 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -641,8 +641,18 @@ def register_chunk(cls, code: 'Enum_Chunk', meth: 'str | tuple[ChunkParser, Chun code: SCTP chunk type. meth: Method name or callable to parse and/or construct the chunk. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__chunk__: + incumbent = cls.__chunk__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'chunk {code} already registered, overwriting', RegistryWarning) cls.__chunk__[code] = meth @@ -654,8 +664,18 @@ def register_parameter(cls, code: 'Enum_Parameter', meth: 'str | tuple[Parameter code: SCTP chunk parameter type. meth: Method name or callable to parse and/or construct the parameter. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__parameter__: + incumbent = cls.__parameter__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'parameter {code} already registered, overwriting', RegistryWarning) cls.__parameter__[code] = meth @@ -667,8 +687,18 @@ def register_cause(cls, code: 'Enum_CauseCode', meth: 'str | tuple[CauseParser, code: SCTP error cause code. meth: Method name or callable to parse and/or construct the cause. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__cause__: + incumbent = cls.__cause__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'error cause {code} already registered, overwriting', RegistryWarning) cls.__cause__[code] = meth diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 4c89d8d271..2fe44ecd35 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -646,8 +646,18 @@ def register_option(cls, code: 'Enum_Option', meth: 'str | tuple[OptionParser, O code: TCP option code. meth: Method name or callable to parse and/or construct the option. - """ - if code in cls.__option__: + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + + """ + incumbent = cls.__option__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__option__[code] = meth @@ -659,8 +669,18 @@ def register_mp_option(cls, code: 'Enum_MPTCPOption', meth: 'str | tuple[MPOptio code: MPTCP option code. meth: Method name or callable to parse and/or construct the option. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__mp_option__: + incumbent = cls.__mp_option__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'option {code} already registered, overwriting', RegistryWarning) cls.__mp_option__[code] = meth diff --git a/tests/protocols/application/test_http_unit.py b/tests/protocols/application/test_http_unit.py index 7f5064279d..34d19ad9f9 100644 --- a/tests/protocols/application/test_http_unit.py +++ b/tests/protocols/application/test_http_unit.py @@ -1839,7 +1839,7 @@ def test_httpv2_id_length_make_bytes_and_register_frame_warning(self) -> None: try: schema = proto.make(type=Frame.DATA, flags=0x01, sid=3, frame=b'data') with mock.patch('pcapkit.protocols.application.httpv2.warn') as warn: - HTTPv2.register_frame(Frame.DATA, 'data') + HTTPv2.register_frame(Frame.DATA, 'headers') self.assertEqual(HTTPv2.id(), ('HTTP', 'HTTPv2')) self.assertEqual(proto.alias, 'HTTP/2') diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index d5afc0a1f7..b6006a2869 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -239,9 +239,9 @@ def test_hip_register_parameter_warns_on_overwrite(self) -> None: original = registry[Parameter.ESP_INFO] try: with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: - HIP.register_parameter(Parameter.ESP_INFO, 'esp_info') + HIP.register_parameter(Parameter.ESP_INFO, 'unassigned') warn.assert_called_once() - self.assertEqual(registry[Parameter.ESP_INFO], 'esp_info') + self.assertEqual(registry[Parameter.ESP_INFO], 'unassigned') finally: registry[Parameter.ESP_INFO] = original @@ -257,8 +257,13 @@ def test_hip_register_parameter_warns_on_overwrite(self) -> None: warn.assert_not_called() self.assertEqual(registry[custom], 'unassigned') + # Handing back the entry just stored is no overwrite either (#1362). with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: HIP.register_parameter(custom, 'unassigned') + warn.assert_not_called() + + with mock.patch('pcapkit.protocols.internet.hip.warn') as warn: + HIP.register_parameter(custom, 'esp_info') warn.assert_called_once() finally: registry.pop(custom, None) diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index dd4e2684e6..5de94d1857 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -1102,9 +1102,9 @@ def test_ipv4_register_option_warns_on_overwrite(self) -> None: original = registry[OptionNumber.EOOL] try: with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: - IPv4.register_option(OptionNumber.EOOL, 'eool') + IPv4.register_option(OptionNumber.EOOL, 'nop') warn.assert_called_once() - self.assertEqual(registry[OptionNumber.EOOL], 'eool') + self.assertEqual(registry[OptionNumber.EOOL], 'nop') finally: registry[OptionNumber.EOOL] = original @@ -1120,8 +1120,13 @@ def test_ipv4_register_option_warns_on_overwrite(self) -> None: warn.assert_not_called() self.assertEqual(registry[custom], 'unassigned') + # Handing back the entry just stored is no overwrite either (#1362). with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: IPv4.register_option(custom, 'unassigned') + warn.assert_not_called() + + with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: + IPv4.register_option(custom, 'nop') warn.assert_called_once() finally: registry.pop(custom, None) diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 5c09a687a1..c86a4c8e00 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -63,9 +63,9 @@ def test_hopopt_register_option_warns_on_overwrite(self) -> None: original = HOPOPT.__dict__['__option__'][Option.Pad1] try: with mock.patch('pcapkit.protocols.internet.hopopt.warn') as warn: - HOPOPT.register_option(Option.Pad1, 'pad') + HOPOPT.register_option(Option.Pad1, 'tun') warn.assert_called_once() - self.assertEqual(HOPOPT.__dict__['__option__'][Option.Pad1], 'pad') + self.assertEqual(HOPOPT.__dict__['__option__'][Option.Pad1], 'tun') finally: HOPOPT.__dict__['__option__'][Option.Pad1] = original @@ -76,9 +76,9 @@ def test_ipv6_opts_register_option_warns_on_overwrite(self) -> None: original = IPv6_Opts.__dict__['__option__'][Option.Pad1] try: with mock.patch('pcapkit.protocols.internet.ipv6_opts.warn') as warn: - IPv6_Opts.register_option(Option.Pad1, 'pad') + IPv6_Opts.register_option(Option.Pad1, 'tun') warn.assert_called_once() - self.assertEqual(IPv6_Opts.__dict__['__option__'][Option.Pad1], 'pad') + self.assertEqual(IPv6_Opts.__dict__['__option__'][Option.Pad1], 'tun') finally: IPv6_Opts.__dict__['__option__'][Option.Pad1] = original @@ -89,9 +89,9 @@ def test_ipv6_route_register_routing_warns_on_overwrite(self) -> None: original = IPv6_Route.__dict__['__routing__'][Routing.Source_Route] try: with mock.patch('pcapkit.protocols.internet.ipv6_route.warn') as warn: - IPv6_Route.register_routing(Routing.Source_Route, 'src') + IPv6_Route.register_routing(Routing.Source_Route, 'rpl') warn.assert_called_once() - self.assertEqual(IPv6_Route.__dict__['__routing__'][Routing.Source_Route], 'src') + self.assertEqual(IPv6_Route.__dict__['__routing__'][Routing.Source_Route], 'rpl') finally: IPv6_Route.__dict__['__routing__'][Routing.Source_Route] = original diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 885748a790..dbae981fda 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -54,9 +54,9 @@ def test_mh_register_message_warns_on_overwrite(self) -> None: original = MH.__dict__['__message__'][Packet.Binding_Refresh_Request] try: with mock.patch('pcapkit.protocols.internet.mh.warn') as warn: - MH.register_message(Packet.Binding_Refresh_Request, 'brr') + MH.register_message(Packet.Binding_Refresh_Request, 'hoti') warn.assert_called_once() - self.assertEqual(MH.__dict__['__message__'][Packet.Binding_Refresh_Request], 'brr') + self.assertEqual(MH.__dict__['__message__'][Packet.Binding_Refresh_Request], 'hoti') finally: MH.__dict__['__message__'][Packet.Binding_Refresh_Request] = original @@ -67,9 +67,9 @@ def test_mh_register_option_warns_on_overwrite(self) -> None: original = MH.__dict__['__option__'][Option.Pad1] try: with mock.patch('pcapkit.protocols.internet.mh.warn') as warn: - MH.register_option(Option.Pad1, 'pad') + MH.register_option(Option.Pad1, 'bra') warn.assert_called_once() - self.assertEqual(MH.__dict__['__option__'][Option.Pad1], 'pad') + self.assertEqual(MH.__dict__['__option__'][Option.Pad1], 'bra') finally: MH.__dict__['__option__'][Option.Pad1] = original @@ -80,9 +80,9 @@ def test_mh_register_extension_warns_on_overwrite(self) -> None: original = MH.__dict__['__extension__'][CGAExtension.Multi_Prefix] try: with mock.patch('pcapkit.protocols.internet.mh.warn') as warn: - MH.register_extension(CGAExtension.Multi_Prefix, 'multiprefix') + MH.register_extension(CGAExtension.Multi_Prefix, 'exp') warn.assert_called_once() - self.assertEqual(MH.__dict__['__extension__'][CGAExtension.Multi_Prefix], 'multiprefix') + self.assertEqual(MH.__dict__['__extension__'][CGAExtension.Multi_Prefix], 'exp') finally: MH.__dict__['__extension__'][CGAExtension.Multi_Prefix] = original diff --git a/tests/protocols/test_parser_registrar_identity_unit.py b/tests/protocols/test_parser_registrar_identity_unit.py new file mode 100644 index 0000000000..5bcaeedbe3 --- /dev/null +++ b/tests/protocols/test_parser_registrar_identity_unit.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +"""GitHub issues #1362 and #1363: the parser registrars warn only on a real displacement. + +The registrars that store a parser -- a method name or a ``(parser, +constructor)`` pair -- tested only whether ``code`` was present, so handing back +the entry already stored warned that it was being overwritten. They now compare +the incumbent with the replacement, as the class-storing registrars do since +#718 and #739: an equal entry is a silent no-op. Under the #1363 ruling, the +entry a code shipped with is accepted back after an override, which undoes it. + +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 importlib +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class + +#: ``(module, class, registrar, registry)`` for every parser registrar but +#: PCAP-NG's, which another change owns. +SITES = ( + ('pcapkit.protocols.internet.ipv4', 'IPv4', 'register_option', '__option__'), + ('pcapkit.protocols.internet.hopopt', 'HOPOPT', 'register_option', '__option__'), + ('pcapkit.protocols.internet.ipv6_opts', 'IPv6_Opts', 'register_option', '__option__'), + ('pcapkit.protocols.internet.ipv6_route', 'IPv6_Route', 'register_routing', '__routing__'), + ('pcapkit.protocols.internet.hip', 'HIP', 'register_parameter', '__parameter__'), + ('pcapkit.protocols.internet.mh', 'MH', 'register_message', '__message__'), + ('pcapkit.protocols.internet.mh', 'MH', 'register_option', '__option__'), + ('pcapkit.protocols.internet.mh', 'MH', 'register_extension', '__extension__'), + ('pcapkit.protocols.transport.tcp', 'TCP', 'register_option', '__option__'), + ('pcapkit.protocols.transport.tcp', 'TCP', 'register_mp_option', '__mp_option__'), + ('pcapkit.protocols.transport.sctp', 'SCTP', 'register_chunk', '__chunk__'), + ('pcapkit.protocols.transport.sctp', 'SCTP', 'register_parameter', '__parameter__'), + ('pcapkit.protocols.transport.sctp', 'SCTP', 'register_cause', '__cause__'), + ('pcapkit.protocols.application.httpv2', 'HTTP', 'register_frame', '__frame__'), +) + + +def _parse(*args: object, **kwargs: object) -> None: + """Stand-in parser.""" + + +def _make(*args: object, **kwargs: object) -> None: + """Stand-in constructor.""" + + +class TestParserRegistrarIdentity(unittest.TestCase): + """Pin when a parser registrar warns.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _each_site(self, check): # type: ignore[no-untyped-def] + """Run ``check(register, registry)`` for every site, each in its own ``subTest``.""" + for module, name, meth, attr in SITES: + cls = getattr(importlib.import_module(module), name) + registry = getattr(cls, attr) + with self.subTest(site=f'{name}.{meth}'), mock.patch.dict(registry): + check(getattr(cls, meth), registry) + + def _count(self, call, *args) -> int: # type: ignore[no-untyped-def] + """Run ``call(*args)`` and return how many registry warnings it raised.""" + from pcapkit.utilities.warnings import RegistryWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + call(*args) + return sum(issubclass(item.category, RegistryWarning) for item in caught) + + def test_reregistering_the_stored_entry_is_silent(self) -> None: + def check(register, registry): # type: ignore[no-untyped-def] + for code, value in list(registry.items()): + self.assertEqual(self._count(register, code, value), 0, code) + self.assertIs(registry[code], value) + # An equal method name built at run time is the same entry, and + # the stored object is kept. + if isinstance(value, str): + self.assertEqual(self._count(register, code, ''.join(list(value))), 0, code) + self.assertIs(registry[code], value) + self._each_site(check) + + def test_reregistering_a_parser_pair_is_silent(self) -> None: + def check(register, registry): # type: ignore[no-untyped-def] + code = next(iter(registry)) + pair = (_parse, _make) + self.assertEqual(self._count(register, code, pair), 1) + self.assertIs(registry[code], pair) + self.assertEqual(self._count(register, code, pair), 0) + self.assertEqual(self._count(register, code, (_parse, _make)), 0) + self.assertIs(registry[code], pair) + self._each_site(check) + + def test_restoring_the_shipped_entry_undoes_an_override(self) -> None: + def check(register, registry): # type: ignore[no-untyped-def] + code, shipped = next(iter(registry.items())) + self.assertEqual(self._count(register, code, 'w1362_override'), 1) + self.assertEqual(registry[code], 'w1362_override') + # Putting the shipped entry back displaces the override, so it warns. + self.assertEqual(self._count(register, code, shipped), 1) + self.assertIs(registry[code], shipped) + self.assertEqual(self._count(register, code, shipped), 0) + self._each_site(check) + + def test_a_foreign_entry_still_warns(self) -> None: + def check(register, registry): # type: ignore[no-untyped-def] + code, shipped = next(iter(registry.items())) + self.assertEqual(self._count(register, code, (_parse, _make)), 1) + self.assertEqual(self._count(register, code, f'{shipped}_w1362'), 1) + self.assertEqual(registry[code], f'{shipped}_w1362') + self._each_site(check) + + def test_an_unregistered_code_is_silent(self) -> None: + def check(register, registry): # type: ignore[no-untyped-def] + code = next(iter(registry)) + registry.pop(code) + self.assertEqual(self._count(register, code, 'w1362_new'), 0) + self.assertEqual(registry[code], 'w1362_new') + self._each_site(check) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/transport/test_tcp_udp_unit.py b/tests/protocols/transport/test_tcp_udp_unit.py index 4337e07c30..915c027502 100644 --- a/tests/protocols/transport/test_tcp_udp_unit.py +++ b/tests/protocols/transport/test_tcp_udp_unit.py @@ -322,9 +322,9 @@ def test_tcp_register_option_warns_on_overwrite(self) -> None: original = TCP.__dict__['__option__'][Option.End_of_Option_List] try: with mock.patch('pcapkit.protocols.transport.tcp.warn') as warn: - TCP.register_option(Option.End_of_Option_List, 'eool') + TCP.register_option(Option.End_of_Option_List, 'nop') warn.assert_called_once() - self.assertEqual(TCP.__dict__['__option__'][Option.End_of_Option_List], 'eool') + self.assertEqual(TCP.__dict__['__option__'][Option.End_of_Option_List], 'nop') finally: TCP.__dict__['__option__'][Option.End_of_Option_List] = original @@ -335,9 +335,9 @@ def test_tcp_register_mptcp_option_warns_on_overwrite(self) -> None: original = TCP.__dict__['__mp_option__'][MPTCPOption.MP_CAPABLE] try: with mock.patch('pcapkit.protocols.transport.tcp.warn') as warn: - TCP.register_mp_option(MPTCPOption.MP_CAPABLE, 'capable') + TCP.register_mp_option(MPTCPOption.MP_CAPABLE, 'join') warn.assert_called_once() - self.assertEqual(TCP.__dict__['__mp_option__'][MPTCPOption.MP_CAPABLE], 'capable') + self.assertEqual(TCP.__dict__['__mp_option__'][MPTCPOption.MP_CAPABLE], 'join') finally: TCP.__dict__['__mp_option__'][MPTCPOption.MP_CAPABLE] = original