From 9970fa305d20261c05c14dabd9f5b9c1ad3d9bfd Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 26 Oct 2026 16:42:07 -0400 Subject: [PATCH] fix(protocols): drop the warnings of a trial parse that is kept as Raw (#2308) Problem: a held trial parse in _parse_next_layer that raised had its warnings replayed by hold_warnings.__exit__ whatever discard said, and beholder then kept the layer as Raw, so an Ethernet frame cut inside its IPv4 header warned "packet length < 0" for a layer it does not have. Fix: hold_warnings honours discard on an Exception exit too; beholder marks its call in a context variable, and _parse_next_layer discards a raising trial's warnings under it. Outside beholder the error reaches the caller and the warnings are still reported. The Raw's error names the cut. Test: tests/protocols/test_trial_parse_raises_2308_unit.py; the #2236 test pins the cap_len argument directly and asserts the Raw's error; the #1580 tests that pinned the leak are retargeted to kept layers. --- pcapkit/corekit/decorators.py | 19 ++- pcapkit/protocols/misc/pcap/frame.py | 14 +-- pcapkit/protocols/protocol.py | 20 ++- pcapkit/utilities/warnings.py | 19 ++- .../pcap/test_frame_read_cap_len_2236_unit.py | 50 ++++---- .../test_trial_parse_raises_2308_unit.py | 118 ++++++++++++++++++ .../test_trial_parse_warnings_1580_unit.py | 85 +++++++------ 7 files changed, 238 insertions(+), 87 deletions(-) create mode 100644 tests/protocols/test_trial_parse_raises_2308_unit.py diff --git a/pcapkit/corekit/decorators.py b/pcapkit/corekit/decorators.py index 730802bc4a..633e86aae6 100644 --- a/pcapkit/corekit/decorators.py +++ b/pcapkit/corekit/decorators.py @@ -24,6 +24,7 @@ decorators either. """ +import contextvars import functools import io import os @@ -53,6 +54,12 @@ #: :data:`pcapkit.utilities.logging.logger`. logger = get_logger(__name__) +#: contextvars.ContextVar[bool]: Whether a call :func:`beholder` wraps is under +#: way, so that an :exc:`Exception` raised inside it ends with the layer that +#: raised replaced; see :meth:`ProtocolBase._parse_next_layer +#: ` (:issue:`2308`). +_beholding = contextvars.ContextVar('_beholding', default=False) + def seekset(func: 'Callable[Concatenate[ProtocolBase, P], R_seekset]') -> 'Callable[P, R_seekset]': """Read file from start then set back to original. @@ -109,6 +116,12 @@ def beholder( :class:`~pcapkit.protocols.misc.null.NoPayload`, if the error is an end-of-file :exc:`~pcapkit.utilities.exceptions.StructError`. + When :meth:`ProtocolBase._parse_next_layer + ` held the + warnings of the layer that raised, they are dropped with it: they describe a + layer the result does not have, and the replacement's ``error`` reports the + failure (:issue:`2308`). + Note: The decorated function should have the following signature:: @@ -141,7 +154,11 @@ def behold(*args: 'P.args', **kwargs: 'P.kwargs') -> 'R_beholder': try: # call method - return func(*args, **kwargs) + token = _beholding.set(True) + try: + return func(*args, **kwargs) + finally: + _beholding.reset(token) except Exception as exc: if isinstance(exc, StructError) and exc.eof: # pylint: disable=no-member from pcapkit.protocols.misc.null import NoPayload as protocol # isort: skip # pylint: disable=import-outside-toplevel diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index 07a11a0bca..9bc7dd6c33 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -386,14 +386,12 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, # snapshot length cut short the latter exceeds what the file holds, and # handing it over is the declared-length-exceeds-available-octets fault. # - # The schemas count their fields down from this length, so it is what - # lets a truncation be noticed: a frame cut inside its IPv4 header - # leaves IPv4 fewer octets than its header, and the count running below - # zero issues a ``SchemaWarning`` (``packet length < 0``). Handed the - # on-wire length, IPv4 would be told of octets the file does not hold - # and the truncation would pass silently. - # ``test_truncated_ipv4_header_warns_whatever_the_on_wire_length`` pins it - # (#2236). + # A frame cut inside its IPv4 header keeps IPv4 as ``Raw``, whose + # ``error`` names the cut, for either length: the trial parse of IPv4 + # raises, and its ``packet length < 0`` warnings go with the layer + # (#2308). As the dissection does not tell the two apart, + # ``test_truncated_ipv4_header_is_raw_with_its_error`` pins the argument + # where it is passed (#2236). return self._decode_next_layer(frame, self._ghdr.network, frame.cap_len) def make(self, diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index c144fbdc22..04eb27bf2d 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -44,7 +44,7 @@ import aenum from pcapkit.corekit.context import ContextRegistry -from pcapkit.corekit.decorators import beholder, seekset +from pcapkit.corekit.decorators import _beholding, beholder, seekset from pcapkit.corekit.fields.misc import ConditionalField, SchemaField, SwitchField from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.protochain import ProtoChain @@ -2352,8 +2352,13 @@ def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int' 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. + still warns. The held warnings of a parse that raises are dropped + too when :func:`~pcapkit.corekit.decorators.beholder` replaces the + layer, as it does around :meth:`_import_next_layer`, since the + replacement's ``error`` reports the failure (:issue:`2308`); outside + it, they are reported as the error leaves. A layer captured at least + as long as its hint cannot be replaced, and is parsed holding + nothing, even inside one that holds. A layer past :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` in its frame is not parsed: it and the rest of the frame are kept as @@ -2387,7 +2392,14 @@ def _parse_next_layer(klass: 'Type[ProtocolBase]', file_: 'bytes', length: 'int' token = _frame_depth.set(depth) try: with hold_warnings(_short_of_hint(klass, captured)) as held: - next_ = klass(file_, length, **kwargs) # type: ignore[abstract] + try: + next_ = klass(file_, length, **kwargs) # type: ignore[abstract] + except Exception: + # NOTE: Under ``beholder`` the layer that raised is replaced, + # so its warnings go with it; outside it, the error reaches + # the caller and they are reported as it leaves (#2308). + held.discard = _beholding.get() + raise length_hint = getattr(next_, '__length_hint__', None) hint = None if length_hint is None else length_hint() diff --git a/pcapkit/utilities/warnings.py b/pcapkit/utilities/warnings.py index 0727deb090..cce34c1cc8 100644 --- a/pcapkit/utilities/warnings.py +++ b/pcapkit/utilities/warnings.py @@ -237,11 +237,11 @@ class hold_warnings: # pylint: disable=invalid-name 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. + channels, unless :attr:`discard` was set, in which case they are dropped, + whether the block ends or is left by an :exc:`Exception`. A block left by any + other :exc:`BaseException` -- :exc:`KeyboardInterrupt`, :exc:`SystemExit`, + :exc:`GeneratorExit` -- drops them whatever :attr:`discard` says, 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 @@ -255,8 +255,9 @@ class hold_warnings: # pylint: disable=invalid-name :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`). + header was not captured, or that :func:`~pcapkit.corekit.decorators.beholder` + replaces when it raises: the warnings describe a layer the result does not + have (:issue:`1580`, :issue:`2308`). Note: Nothing here touches ``warnings.filters`` or a module's @@ -310,9 +311,7 @@ def __exit__(self, exc_type: 'Optional[Type[BaseException]]', _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): + if self.discard or (exc_type is not None and not issubclass(exc_type, Exception)): return for message, category, target in held: _replay(message, category, target) diff --git a/tests/protocols/misc/pcap/test_frame_read_cap_len_2236_unit.py b/tests/protocols/misc/pcap/test_frame_read_cap_len_2236_unit.py index f48b36ea18..2f9459ad63 100644 --- a/tests/protocols/misc/pcap/test_frame_read_cap_len_2236_unit.py +++ b/tests/protocols/misc/pcap/test_frame_read_cap_len_2236_unit.py @@ -3,11 +3,12 @@ This pins the length argument :meth:`Frame.read ` passes to ``_decode_next_layer`` -(#2236). The dissector is handed ``frame.cap_len``, the octets present, so a -frame the snapshot length cut inside its IPv4 header reports the truncation as a -:class:`~pcapkit.utilities.warnings.SchemaWarning` ``packet length < 0``. Handed -the on-wire ``frame.len`` instead, the schema believes the octets it lacks are -there and the truncation passes silently. +(#2236): ``frame.cap_len``, the octets present, never the on-wire ``frame.len``. +A frame the snapshot length cut inside its IPv4 header keeps IPv4 as +:class:`~pcapkit.protocols.misc.raw.Raw`, whose ``error`` names the cut, and the +trial parse's ``packet length < 0`` warnings go with the layer (#2308). The +dissection reads the same for either length, so the argument is read where it +is passed. Every case builds its own octets and reads no capture. Classes are imported inside each test, after :func:`~tests._support.reimport_once_per_class`. @@ -17,6 +18,7 @@ import io import struct import unittest +import unittest.mock import warnings from tests._support import reimport_once_per_class @@ -42,29 +44,31 @@ class TestFrameReadCapLen(unittest.TestCase): def setUp(self) -> None: reimport_once_per_class(self) - def schema_warnings(self, orig_len: 'int') -> 'list[str]': - """The :class:`SchemaWarning` messages from parsing the record for ``orig_len``.""" + def test_truncated_ipv4_header_is_raw_with_its_error(self) -> None: + """The cut IPv4 header is kept as ``Raw`` naming the cut, with no warning. + + IPv4 is handed the 10 octets Ethernet leaves of the captured 24, short + of its 20-octet header, so the trial parse raises and is replaced. Its + warnings, ``packet length < 0`` down to ``-10``, describe the IPv4 layer + the result does not have, and are dropped with it. + """ from pcapkit.protocols.misc.pcap.frame import Frame from pcapkit.protocols.misc.pcap.header import Header + from pcapkit.protocols.misc.raw import Raw from pcapkit.utilities.warnings import SchemaWarning - with warnings.catch_warnings(record=True) as caught: + decode = unittest.mock.patch.object(Frame, '_decode_next_layer', autospec=True, + side_effect=Frame._decode_next_layer) # pylint: disable=protected-access + with warnings.catch_warnings(record=True) as caught, decode as spy: warnings.simplefilter('always') - frame = Frame(io.BytesIO(_record(orig_len)), num=1, header=Header(GLOBAL).info) - self.assertEqual(frame.info.cap_len, len(PACKET)) - self.assertEqual(frame.info.len, orig_len) - return [str(item.message) for item in caught if issubclass(item.category, SchemaWarning)] - - def test_truncated_ipv4_header_warns_whatever_the_on_wire_length(self) -> None: - """The cut IPv4 header warns, the same as for an uncut frame of the same octets. - - IPv4 is handed the 10 octets Ethernet leaves of the captured 24, so its - header runs the remaining length to ``-10``. Were the on-wire 154 handed - over instead, IPv4 would be told of 140 octets and warn of nothing. - """ - truncated = self.schema_warnings(ORIG_LEN) - self.assertIn('packet length < 0: -10', truncated) - self.assertEqual(truncated, self.schema_warnings(len(PACKET))) + frame = Frame(io.BytesIO(_record(ORIG_LEN)), num=1, header=Header(GLOBAL).info) + self.assertEqual((frame.info.cap_len, frame.info.len), (len(PACKET), ORIG_LEN)) + self.assertEqual(spy.call_args.args[3], len(PACKET)) + + ipv4 = frame.payload.payload + self.assertIsInstance(ipv4, Raw) + self.assertEqual(ipv4.info.error, 'IPv4: header length 20 runs past the end of the data (10 octets)') + self.assertEqual([str(item.message) for item in caught if issubclass(item.category, SchemaWarning)], []) if __name__ == '__main__': diff --git a/tests/protocols/test_trial_parse_raises_2308_unit.py b/tests/protocols/test_trial_parse_raises_2308_unit.py new file mode 100644 index 0000000000..e397336260 --- /dev/null +++ b/tests/protocols/test_trial_parse_raises_2308_unit.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2308: a trial parse that raises and is kept as ``Raw`` warns nothing. + +:meth:`~pcapkit.protocols.protocol.ProtocolBase._parse_next_layer` holds the +warnings of a next layer captured shorter than its length hint (#1580). When +that parse raises, :func:`~pcapkit.corekit.decorators.beholder` replaces the +layer, and without the fix the held warnings were reported anyway: an Ethernet +frame cut inside its IPv4 header warned ``packet length < 0`` three times for an +IPv4 layer the result keeps as ``Raw``. They are dropped with the layer, whose +``error`` names the cut. A held layer that is kept still warns, and so does one +that raises where no :func:`~pcapkit.corekit.decorators.beholder` replaces it. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import re +import unittest +import warnings + +from tests._support import reimport_once_per_class +from tests.utilities._harness import capture + +#: Test tier (#1958): exercises the warning machinery, which free-threaded +#: builds make context-aware. +__tier__ = 'ALL' + +#: The first 10 of the 20 octets of an IPv4 header. +IPV4_CUT = bytes.fromhex('4500008c000100004011') +#: An Ethernet frame carrying :data:`IPV4_CUT`. +FRAME = bytes(6) + bytes(range(1, 7)) + b'\x08\x00' + IPV4_CUT +#: The ``error`` of the ``Raw`` that :data:`FRAME`'s IPv4 is kept as. +ERROR = 'IPv4: header length 20 runs past the end of the data (10 octets)' + + +def _ipv6(next_header: 'int') -> 'bytes': + """An Ethernet frame carrying IPv6 whose ``next_header`` is cut to 1 octet.""" + return (bytes(6) + bytes(range(1, 7)) + b'\x86\xdd' + + bytes.fromhex('600000000001') + bytes([next_header, 64]) + + bytes(15) + b'\x01' + bytes(15) + b'\x02' + b'\x00') + + +#: The warnings of the kept, cut ``IPv6_Ext`` that each :func:`_ipv6` frame ends in. +IPV6_EXT_CUT = ['packet length < 0: -1', + 'IPv6: extension header declares a length of 8 octet(s) with 1 octet(s) ' + 'left in the chain; stopping the walk instead of skipping past it'] + + +class TestTrialParseRaises(unittest.TestCase): + """Pin that a raising trial parse replaced by ``beholder`` reports nothing.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _parse(self, frame: 'bytes', action: 'str' = 'always') -> 'tuple[object, list[str]]': + """Parse ``frame`` as Ethernet, with the :class:`SchemaWarning` messages it reports.""" + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.utilities.warnings import SchemaWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter(action) + parsed = Ethernet(frame, len(frame)) + return parsed, [str(item.message) for item in caught if issubclass(item.category, SchemaWarning)] + + def test_cut_ipv4_header_is_raw_and_warns_nothing(self) -> None: + """IPv4 cut to 10 octets raises, is kept as ``Raw`` naming the cut, and reports nothing.""" + from pcapkit.protocols.misc.raw import Raw + from pcapkit.utilities.logging import logger + + with capture(logger) as recorder: + parsed, messages = self._parse(FRAME) + ipv4 = parsed.payload # type: ignore[attr-defined] + self.assertIsInstance(ipv4, Raw) + self.assertEqual(ipv4.info.error, ERROR) + self.assertEqual(messages, []) + self.assertEqual([record.getMessage() for record in recorder.records + if record.getMessage().startswith('packet length')], []) + + def test_error_filter_keeps_the_cut_as_the_error(self) -> None: + """Under ``error``, the ``Raw`` names the cut header, not a warning it dropped.""" + from pcapkit.protocols.misc.raw import Raw + + parsed, messages = self._parse(FRAME, 'error') + self.assertIsInstance(parsed.payload, Raw) # type: ignore[attr-defined] + self.assertEqual(parsed.payload.info.error, ERROR) # type: ignore[attr-defined] + self.assertEqual(messages, []) + + def test_replaced_by_ipv6_ext_reports_only_the_replacement(self) -> None: + """A cut Hop-by-Hop header raises and is parsed as ``IPv6_Ext``, which alone warns.""" + parsed, messages = self._parse(_ipv6(0)) + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv6:IPv6-Ext') # type: ignore[attr-defined] + self.assertEqual(messages, IPV6_EXT_CUT) + + def test_kept_trial_still_warns(self) -> None: + """A cut Shim6 header is held, kept, and reports its warnings as before.""" + parsed, messages = self._parse(_ipv6(140)) + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv6:IPv6-Ext') # type: ignore[attr-defined] + self.assertEqual(messages, IPV6_EXT_CUT) + + def test_raise_outside_beholder_reports(self) -> None: + """With no ``beholder`` to replace it, the layer's error and its warnings both reach the caller.""" + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.protocols.protocol import ProtocolBase + from pcapkit.utilities.exceptions import ProtocolError + from pcapkit.utilities.warnings import SchemaWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with self.assertRaisesRegex(ProtocolError, re.escape(ERROR)): + ProtocolBase._parse_next_layer(IPv4, IPV4_CUT, len(IPV4_CUT)) # pylint: disable=protected-access + self.assertEqual([str(item.message) for item in caught if issubclass(item.category, SchemaWarning)], + ['packet length < 0: -2', 'packet length < 0: -6', 'packet length < 0: -10', + 'packet length < 0: -10']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_trial_parse_warnings_1580_unit.py b/tests/protocols/test_trial_parse_warnings_1580_unit.py index 3a60693ef8..5faf111e3a 100644 --- a/tests/protocols/test_trial_parse_warnings_1580_unit.py +++ b/tests/protocols/test_trial_parse_warnings_1580_unit.py @@ -56,10 +56,14 @@ #: 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 the same with -#: and without 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'] +#: An Ethernet frame carrying IPv6 whose Shim6 header (140) is cut to 1 octet. +#: The header's length field is cut off, so its length reads as 0 and it is kept. +SHIM6_CUT = (MACS + b'\x86\xdd' + bytes.fromhex('600000000001' '8c40') + bytes(15) + b'\x01' + + bytes(15) + b'\x02' + b'\x00') +#: Messages for :data:`SHIM6_CUT`, measured the same with and without the fix. +SHIM6_CUT_MESSAGES = ['packet length < 0: -1', + 'IPv6: extension header declares a length of 8 octet(s) with 1 octet(s) ' + 'left in the chain; stopping the walk instead of skipping past it'] #: Whole frames cut at every length by the chain comparison: Ethernet, IPv4 with @@ -189,32 +193,24 @@ def test_ethernet_asked_for_still_warns(self) -> None: 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.""" + """A held layer that is kept reports as it 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]) + # Shim6 cut to 1 octet is held, inside IPv6 and Ethernet, which are not. + 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(SHIM6_CUT, len(SHIM6_CUT)) + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv6:IPv6-Ext') + self.assertEqual(self._schema(caught), SHIM6_CUT_MESSAGES) + self.assertEqual({(item.category, item.filename, item.lineno) for item in caught}, + {(SchemaWarning, __file__, line)}) + self.assertEqual([(record.getMessage(), record.pathname, record.lineno) + for record in recorder.records if record.getMessage() in SHIM6_CUT_MESSAGES], + [(message, __file__, line) for message in SHIM6_CUT_MESSAGES]) def test_cut_layer_logs_nothing(self) -> None: """The logger channel drops a discarded trial's warnings too.""" @@ -258,7 +254,11 @@ def test_holds_nest(self) -> None: 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.""" + """TCP of 24 octets is past its hint, so even a discarding block cannot drop its warnings. + + It holds nothing, so they are reported as they are raised, and stay + reported when the parse then raises and is kept as ``Raw`` (#2308). + """ from pcapkit.protocols.link.ethernet import Ethernet from pcapkit.protocols.misc.raw import Raw from pcapkit.protocols.protocol import ProtocolBase, _short_of_hint @@ -274,14 +274,15 @@ def __length_hint__(self) -> 'int': _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] + frame = FRAMES[0][:62] # TCP cut to 24 octets, past its hint, raises 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) + self.assertEqual(str(parsed.protochain), 'Ethernet:IPv4:TCP') + self.assertIsInstance(parsed.payload.payload, Raw) + self.assertEqual(self._schema(caught), ['packet length < 0: -95']) def test_error_filter_chains_match_holding_nothing(self) -> None: """Under ``error``, every cut of :data:`FRAMES` parses to the chain it has holding nothing. @@ -362,17 +363,19 @@ def test_base_exception_is_let_through(self) -> None: 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.""" + """A block left by an exception reports what it held unless discarded (#2308).""" 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']) + for discard, reported in ((False, ['held, then an error']), (True, [])): + with self.subTest(discard=discard): + 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 = discard + raise KeyError('mid-parse') + self.assertEqual(self._schema(caught), reported) 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.""" @@ -481,13 +484,13 @@ def test_filters_and_registries_are_untouched(self) -> None: _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 + Ethernet(SHIM6_CUT) # 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) + self.assertEqual(self._schema(caught), SHIM6_CUT_MESSAGES) def test_hold_warnings_is_exported(self) -> None: """``import *`` brings ``hold_warnings`` in, as it does ``warnings.catch_warnings`` (#1602)."""