From 445e1ef4e56515a7f6a6b9377c442acf00ea767a Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 02:08:32 -0400 Subject: [PATCH] fix(protocols): drop the warnings of a next layer replaced by Raw (#1580) ProtocolBase._parse_next_layer parses a next layer on trial and keeps it as Raw when its header was not captured (#1170), but the trial's warnings were reported anyway. A PCAP-NG SPB, EPB or Packet Block, or a PCAP record, cut to 10 octets warned "packet length < 0" twice for an Ethernet header the result does not have, on parse, from_data and make. * pcapkit/utilities/warnings.py: hold_warnings, a ContextVar-backed block that holds warn() calls with the frame each names. On keep it replays them where it is left, so an error filter raises inside the same guard; on discard it drops them. It lets a BaseException through, restores by token, steps over blocks left out of order, and touches no filter. * pcapkit/protocols/protocol.py: _parse_next_layer holds only a layer captured shorter than its class's length hint and discards on the Raw fallback; every other layer is parsed holding nothing. * tests: tests/protocols/test_trial_parse_warnings_1580_unit.py. --- docs/source/pcapkit/utilities/functools.rst | 3 + docs/source/pcapkit/utilities/warnings.rst | 7 + pcapkit/protocols/protocol.py | 48 +- pcapkit/utilities/warnings.py | 176 ++++++- .../test_trial_parse_warnings_1580_unit.py | 473 ++++++++++++++++++ 5 files changed, 700 insertions(+), 7 deletions(-) create mode 100644 tests/protocols/test_trial_parse_warnings_1580_unit.py diff --git a/docs/source/pcapkit/utilities/functools.rst b/docs/source/pcapkit/utilities/functools.rst index b5b2114c90..256447377b 100644 --- a/docs/source/pcapkit/utilities/functools.rst +++ b/docs/source/pcapkit/utilities/functools.rst @@ -42,3 +42,6 @@ Error Handling Utilities .. autofunction:: pcapkit.utilities.exceptions.stacklevel .. autofunction:: pcapkit.utilities.warnings.warn + +.. autoclass:: pcapkit.utilities.warnings.hold_warnings + :members: discard, holding diff --git a/docs/source/pcapkit/utilities/warnings.rst b/docs/source/pcapkit/utilities/warnings.rst index 9d3f07671e..f864c74bf0 100644 --- a/docs/source/pcapkit/utilities/warnings.rst +++ b/docs/source/pcapkit/utilities/warnings.rst @@ -27,6 +27,13 @@ The counts are the same in development mode as outside it: the log record carries, never how many records there are. Constructing a warning class is not an act of reporting one -- it emits nothing and has no side effects. +A next layer captured shorter than its length hint is parsed on trial, and +replaced with :class:`~pcapkit.protocols.misc.raw.Raw` when its header was not +captured. The trial's warnings are held back by +:class:`~pcapkit.utilities.warnings.hold_warnings`: a replaced layer's are dropped +on both channels, since they describe a layer the result does not have, and a kept +layer's are reported as above when the trial ends (:issue:`1580`). + Silencing pcapkit Warnings -------------------------- diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 55d78814be..db087f35a5 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -50,7 +50,7 @@ from pcapkit.utilities.exceptions import (BaseError, ProtocolError, ProtocolNotFound, ProtocolNotImplemented, RegistryError, StructError, UnsupportedCall) -from pcapkit.utilities.warnings import RegistryWarning, UnknownFieldWarning, warn +from pcapkit.utilities.warnings import RegistryWarning, UnknownFieldWarning, hold_warnings, warn if TYPE_CHECKING: from enum import IntEnum as StdlibEnum @@ -389,6 +389,31 @@ def _check_construction_keywords(cls: 'type', kwargs: 'dict[str, Any]', UnknownFieldWarning) +def _short_of_hint(klass: 'Type[ProtocolBase]', captured: 'int') -> 'bool': + """Tell whether ``captured`` octets may be too few for a ``klass`` header. + + Args: + klass: Next layer protocol class. + captured: Octets captured for it. + + Returns: + Whether a next layer of ``klass`` parsed from ``captured`` octets may be + replaced with :class:`~pcapkit.protocols.misc.raw.Raw` by + :meth:`ProtocolBase._parse_next_layer`, and so has its warnings held. + + The hint is read from the class, before there is an instance to ask, which + is enough for every protocol shipped: each returns a constant. A hint that + needs an instance raises here instead, and the layer counts as short, which + only means its warnings are held until the parse is known to be kept. + + """ + try: + hint = klass.__length_hint__(cast('ProtocolBase', None)) + except Exception: # pylint: disable=broad-except + return True + return hint is not None and captured < hint + + class ProtocolMeta(abc.ABCMeta): """Metaclass of :class:`ProtocolBase`, and so of every protocol class. @@ -2283,13 +2308,24 @@ def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int' form rather than the shortest. A next layer that reports no hint is kept as parsed. - """ - next_ = klass(file_, length, **kwargs) # type: ignore[abstract] + The warnings of a parse that may be replaced are held until it is + known whether it is. They are dropped when it is, since they describe + a header the result does not have -- ``packet length < 0`` twice for + an Ethernet frame of 10 octets (:issue:`1580`) -- and reported as the + parse ends when it is kept, still inside the guard that catches them + under an ``error`` filter. A layer kept with its header cut short + still warns. A layer captured at least as long as its hint cannot be + replaced, and is parsed holding nothing, even inside one that holds. + """ captured = min(length, len(file_)) - length_hint = getattr(next_, '__length_hint__', None) - hint = None if length_hint is None else length_hint() - if hint is not None and captured < hint and captured < next_.length: + with hold_warnings(_short_of_hint(klass, captured)) as held: + next_ = klass(file_, length, **kwargs) # type: ignore[abstract] + + length_hint = getattr(next_, '__length_hint__', None) + hint = None if length_hint is None else length_hint() + held.discard = hint is not None and captured < hint and captured < next_.length + if held.discard: from pcapkit.protocols.misc.raw import Raw # isort: skip # pylint: disable=import-outside-toplevel next_ = Raw(file_, length, **kwargs) return next_ diff --git a/pcapkit/utilities/warnings.py b/pcapkit/utilities/warnings.py index 6fa7ff2b8f..dc282aa583 100644 --- a/pcapkit/utilities/warnings.py +++ b/pcapkit/utilities/warnings.py @@ -51,7 +51,13 @@ ``pcapkit`` logger is configured separately, e.g. with ``logging.getLogger('pcapkit').setLevel(logging.ERROR)``. +The one exception is a parse :mod:`pcapkit` runs on trial: :class:`hold_warnings` +holds back its warnings, on both channels, until the parse is kept, and drops +them if it is discarded (:issue:`1580`). + """ +import contextvars +import inspect import warnings from typing import TYPE_CHECKING @@ -59,7 +65,14 @@ from pcapkit.utilities.logging import VERBOSE, get_logger if TYPE_CHECKING: - from typing import Optional, Type, Union + from types import FrameType, TracebackType + from typing import List, Optional, Tuple, Type, Union + + from typing_extensions import Self + + #: A warning held back by :class:`hold_warnings`: its message, its category, + #: and the frame it is attributed to, if that frame could be found. + HeldWarning = Tuple[Union[str, Warning], Type[Warning], Optional[FrameType]] __all__ = [ 'warn', @@ -85,6 +98,12 @@ #: :data:`pcapkit.utilities.logging.logger`. logger = get_logger(__name__) +#: The innermost open :class:`hold_warnings` block, or :data:`None` outside any. +#: A context variable rather than a global, so that a parse in another thread or +#: task is neither held nor released by this one. +_HELD = contextvars.ContextVar( + 'pcapkit_held_warnings', default=None) # type: contextvars.ContextVar[Optional[hold_warnings]] + def warn(message: 'Union[str, Warning]', category: 'Type[Warning]', stacklevel: 'Optional[int]' = None) -> 'None': @@ -112,6 +131,11 @@ def warn(message: 'Union[str, Warning]', category: 'Type[Warning]', :mod:`pcapkit.utilities.warnings` for the emission model in full, and for how to silence either channel. + Note: + Inside a :class:`hold_warnings` block the warning is held back instead, + with the frame it is attributed to, and reported from there only if the + block keeps it. + """ if stacklevel is None: # Computed here, so the level is already relative to *this* frame, which @@ -126,11 +150,161 @@ def warn(message: 'Union[str, Warning]', category: 'Type[Warning]', # whole exercise exists to skip past. stacklevel += 1 + block = _open_block(_HELD.get()) + if block is not None and block.holding: + # Level 1 is this frame, so the frame named is ``stacklevel - 1`` out. + target = inspect.currentframe() + for _ in range(stacklevel - 1): + if target is None: + break + target = target.f_back + block._held.append((message, category, target)) # pylint: disable=protected-access + del target + return + logger.warning(message, exc_info=VERBOSE, stack_info=VERBOSE, stacklevel=stacklevel) warnings.warn(message, category, stacklevel) +def _open_block(block: 'Optional[hold_warnings]') -> 'Optional[hold_warnings]': + """The innermost open block from ``block`` outwards. + + Args: + block: Block to start from, usually the one :data:`_HELD` names. + + Returns: + ``block``, or the nearest block enclosing it that is still open. A block + is left closed but still named only when it was left out of order, or in + another context, where it could not unset itself in this one. + + """ + while block is not None and not block._open: # pylint: disable=protected-access + block = block._outer # pylint: disable=protected-access + return block + + +def _replay(message: 'Union[str, Warning]', category: 'Type[Warning]', + target: 'Optional[FrameType]') -> 'None': + """Report a warning :class:`hold_warnings` held back, as :func:`warn` would have. + + Args: + message: Warning message. + category: Warning category. + target: Frame the warning was attributed to when it was held. + + The warning is attributed to ``target`` again, found by walking out from this + frame. ``target`` lies outside the parse that held it whenever its level came + from :func:`~pcapkit.utilities.exceptions.stacklevel`, and so is still on the + stack; were it not, the level is computed afresh. It is reported here and + now, not held again by an enclosing block. + + """ + level = 1 # this frame, as both consumers below count it + frame = inspect.currentframe() + while frame is not None and frame is not target: + frame = frame.f_back + level += 1 + found = frame is not None + del frame + if not found: + level = stacklevel_calculator() + + logger.warning(message, exc_info=VERBOSE, stack_info=VERBOSE, stacklevel=level) + warnings.warn(message, category, level) + + +class hold_warnings: # pylint: disable=invalid-name + """Hold back the warnings a parse reports until it is known to be kept. + + Args: + hold: Hold the warnings reported inside the block. When false, they are + reported at once instead, even inside an enclosing block that holds. + + Every :func:`warn` call inside the block is held, with the frame it is + attributed to. On leaving the block they are reported in order, with the + same message and category and attributed to the same frame on both + channels, unless :attr:`discard` was set, in which case they are dropped. A + block left by an :exc:`Exception` reports them whatever :attr:`discard` + says; one left by any other :exc:`BaseException` -- :exc:`KeyboardInterrupt`, + :exc:`SystemExit`, :exc:`GeneratorExit` -- drops them, so that a filter + turning them into errors cannot replace it. + + The line named is the frame's line when the block is left. For the frame + :func:`~pcapkit.utilities.exceptions.stacklevel` names -- the caller of + :mod:`pcapkit`, still inside that call -- it is the line named when held. + + Blocks nest, and each reports what it keeps when it is left, never handing + it to an enclosing block. Under a filter turning warnings into errors, the + error is then raised where the block that kept them is left -- the end of + the trial parse, inside the guard it was raised in when nothing was held. + + :meth:`ProtocolBase._parse_next_layer + ` uses this for the + trial parse it replaces with :class:`~pcapkit.protocols.misc.raw.Raw` when the + header was not captured: the warnings describe a layer the result does not + have (:issue:`1580`). + + Note: + Nothing here touches :data:`warnings.filters` or a module's + ``__warningregistry__``; :func:`warnings.catch_warnings` would, which is + what made unrelated warnings re-fire up to v1.4.1 (see + :class:`BaseWarning`). + + A block left while one opened inside it is still open -- a generator + suspended inside its own block -- stays in place for that one, which + steps over it when it is left. While it is suspended, though, the + generator's block holds its caller's warnings too: a context variable + follows the thread, not the frame. + + """ + + __slots__ = ('discard', 'holding', '_held', '_outer', '_token', '_open') + + def __init__(self, hold: 'bool' = True) -> 'None': + #: Drop the held warnings on leaving the block, rather than report them. + self.discard = False + #: Whether warnings reported inside the block are held. + self.holding = hold + + self._held = [] # type: List[HeldWarning] + self._outer = None # type: Optional[hold_warnings] + self._token = None # type: Optional[contextvars.Token[Optional[hold_warnings]]] + self._open = False + + def __enter__(self) -> 'Self': + self._held = [] + self._outer = _HELD.get() + self._token = _HELD.set(self) + self._open = True + return self + + def __exit__(self, exc_type: 'Optional[Type[BaseException]]', + exc_value: 'Optional[BaseException]', + traceback: 'Optional[TracebackType]') -> 'None': + token, self._token = self._token, None + # a block opened inside this one and still open -- a suspended generator's + # -- stays in place, and steps over this one when it is left in turn + innermost = _open_block(_HELD.get()) is self + self._open = False + if innermost and token is not None: + try: + _HELD.reset(token) + except ValueError: # left in a copy of the context it was entered in + _HELD.set(self._outer) + outer = _open_block(_HELD.get()) + if outer is not _HELD.get(): + _HELD.set(outer) + + held, self._held = self._held, [] + if self.discard and exc_type is None: + return + if exc_type is not None and not issubclass(exc_type, Exception): + return + for message, category, target in held: + _replay(message, category, target) + + ############################################################################## # BaseWarning (abc of warnings) session. ############################################################################## diff --git a/tests/protocols/test_trial_parse_warnings_1580_unit.py b/tests/protocols/test_trial_parse_warnings_1580_unit.py new file mode 100644 index 0000000000..4caf9552b4 --- /dev/null +++ b/tests/protocols/test_trial_parse_warnings_1580_unit.py @@ -0,0 +1,473 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1580: a next layer replaced by ``Raw`` reports no warnings. + +:meth:`~pcapkit.protocols.protocol.ProtocolBase._parse_next_layer` parses a next +layer on trial and keeps it as :class:`~pcapkit.protocols.misc.raw.Raw` when +fewer octets were captured than its header needs (#1170). The trial's warnings +were reported anyway, for a layer the result does not have. A PCAP-NG Simple +Packet Block cut to 10 octets by its interface's snaplen warned ``packet length +< 0: -2`` and ``-4`` from the Ethernet header it then dropped, on parse, on both +``from_data`` rebuilds and on ``make``. Nothing in it is PCAP-NG's: an Enhanced +Packet Block, an obsolete Packet Block and a PCAP record cut the same way warned +the same, in either byte order. + +A next layer captured shorter than its length hint is now parsed under +:class:`~pcapkit.utilities.warnings.hold_warnings`, and its warnings are dropped +with the layer. A layer that is kept reports them from the caller's line when its +trial ends, inside the same guard -- so under an ``error`` filter every chain is +the one parsed holding nothing -- and a cut header that stays in the chain still +warns (#1470). Any other layer is parsed holding nothing. Holding leaves the +warnings filters and every ``__warningregistry__`` alone, lets a +:exc:`BaseException` through untouched, and survives a block left out of order. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import contextvars +import inspect +import struct +import unittest +import unittest.mock +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class +from tests.utilities import _unrelated_warning +from tests.utilities._harness import capture + +if TYPE_CHECKING: + from typing import Any + +#: Snaplen of the capture's interface, and so the octets each packet keeps. +SNAPLEN = 10 +#: Original length of each packet, longer than the Ethernet header. +ORIGINAL = 57 +#: The packet data captured. +DATA = bytes(range(SNAPLEN)) + +#: Ethernet destination and source addresses. +MACS = bytes(6) + bytes.fromhex('020000000001') +#: An IPv4 header of total length 30, carrying TCP. +IPV4_TCP = bytes.fromhex('4500001e00000000400600007f0000017f000001') + +#: Messages for a 10-octet TCP header kept in the chain, measured unchanged against +#: the tree before the fix: the data offset is cut off, so its length reads as 0. +TCP_CUT_MESSAGES = ['packet length < 0: -2', 'packet length < 0: -3', 'packet length < 0: -4', + 'packet length < 0: -6', 'packet length < 0: -8', 'packet length < 0: -10'] + + +#: Whole frames cut at every length by the chain comparison: Ethernet, IPv4 with +#: an option and TCP with an MSS option; IPv6 and UDP; a VLAN tag, IPv4 and UDP; +#: and ARP. +FRAMES = ( + MACS + b'\x08\x00' + bytes.fromhex('46000034000000004006000' '07f0000017f000002' '01010100') + + bytes.fromhex('30390050000000010000000060020' 'fff000000' '020405b4') + b'abcd', + MACS + b'\x86\xdd' + bytes.fromhex('60000000000c1140') + bytes(15) + b'\x01' + bytes(15) + b'\x02' + + bytes.fromhex('00350035000c0000') + b'abcd', + MACS + b'\x81\x00' + bytes.fromhex('00010800') + bytes.fromhex('45000020000000004011000' '07f0000017f000002') + + bytes.fromhex('00350035000c0000') + b'abcd', + MACS + b'\x08\x06' + bytes.fromhex('0001080006040001') + bytes(6) + bytes.fromhex('7f000001') + + bytes(6) + bytes.fromhex('7f000002'), +) + + +class _HoldNothing: + """A stand-in for :class:`hold_warnings` that holds nothing, as before #1580.""" + + discard = False + + def __init__(self, hold: 'bool' = True) -> None: + del hold + + def __enter__(self) -> '_HoldNothing': + return self + + def __exit__(self, *exc_info: 'object') -> None: + return None + + +def _block(endian: 'str', type_: 'int', body: 'bytes') -> 'bytes': + """A PCAP-NG block of ``type_`` around ``body``.""" + length = len(body) + 12 + return struct.pack(f'{endian}II', type_, length) + body + struct.pack(f'{endian}I', length) + + +def _pcapng(endian: 'str', type_: 'int') -> 'tuple[bytes, bytes]': + """A one-packet capture over a snaplen of 10, and its packet block.""" + shb = _block(endian, 0x0A0D0D0A, struct.pack(f'{endian}IHHq', 0x1A2B3C4D, 1, 0, -1)) + idb = _block(endian, 1, struct.pack(f'{endian}HHI', 1, 0, SNAPLEN)) # Ethernet + data = DATA + bytes(-len(DATA) % 4) + body = { + 3: struct.pack(f'{endian}I', ORIGINAL) + data, + 6: struct.pack(f'{endian}IIIII', 0, 0, 0, SNAPLEN, ORIGINAL) + data, + 2: struct.pack(f'{endian}HHIIII', 0, 0, 0, 0, SNAPLEN, ORIGINAL) + data, + }[type_] + block = _block(endian, type_, body) + return shb + idb + block, block + + +class TestTrialParseWarnings(unittest.TestCase): + """Pin that a next layer replaced by ``Raw`` reports nothing, and a kept one does.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _schema(self, caught: 'list[warnings.WarningMessage]') -> 'list[str]': + """Messages of the :class:`SchemaWarning` among ``caught``.""" + from pcapkit.utilities.warnings import SchemaWarning + + return [str(item.message) for item in caught if issubclass(item.category, SchemaWarning)] + + def _extract(self, octets: 'bytes', name: 'str') -> 'tuple[Any, list[warnings.WarningMessage]]': + """Extract ``octets`` in memory, with the warnings it reports.""" + from pcapkit.foundation.extraction import Extractor + from tests.protocols.misc.test_pcapng_unit import NamedBuffer + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + extractor = Extractor(NamedBuffer(octets, name), nofile=True, store=True) + return extractor, caught + + def test_cut_packet_blocks_warn_nothing(self) -> None: + """SPB, EPB and Packet Block cut to 10 octets, in both byte orders, on every path.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.misc.raw import Raw + + for endian in '<>': + for type_ in (BlockType.Simple_Packet_Block, BlockType.Enhanced_Packet_Block, + BlockType.Packet_Block): + with self.subTest(endian=endian, type=type_): + octets, block = _pcapng(endian, int(type_)) + extractor, caught = self._extract(octets, 'truncated.pcapng') + self.assertEqual(self._schema(caught), []) + + frame, = extractor.frame + self.assertEqual(frame.info.captured_len, SNAPLEN) + self.assertEqual(frame.info.original_len, ORIGINAL) + self.assertIsInstance(frame.payload, Raw) + self.assertEqual(frame.payload.data, DATA) + + context = extractor.engine._ctx_list[0] # pylint: disable=protected-access + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + rebuilt = [PCAPNG.from_data(info, num=1, sct=1, ctx=context).data + for info in (frame.info, frame.info.to_dict())] + made = PCAPNG(num=1, sct=1, ctx=context, type=type_, + block={'packet_data': bytes(range(ORIGINAL)), 'timestamp': 0}) + self.assertEqual(self._schema(caught), []) + self.assertEqual(rebuilt, [block, block]) + self.assertEqual(made.data, block) + + def test_cut_pcap_record_warns_nothing(self) -> None: + """A PCAP record of 10 octets: the Ethernet header was not PCAP-NG's to drop.""" + from pcapkit.protocols.misc.raw import Raw + + header = struct.pack(' None: + """``Ethernet(bytes(10))`` is not on trial: it is built directly, kept, and warns.""" + from pcapkit.protocols.link.ethernet import Ethernet + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = Ethernet(bytes(10)) + self.assertIsInstance(parsed, Ethernet) + self.assertEqual(self._schema(caught), ['packet length < 0: -2', 'packet length < 0: -4']) + + def test_kept_cut_layers_still_warn_from_the_callers_line(self) -> None: + """Held layers that are kept, or that raise, report as they did, on both channels.""" + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.utilities.logging import logger + from pcapkit.utilities.warnings import SchemaWarning + + # TCP cut to 10 octets is held, inside IPv4 and Ethernet, which are not; + # IPv4 cut to 18 octets is held and raises, and the error path keeps it. + cases = ((MACS + b'\x08\x00' + IPV4_TCP + bytes(10), 'Ethernet:IPv4:TCP', TCP_CUT_MESSAGES), + (MACS + b'\x08\x00' + IPV4_TCP[:18], 'Ethernet:Internet_Protocol_version_4', + ['packet length < 0: -2'] * 2)) + for frame, chain, messages in cases: + with self.subTest(chain=chain): + with capture(logger) as recorder: + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + line = inspect.currentframe().f_lineno + 1 # type: ignore[union-attr] + parsed = Ethernet(frame, len(frame)) + self.assertEqual(str(parsed.protochain), chain) + self.assertEqual(self._schema(caught), messages) + self.assertEqual({(item.category, item.filename, item.lineno) for item in caught}, + {(SchemaWarning, __file__, line)}) + # the error path also logs the error itself, which is not a warning + self.assertEqual([(record.getMessage(), record.pathname, record.lineno) + for record in recorder.records + if record.getMessage().startswith('packet length')], + [(message, __file__, line) for message in messages]) + + def test_cut_layer_logs_nothing(self) -> None: + """The logger channel drops a discarded trial's warnings too.""" + from pcapkit.utilities.logging import logger + + octets, _ = _pcapng('<', 3) + with capture(logger) as recorder: + self._extract(octets, 'truncated.pcapng') + self.assertFalse([record for record in recorder.records + if record.getMessage().startswith('packet length')], recorder.messages) + + def test_holds_nest(self) -> None: + """Each block reports what it keeps when it is left; a discarded one drops its own.""" + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with hold_warnings(): + warn('outer, before', SchemaWarning) + with hold_warnings() as inner: + warn('inner, discarded', SchemaWarning) + inner.discard = True + trial = ProtocolBase._parse_next_layer(Ethernet, bytes(10), 10) # pylint: disable=protected-access + with hold_warnings(): + warn('inner, kept', SchemaWarning) + with hold_warnings(False): + warn('held by nothing', SchemaWarning) + self.assertEqual(self._schema(caught), ['inner, kept', 'held by nothing']) + warn('outer, after', SchemaWarning) + with hold_warnings() as outer: + with hold_warnings(): + warn('kept where the inner block is left', SchemaWarning) + warn('dropped with the outer block', SchemaWarning) + outer.discard = True + self.assertIsInstance(trial, Raw) + self.assertEqual(self._schema(caught), ['inner, kept', 'held by nothing', 'outer, before', + 'outer, after', 'kept where the inner block is left']) + self.assertEqual({item.filename for item in caught}, {__file__}) + + def test_layer_that_cannot_be_replaced_holds_nothing(self) -> None: + """Ethernet of 32 octets is past its hint, so even a discarding block cannot drop its warnings.""" + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.protocol import ProtocolBase, _short_of_hint + from pcapkit.utilities.warnings import hold_warnings + + class NeedsAnInstance: + """A hint that cannot be read from the class.""" + + def __length_hint__(self) -> 'int': + return len(self.__dict__) + + self.assertEqual([_short_of_hint(Ethernet, 13), _short_of_hint(Ethernet, 14), + _short_of_hint(Raw, 0), _short_of_hint(NeedsAnInstance, 10 ** 6)], # type: ignore[arg-type] + [True, False, False, True]) + + frame = MACS + b'\x08\x00' + IPV4_TCP[:18] + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with hold_warnings() as outer: + parsed = ProtocolBase._parse_next_layer(Ethernet, frame, len(frame)) # pylint: disable=protected-access + outer.discard = True + self.assertEqual(str(parsed.protochain), 'Ethernet:Internet_Protocol_version_4') + self.assertEqual(self._schema(caught), ['packet length < 0: -2'] * 2) + + def test_error_filter_chains_match_holding_nothing(self) -> None: + """Under ``error``, every cut of four frames parses to the chain it has holding nothing. + + A warning made an error is raised where the trial that kept it ends, + inside the guard it was raised in when nothing was held, so the same + layer, and only it, becomes ``Raw``. Holding until the outermost trial + ended instead turned the whole subtree into ``Raw``. + + """ + import pcapkit.protocols.protocol as protocol_module + from pcapkit.protocols.link.ethernet import Ethernet + + def chains() -> 'list[str]': + found = [] + with warnings.catch_warnings(): + warnings.simplefilter('error') + for frame in FRAMES: + for cut in range(1, len(frame) + 1): + try: + found.append(str(Ethernet(frame[:cut], cut).protochain)) + except Exception as exc: # pylint: disable=broad-except + found.append(f'raised {type(exc).__name__}') + return found + + held = chains() + with unittest.mock.patch.object(protocol_module, 'hold_warnings', _HoldNothing): + reference = chains() + self.assertEqual(len(held), sum(map(len, FRAMES))) + self.assertEqual(held, reference) + self.assertGreater(len(set(held)), 8, set(held)) + + frame = MACS + b'\x08\x00' + IPV4_TCP + bytes(10) + with warnings.catch_warnings(): + warnings.simplefilter('error') + self.assertEqual(str(Ethernet(frame, len(frame)).protochain), 'Ethernet:IPv4:TCP') + + def test_always_filter_only_drops_warnings(self) -> None: + """Under ``always``, every cut reports the warnings it reports holding nothing, less some.""" + import pcapkit.protocols.protocol as protocol_module + from pcapkit.protocols.link.ethernet import Ethernet + + def reports() -> 'list[list[tuple[Any, ...]]]': + found = [] + for frame in FRAMES: + for cut in range(1, len(frame) + 1): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + try: + Ethernet(frame[:cut], cut) + except Exception: # pylint: disable=broad-except + pass + found.append([(item.category, str(item.message), item.filename, item.lineno) + for item in caught]) + return found + + held = reports() + with unittest.mock.patch.object(protocol_module, 'hold_warnings', _HoldNothing): + reference = reports() + for kept, every in zip(held, reference): + remaining = iter(every) + self.assertTrue(all(item in remaining for item in kept), (kept, every)) + self.assertLess(sum(map(len, held)), sum(map(len, reference))) + + def test_base_exception_is_let_through(self) -> None: + """A :exc:`BaseException` leaves a block as itself, its held warnings dropped.""" + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + for error in (KeyboardInterrupt, SystemExit, GeneratorExit): + for action in ('error', 'always'): + with self.subTest(error=error.__name__, action=action): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter(action) + with self.assertRaises(error): + with hold_warnings(): + warn('held when the interrupt came', SchemaWarning) + raise error() + self.assertEqual(caught, []) + + def test_hold_left_by_an_error_reports(self) -> None: + """A block left by an exception reports what it held, discarded or not.""" + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with self.assertRaises(KeyError): + with hold_warnings() as held: + warn('held, then an error', SchemaWarning) + held.discard = True + raise KeyError('mid-parse') + self.assertEqual(self._schema(caught), ['held, then an error']) + + def test_held_frame_gone_is_reported_outside_pcapkit(self) -> None: + """A level past the stack, with no frame to find again, still blames this file.""" + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with hold_warnings(): + warn('past the outermost frame', SchemaWarning, stacklevel=10 ** 6) + self.assertEqual([(str(item.message), item.filename) for item in caught], + [('past the outermost frame', __file__)]) + + def test_hold_is_local_to_its_context(self) -> None: + """A warning reported in another context is not held by this one's block.""" + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with hold_warnings() as held: + contextvars.Context().run(warn, 'another context', SchemaWarning) + self.assertEqual(self._schema(caught), ['another context']) + held.discard = True + self.assertEqual(self._schema(caught), ['another context']) + + def test_hold_left_out_of_order(self) -> None: + """A generator's block left out of order neither ends another block nor outlives itself.""" + import pcapkit.utilities.warnings as warnings_module + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + def generator() -> 'Any': + with hold_warnings(): + warn('generator', SchemaWarning) + yield + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + # the generator's block is left while the caller's, opened after it, is open + suspended = generator() + next(suspended) + with hold_warnings() as caller: + next(suspended, None) + warn('caller, discarded', SchemaWarning) + caller.discard = True + warn('after both', SchemaWarning) + + # the caller's block is left while the generator's, opened inside it, is open + with hold_warnings(): + suspended = generator() + next(suspended) + next(suspended, None) + warn('after both again', SchemaWarning) + self.assertEqual(self._schema(caught), ['generator', 'after both', 'generator', 'after both again']) + self.assertIsNone(warnings_module._HELD.get()) # pylint: disable=protected-access + + def test_hold_left_in_a_copied_context(self) -> None: + """A block left in a copy of its context unsets itself there, and is stepped over here.""" + import pcapkit.utilities.warnings as warnings_module + from pcapkit.utilities.warnings import SchemaWarning, hold_warnings, warn + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with hold_warnings() as outer: + block = hold_warnings() + block.__enter__() # pylint: disable=unnecessary-dunder-call + contextvars.copy_context().run(block.__exit__, None, None, None) + warn('held by the outer block', SchemaWarning) + outer.discard = True + warn('after both', SchemaWarning) + self.assertEqual(self._schema(caught), ['after both']) + self.assertIsNone(warnings_module._HELD.get()) # pylint: disable=protected-access + + def test_filters_and_registries_are_untouched(self) -> None: + """Holding and dropping changes no filter, so a once-only warning stays once. + + Any change to the filters invalidates every ``__warningregistry__``, and + the unrelated module's second warning would then be reported again. + + """ + from pcapkit.foundation.extraction import Extractor + from pcapkit.protocols.link.ethernet import Ethernet + from tests.protocols.misc.test_pcapng_unit import NamedBuffer + + octets, _ = _pcapng('>', 3) + with warnings.catch_warnings(record=True) as caught: + warnings.resetwarnings() + warnings.simplefilter('default') + snapshot = warnings.filters[:] + + _unrelated_warning.emit() + Extractor(NamedBuffer(octets, 'truncated.pcapng'), nofile=True, store=True) # dropped + Ethernet(MACS + b'\x08\x00' + IPV4_TCP + bytes(10)) # held and kept + _unrelated_warning.emit() + + self.assertEqual(warnings.filters, snapshot) + unrelated = [item for item in caught if item.category is _unrelated_warning.UnrelatedWarning] + self.assertEqual(len(unrelated), 1) + self.assertEqual(self._schema(caught), TCP_CUT_MESSAGES) + + +if __name__ == '__main__': + unittest.main()