From e73ba4be177be492fb55b97f838a4bb2c4e8153a Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 27 Oct 2026 14:14:28 -0400 Subject: [PATCH] fix(toolkit): lead every adapter message with its engine's own __engine_name__ (#2304) The pcapkit.toolkit adapters put the engine name mid-sentence instead of leading with it, unlike the : : ... {value!r} form #2232/#2257 established, and hand-typed the name rather than reading it off the engine class's own __engine_name__. Each adapter module gains _ENGINE = .__engine_name__ (plus the import), so a rename in pcapkit/foundation/engines/ cannot leave the adapter spelling the old name. Four modules, five sub-bugs: - pcap_ct.py/pypcap.py: "IPv4 reassembly is not supported by the pcap-ct engine: ..." -> "PCAP_CT: IPv4 reassembly is not supported: ..." (and the PyPCAP sibling). - pypcapfile.py: "invalid PyPCAPFile IPv4 address: 5" -> "PyPCAPFile: invalid IPv4 address: 5" (same for MAC address); the IPv6 UnsupportedCall and the "Frame N: TCP over IPv6 ..." warning get the same treatment. - pyshark.py: boolean-field, frame.time_epoch and no-frame-layer messages all reordered the same way. - scapy.py: the import-time warning reordered; the quoted package name is lowercased to 'scapy' to match the actual failing import and the sibling dependency warnings elsewhere in the codebase, which quote the installable name. The engine classes' toolkit imports are all function-local, so the new toolkit -> foundation.engines import introduces no cycle. Tests: new tests/toolkit/test_engine_message_prefix_2304_unit.py (9 tests) pins each message whole and asserts each module's _ENGINE is the engine class's own attribute. Existing pins in the four engines' test files updated for the new text. --- pcapkit/toolkit/pcap_ct.py | 16 +- pcapkit/toolkit/pypcap.py | 16 +- pcapkit/toolkit/pypcapfile.py | 24 +- pcapkit/toolkit/pyshark.py | 14 +- pcapkit/toolkit/scapy.py | 10 +- .../test_engine_message_prefix_2304_unit.py | 223 ++++++++++++++++++ tests/toolkit/test_pcap_ct_unit.py | 4 + tests/toolkit/test_pypcap_unit.py | 4 + tests/toolkit/test_pypcapfile_unit.py | 16 +- tests/toolkit/test_pyshark_unit.py | 7 +- 10 files changed, 305 insertions(+), 29 deletions(-) create mode 100644 tests/toolkit/test_engine_message_prefix_2304_unit.py diff --git a/pcapkit/toolkit/pcap_ct.py b/pcapkit/toolkit/pcap_ct.py index ac6e4276ae..1656c143de 100644 --- a/pcapkit/toolkit/pcap_ct.py +++ b/pcapkit/toolkit/pcap_ct.py @@ -32,6 +32,7 @@ """ from typing import TYPE_CHECKING +from pcapkit.foundation.engines.pcap_ct import PCAP_CT from pcapkit.utilities.exceptions import UnsupportedCall if TYPE_CHECKING: @@ -48,6 +49,13 @@ 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', ] +#: Message prefix shared by every adapter below, read off +#: :attr:`PCAP_CT.__engine_name__ +#: ` rather than +#: spelled out, so the engine cannot be renamed there and still be named the old +#: way here (:issue:`2304`). +_ENGINE = PCAP_CT.__engine_name__ + #: Explanatory suffix shared by every unsupported adapter below. _NO_DISSECTION = ("'pcap-ct' is a libpcap binding and performs no protocol " "dissection, so there is no protocol layer to read") @@ -109,7 +117,7 @@ def ipv4_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv4Add UnsupportedCall: Always, as `pcap-ct`_ provides no IPv4 layer to read. """ - raise UnsupportedCall(f'IPv4 reassembly is not supported by the pcap-ct engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: IPv4 reassembly is not supported: {_NO_DISSECTION}') def ipv6_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None': @@ -123,7 +131,7 @@ def ipv6_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv6Add UnsupportedCall: Always, as `pcap-ct`_ provides no IPv6 layer to read. """ - raise UnsupportedCall(f'IPv6 reassembly is not supported by the pcap-ct engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: IPv6 reassembly is not supported: {_NO_DISSECTION}') def tcp_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'TCP_Packet | None': @@ -137,7 +145,7 @@ def tcp_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'TCP_Packet | None' UnsupportedCall: Always, as `pcap-ct`_ provides no TCP layer to read. """ - raise UnsupportedCall(f'TCP reassembly is not supported by the pcap-ct engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: TCP reassembly is not supported: {_NO_DISSECTION}') def tcp_traceflow(packet: 'bytes', timestamp: 'float', *, data_link: 'Enum_LinkType', @@ -154,4 +162,4 @@ def tcp_traceflow(packet: 'bytes', timestamp: 'float', *, data_link: 'Enum_LinkT UnsupportedCall: Always, as `pcap-ct`_ provides no TCP layer to read. """ - raise UnsupportedCall(f'TCP flow tracing is not supported by the pcap-ct engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: TCP flow tracing is not supported: {_NO_DISSECTION}') diff --git a/pcapkit/toolkit/pypcap.py b/pcapkit/toolkit/pypcap.py index a791d84b58..a3c229a607 100644 --- a/pcapkit/toolkit/pypcap.py +++ b/pcapkit/toolkit/pypcap.py @@ -31,6 +31,7 @@ """ from typing import TYPE_CHECKING +from pcapkit.foundation.engines.pypcap import PyPCAP from pcapkit.utilities.exceptions import UnsupportedCall if TYPE_CHECKING: @@ -47,6 +48,13 @@ 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', ] +#: Message prefix shared by every adapter below, read off +#: :attr:`PyPCAP.__engine_name__ +#: ` rather than +#: spelled out, so the engine cannot be renamed there and still be named the old +#: way here (:issue:`2304`). +_ENGINE = PyPCAP.__engine_name__ + #: Explanatory suffix shared by every unsupported adapter below. _NO_DISSECTION = ("'pypcap' is a libpcap binding and performs no protocol " "dissection, so there is no protocol layer to read") @@ -108,7 +116,7 @@ def ipv4_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv4Add UnsupportedCall: Always, as `PyPCAP`_ provides no IPv4 layer to read. """ - raise UnsupportedCall(f'IPv4 reassembly is not supported by the PyPCAP engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: IPv4 reassembly is not supported: {_NO_DISSECTION}') def ipv6_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None': @@ -122,7 +130,7 @@ def ipv6_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'IP_Packet[IPv6Add UnsupportedCall: Always, as `PyPCAP`_ provides no IPv6 layer to read. """ - raise UnsupportedCall(f'IPv6 reassembly is not supported by the PyPCAP engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: IPv6 reassembly is not supported: {_NO_DISSECTION}') def tcp_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'TCP_Packet | None': @@ -136,7 +144,7 @@ def tcp_reassembly(packet: 'bytes', *, count: 'int' = -1) -> 'TCP_Packet | None' UnsupportedCall: Always, as `PyPCAP`_ provides no TCP layer to read. """ - raise UnsupportedCall(f'TCP reassembly is not supported by the PyPCAP engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: TCP reassembly is not supported: {_NO_DISSECTION}') def tcp_traceflow(packet: 'bytes', timestamp: 'float', *, data_link: 'Enum_LinkType', @@ -153,4 +161,4 @@ def tcp_traceflow(packet: 'bytes', timestamp: 'float', *, data_link: 'Enum_LinkT UnsupportedCall: Always, as `PyPCAP`_ provides no TCP layer to read. """ - raise UnsupportedCall(f'TCP flow tracing is not supported by the PyPCAP engine: {_NO_DISSECTION}') + raise UnsupportedCall(f'{_ENGINE}: TCP flow tracing is not supported: {_NO_DISSECTION}') diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index e4ca7e15e0..551f4be9a8 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -48,6 +48,7 @@ from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.foundation.engines.pypcapfile import PyPCAPFile from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet @@ -77,6 +78,13 @@ 'ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', ] +#: Message prefix shared by every message below, read off +#: :attr:`PyPCAPFile.__engine_name__ +#: ` rather +#: than spelled out, so the engine cannot be renamed there and still be named +#: the old way here (:issue:`2304`). +_ENGINE = PyPCAPFile.__engine_name__ + #: IPv4 **DF** (don't fragment) bit, within the three-bit flags field. IPV4_FLAG_DF = 0b010 @@ -163,7 +171,7 @@ def _parse_ipv4_address(value: 'Any') -> 'IPv4Address': """ if not isinstance(value, bytes): - raise ProtocolError(f'invalid PyPCAPFile IPv4 address: {value!r}') + raise ProtocolError(f'{_ENGINE}: invalid IPv4 address: {value!r}') try: if len(value) != 4: # Dotted-decimal ASCII text, the actual PyPCAPFile 0.12.0 form -- @@ -172,7 +180,7 @@ def _parse_ipv4_address(value: 'Any') -> 'IPv4Address': return ipaddress.IPv4Address(value.decode('ascii')) return ipaddress.IPv4Address(value) except (ValueError, UnicodeDecodeError) as error: - raise ProtocolError(f'invalid PyPCAPFile IPv4 address: {value!r}') from error + raise ProtocolError(f'{_ENGINE}: invalid IPv4 address: {value!r}') from error def _parse_mac_address(value: 'Any') -> 'str': @@ -201,14 +209,14 @@ def _parse_mac_address(value: 'Any') -> 'str': """ if not isinstance(value, (bytes, bytearray)): - raise ProtocolError(f'invalid PyPCAPFile MAC address: {value!r}') + raise ProtocolError(f'{_ENGINE}: invalid MAC address: {value!r}') raw = bytes(value) if len(raw) == 6: return raw.hex(':') try: return raw.decode('ascii') except UnicodeDecodeError as error: - raise ProtocolError(f'invalid PyPCAPFile MAC address: {value!r}') from error + raise ProtocolError(f'{_ENGINE}: invalid MAC address: {value!r}') from error def _maybe_unhex(value: 'bytes') -> 'bytes': @@ -410,9 +418,9 @@ def _decline_ipv6_tcp(packet: 'Packet', count: 'int') -> 'None': _ipv6_tcp_warned[key] = anchor while config.IPV6_TCP_WARNED_LIMIT != -1 and len(_ipv6_tcp_warned) > config.IPV6_TCP_WARNED_LIMIT: _ipv6_tcp_warned.popitem(last=False) - warn(f"Frame {count}: TCP over IPv6 is left out of TCP reassembly and flow tracing, as " - "'pypcapfile' has no IPv6 decoder; later such frames of this capture are left out " - 'without another warning', AttributeWarning, stacklevel=stacklevel()) + warn(f'{_ENGINE}: frame {count}: TCP over IPv6 is left out of TCP reassembly and flow ' + "tracing, as 'pypcapfile' has no IPv6 decoder; later such frames of this capture are " + 'left out without another warning', AttributeWarning, stacklevel=stacklevel()) def _transport(ipv4: 'IP') -> 'Optional[bytes]': @@ -747,7 +755,7 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad UnsupportedCall: Always, as `PyPCAPFile`_ has no IPv6 decoder. """ - raise UnsupportedCall('IPv6 reassembly is not supported by the PyPCAPFile engine: ' + raise UnsupportedCall(f'{_ENGINE}: IPv6 reassembly is not supported: ' "'pypcapfile' has no IPv6 decoder, so no IPv6 fragment header " 'can be read') diff --git a/pcapkit/toolkit/pyshark.py b/pcapkit/toolkit/pyshark.py index 09e7519f9e..2b4b6e1d1a 100644 --- a/pcapkit/toolkit/pyshark.py +++ b/pcapkit/toolkit/pyshark.py @@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, cast from pcapkit.const.reg.linktype import LinkType as Enum_LinkType +from pcapkit.foundation.engines.pyshark import PyShark from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.utilities.exceptions import MissingKeyError, ProtocolError @@ -29,6 +30,13 @@ __all__ = ['packet2dict', 'tcp_traceflow', 'ENCAP_TYPE_TO_LINKTYPE', 'FILTER_NAME_TO_LINKTYPE'] +#: Message prefix shared by every message below, read off +#: :attr:`PyShark.__engine_name__ +#: ` rather than +#: spelled out, so the engine cannot be renamed there and still be named the old +#: way here (:issue:`2304`). +_ENGINE = PyShark.__engine_name__ + #: ``frame.encap_type`` -- Wireshark's internal ``WTAP_ENCAP_*`` number, which |pyshark|_ #: exposes as ``packet.frame_info.encap_type`` -- to the matching #: :class:`~pcapkit.const.reg.linktype.LinkType` member. This is :func:`tcp_traceflow`'s primary @@ -343,7 +351,7 @@ def _parse_flag(value: 'Any', name: 'str') -> 'bool': flag = _BOOLEAN_FIELD.get(str(value)) if flag is not None: return flag - raise ProtocolError(f'invalid PyShark boolean field {name}: {value!r}') + raise ProtocolError(f'{_ENGINE}: invalid boolean field {name}: {value!r}') def _parse_epoch(value: 'Any') -> 'float': @@ -362,7 +370,7 @@ def _parse_epoch(value: 'Any') -> 'float': try: return float(decimal.Decimal(str(value))) except decimal.InvalidOperation: - raise ProtocolError(f'invalid PyShark frame.time_epoch: {value!r}') from None + raise ProtocolError(f'{_ENGINE}: invalid frame.time_epoch: {value!r}') from None def _absolute(tcp: 'Any', name: 'str') -> 'int': @@ -600,7 +608,7 @@ def tcp_traceflow(packet: 'Packet', *, count: 'int' = -1) -> 'TF_TCP_Packet | No # first layer that is not the frame layer. frame = _frame_layer(packet) if frame is None: - raise ProtocolError(f'PyShark packet {packet.number} has no frame layer') + raise ProtocolError(f'{_ENGINE}: packet {packet.number} has no frame layer') encap_type = getattr(frame, 'encap_type', None) if encap_type is None: name = [layer for layer in packet.layers if layer is not frame][0].layer_name diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 4367ff6c58..f4457d2c6c 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -39,6 +39,7 @@ from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.const.reg.transtype import TransType as Enum_TransType +from pcapkit.foundation.engines.scapy import Scapy from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet from pcapkit.foundation.traceflow.data.data import FrameRecord @@ -55,11 +56,18 @@ from pcapkit.utilities.exceptions import MissingKeyError, ModuleNotFound, stacklevel from pcapkit.utilities.warnings import ScapyWarning, warn +#: Message prefix shared by every message below, read off +#: :attr:`Scapy.__engine_name__ +#: ` rather than spelled +#: out, so the engine cannot be renamed there and still be named the old way +#: here (:issue:`2304`). +_ENGINE = Scapy.__engine_name__ + try: import scapy except ModuleNotFoundError: scapy = None - warn("dependency package 'Scapy' not found", + warn(f"{_ENGINE}: dependency package 'scapy' not found", ScapyWarning, stacklevel=stacklevel()) if TYPE_CHECKING: diff --git a/tests/toolkit/test_engine_message_prefix_2304_unit.py b/tests/toolkit/test_engine_message_prefix_2304_unit.py new file mode 100644 index 0000000000..a4a38419f3 --- /dev/null +++ b/tests/toolkit/test_engine_message_prefix_2304_unit.py @@ -0,0 +1,223 @@ +# -*- coding: utf-8 -*- +"""Every toolkit adapter message leads with its engine's name. C.f. #2304. + +The adapters under :mod:`pcapkit.toolkit` put the engine name inside the +sentence -- ``IPv4 reassembly is not supported by the pcap-ct engine: ...``, +``invalid PyShark boolean field tcp.flags_fin: ...`` -- rather than leading with +it as the ``: : ... {value!r}`` form asks for +(#2232, #2257), and spelled it their own way rather than the way the engine's +own ``__engine_name__`` does: ``pcap-ct`` where the engine is ``PCAP_CT``. + +Each message is pinned whole below, and each prefix is checked against the +engine class's ``__engine_name__`` rather than against a literal, so a prefix +typed by hand here or there regresses even when the two happen to agree today. + +Nothing here imports ``pcap``, ``pcapfile``, ``pyshark`` or ``scapy``: every +message comes off a pure adapter function, or -- for the one warning Scapy +emits on import -- off an import with the package mocked away. So the whole +module runs with no third-party engine installed, which is also why it pins the +messages here rather than only in the per-engine modules, three of which skip +without their backend. + +""" +from __future__ import annotations + +import builtins +import importlib +import importlib.util +import types +import unittest +import warnings +from unittest import mock + +from tests._support import purge_modules, reimport_once_per_class +from tests.toolkit.test_pypcapfile_unit import ipv6_frame, make_packet +from tests.toolkit.test_pyshark_unit import FakePySharkPacket, LayerField + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: The suffix every ``pcap``-binding refusal carries, by engine module name. +NO_DISSECTION = { + 'pcap_ct': ("'pcap-ct' is a libpcap binding and performs no protocol dissection, so there " + 'is no protocol layer to read'), + 'pypcap': ("'pypcap' is a libpcap binding and performs no protocol dissection, so there " + 'is no protocol layer to read'), +} + + +def engine_name(module: 'str') -> 'str': + """``__engine_name__`` of the engine class driving :mod:`pcapkit.toolkit.{module}`.""" + from pcapkit.foundation import engines + + #: Engine class per toolkit module. Spelled out rather than guessed from the + #: module name, since ``PCAP_CT`` is neither ``Pcap_Ct`` nor ``PCAP-CT``. + classes = { + 'pcap_ct': engines.PCAP_CT, 'pypcap': engines.PyPCAP, + 'pypcapfile': engines.PyPCAPFile, 'pyshark': engines.PyShark, + 'scapy': engines.Scapy, + } + return classes[module].__engine_name__ + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class EngineNamePrefixTests(unittest.TestCase): + """Each message is pinned whole, and its prefix is the engine's own name.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def assertPrefixed(self, module: 'str', message: 'str') -> 'None': + """``message`` leads with :mod:`pcapkit.toolkit.{module}`'s ``__engine_name__``.""" + prefix = '%s: ' % engine_name(module) + self.assertTrue(message.startswith(prefix), '%r does not lead with %r' % (message, prefix)) + + def test_the_pcap_bindings_refuse_under_their_own_name(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.utilities.exceptions import UnsupportedCall + + for module in ('pcap_ct', 'pypcap'): + toolkit = importlib.import_module('pcapkit.toolkit.%s' % module) + suffix = NO_DISSECTION[module] + for name, call, description in ( + ('ipv4_reassembly', lambda t=toolkit: t.ipv4_reassembly(b'raw', count=1), + 'IPv4 reassembly is not supported'), + ('ipv6_reassembly', lambda t=toolkit: t.ipv6_reassembly(b'raw', count=1), + 'IPv6 reassembly is not supported'), + ('tcp_reassembly', lambda t=toolkit: t.tcp_reassembly(b'raw', count=1), + 'TCP reassembly is not supported'), + ('tcp_traceflow', lambda t=toolkit: t.tcp_traceflow( + b'raw', 1.0, data_link=LinkType.ETHERNET, count=1), + 'TCP flow tracing is not supported'), + ): + with self.subTest(module=module, function=name): + with self.assertRaises(UnsupportedCall) as caught: + call() + message = str(caught.exception) + self.assertEqual(message, '%s: %s: %s' % (engine_name(module), description, + suffix)) + self.assertPrefixed(module, message) + + def test_pcap_ct_is_named_the_way_its_engine_names_itself(self) -> None: + """The reported regression: the engine is ``PCAP_CT``, not ``pcap-ct``.""" + from pcapkit.toolkit.pcap_ct import ipv4_reassembly + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall) as caught: + ipv4_reassembly(b'raw', count=1) + self.assertEqual(str(caught.exception).split(':')[0], 'PCAP_CT') + self.assertEqual(engine_name('pcap_ct'), 'PCAP_CT') + + def test_pypcapfile_address_rejections(self) -> None: + from pcapkit.toolkit import pypcapfile as toolkit + from pcapkit.utilities.exceptions import ProtocolError + + for parse, label, value in ( + (toolkit._parse_ipv4_address, 'IPv4 address', 5), + (toolkit._parse_ipv4_address, 'IPv4 address', None), + (toolkit._parse_ipv4_address, 'IPv4 address', b'nope!'), + (toolkit._parse_ipv4_address, 'IPv4 address', b'\xff\xfe no ascii'), + (toolkit._parse_mac_address, 'MAC address', 5), + (toolkit._parse_mac_address, 'MAC address', None), + (toolkit._parse_mac_address, 'MAC address', b'\xff\xfe no ascii'), + ): + with self.subTest(parse=parse.__name__, value=value): + with self.assertRaises(ProtocolError) as caught: + parse(value) + self.assertEqual(str(caught.exception), + 'PyPCAPFile: invalid %s: %r' % (label, value)) + self.assertPrefixed('pypcapfile', str(caught.exception)) + + def test_pypcapfile_ipv6_reassembly_refusal(self) -> None: + from pcapkit.toolkit.pypcapfile import ipv6_reassembly + from pcapkit.utilities.exceptions import UnsupportedCall + + with self.assertRaises(UnsupportedCall) as caught: + ipv6_reassembly(make_packet(ipv6_frame(6)), count=1) + self.assertEqual(str(caught.exception), + "PyPCAPFile: IPv6 reassembly is not supported: 'pypcapfile' has no IPv6 " + 'decoder, so no IPv6 fragment header can be read') + self.assertPrefixed('pypcapfile', str(caught.exception)) + + def test_pypcapfile_tcp_over_ipv6_warning(self) -> None: + from pcapkit.toolkit.pypcapfile import tcp_reassembly + from pcapkit.utilities.warnings import AttributeWarning + + header = [types.SimpleNamespace(ns_resolution=False)] + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + self.assertIsNone(tcp_reassembly(make_packet(ipv6_frame(6), header=header), count=7)) + messages = [str(item.message) for item in caught + if issubclass(item.category, AttributeWarning)] + self.assertEqual(messages, [ + 'PyPCAPFile: frame 7: TCP over IPv6 is left out of TCP reassembly and flow tracing, ' + "as 'pypcapfile' has no IPv6 decoder; later such frames of this capture are left out " + 'without another warning']) + self.assertPrefixed('pypcapfile', messages[0]) + + def test_pyshark_field_rejections(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError) as caught: + toolkit._parse_flag(LayerField('x'), 'tcp.flags_fin') + self.assertEqual(str(caught.exception), + "PyShark: invalid boolean field tcp.flags_fin: 'x'") + self.assertPrefixed('pyshark', str(caught.exception)) + + with self.assertRaises(ProtocolError) as caught: + toolkit._parse_epoch(LayerField('Jul 14, 2017')) + self.assertEqual(str(caught.exception), + "PyShark: invalid frame.time_epoch: 'Jul 14, 2017'") + self.assertPrefixed('pyshark', str(caught.exception)) + + def test_pyshark_missing_frame_layer(self) -> None: + from pcapkit.toolkit import pyshark as toolkit + from pcapkit.utilities.exceptions import ProtocolError + + packet = FakePySharkPacket(comment='a packet comment') + packet.layers = packet.layers[1:] + with self.assertRaises(ProtocolError) as caught: + toolkit.tcp_traceflow(packet) + self.assertEqual(str(caught.exception), 'PyShark: packet 12 has no frame layer') + self.assertPrefixed('pyshark', str(caught.exception)) + + def test_scapy_missing_dependency_warning(self) -> None: + """The import-time warning, with |scapy|_ mocked away so it fires either way.""" + from pcapkit.utilities.warnings import ScapyWarning + + # Put a real import back afterwards, as + # ``test_scapy_unit.test_import_without_scapy_sets_none_and_warns`` + # does: cleanups run last-in first-out, so the Scapy-less copy is + # purged and then re-imported (GitHub issue #1065). + self.addCleanup(importlib.import_module, 'pcapkit.toolkit.scapy') + self.addCleanup(purge_modules, ['pcapkit.toolkit.scapy']) + purge_modules(['pcapkit.toolkit.scapy']) + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): # type: ignore[no-untyped-def] + if name == 'scapy': + raise ModuleNotFoundError("No module named 'scapy'") + return real_import(name, *args, **kwargs) + + with mock.patch('builtins.__import__', side_effect=fake_import), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + module = importlib.import_module('pcapkit.toolkit.scapy') + + self.assertIsNone(module.scapy) + messages = [str(item.message) for item in caught + if issubclass(item.category, ScapyWarning)] + self.assertEqual(messages, ["Scapy: dependency package 'scapy' not found"]) + self.assertPrefixed('scapy', messages[0]) + + def test_every_prefix_is_read_off_the_engine_class(self) -> None: + """No adapter spells its engine's name itself.""" + for module in ('pcap_ct', 'pypcap', 'pypcapfile', 'pyshark', 'scapy'): + with self.subTest(module=module): + toolkit = importlib.import_module('pcapkit.toolkit.%s' % module) + self.assertEqual(toolkit._ENGINE, engine_name(module)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pcap_ct_unit.py b/tests/toolkit/test_pcap_ct_unit.py index 4945120130..92aaa57757 100644 --- a/tests/toolkit/test_pcap_ct_unit.py +++ b/tests/toolkit/test_pcap_ct_unit.py @@ -77,6 +77,10 @@ def test_unsupported_adapters_refuse_loudly(self) -> None: # an oversight rather than as a property of a libpcap binding self.assertIn('no protocol dissection', str(caught.exception)) self.assertIn('pcap-ct', str(caught.exception)) + # the engine's own name leads the message, spelled as + # ``PCAP_CT.__engine_name__`` spells it (#2304) + self.assertTrue(str(caught.exception).startswith('PCAP_CT: '), + str(caught.exception)) def test_module_exports_the_documented_surface(self) -> None: from pcapkit.toolkit import pcap_ct diff --git a/tests/toolkit/test_pypcap_unit.py b/tests/toolkit/test_pypcap_unit.py index 30d0ac00a9..c1388125dc 100644 --- a/tests/toolkit/test_pypcap_unit.py +++ b/tests/toolkit/test_pypcap_unit.py @@ -64,6 +64,10 @@ def test_unsupported_adapters_refuse_loudly(self) -> None: with self.assertRaises(UnsupportedCall) as caught: call() self.assertIn('no protocol dissection', str(caught.exception)) + # the engine's own name leads the message, spelled as + # ``PyPCAP.__engine_name__`` spells it (#2304) + self.assertTrue(str(caught.exception).startswith('PyPCAP: '), + str(caught.exception)) def test_module_exports_the_documented_surface(self) -> None: from pcapkit.toolkit import pypcap diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py index bc37011a27..6fe62fdc9c 100644 --- a/tests/toolkit/test_pypcapfile_unit.py +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -498,8 +498,9 @@ def test_tcp_over_ipv6_warns_once_per_capture(self) -> None: self.assertIsNone(tcp_reassembly(packet, count=count)) self.assertIsNone(tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=count)) self.assertEqual([str(item.message) for item in caught if item.category is AttributeWarning], [ - f"Frame {count}: TCP over IPv6 is left out of TCP reassembly and flow tracing, as 'pypcapfile' " - 'has no IPv6 decoder; later such frames of this capture are left out without another warning' + f"PyPCAPFile: frame {count}: TCP over IPv6 is left out of TCP reassembly and flow tracing, " + "as 'pypcapfile' has no IPv6 decoder; later such frames of this capture are left out " + 'without another warning' for count in (3, 4)]) def test_either_tcp_adapter_alone_warns_of_tcp_over_ipv6(self) -> None: @@ -516,8 +517,9 @@ def test_either_tcp_adapter_alone_warns_of_tcp_over_ipv6(self) -> None: for count in (1, 2): self.assertIsNone(adapter(make_packet(ipv6_frame(6), header=header), count=count, **kwargs)) - self.assertEqual([(item.category, str(item.message)[:37]) for item in caught], - [(AttributeWarning, 'Frame 1: TCP over IPv6 is left out of')]) + self.assertEqual([(item.category, str(item.message)[:49]) for item in caught], + [(AttributeWarning, + 'PyPCAPFile: frame 1: TCP over IPv6 is left out of')]) def test_tcp_over_ipv6_is_read_off_a_whole_ipv6_header_only(self) -> None: # #1513: the Next Header of an IPv6 frame, whose 40-octet header was captured @@ -578,7 +580,8 @@ class Record(ctypes.Structure): # as pcapfile.structs.pcap_packet record = Record(ctypes.pointer(header)) record.packet = ipv6_frame(6) self.assertIsNone(tcp_reassembly(record, count=count)) - self.assertEqual([str(item.message)[:8] for item in caught], ['Frame 1:', 'Frame 3:']) + self.assertEqual([str(item.message)[:20] for item in caught], + ['PyPCAPFile: frame 1:', 'PyPCAPFile: frame 3:']) def test_the_captures_warned_of_tcp_over_ipv6_are_bounded(self) -> None: from pcapkit.toolkit import pypcapfile as toolkit @@ -593,7 +596,8 @@ def test_the_captures_warned_of_tcp_over_ipv6_are_bounded(self) -> None: toolkit.tcp_reassembly(make_packet(ipv6_frame(6), header=header), count=1) toolkit.tcp_reassembly(make_packet(ipv6_frame(6), header=headers[0]), count=2) # forgotten toolkit.tcp_reassembly(make_packet(ipv6_frame(6), header=headers[-1]), count=3) # remembered - self.assertEqual([str(item.message)[:8] for item in caught], ['Frame 1:'] * 33 + ['Frame 2:']) + self.assertEqual([str(item.message)[:20] for item in caught], + ['PyPCAPFile: frame 1:'] * 33 + ['PyPCAPFile: frame 2:']) self.assertEqual(len(toolkit._ipv6_tcp_warned), 32) # pylint: disable=protected-access diff --git a/tests/toolkit/test_pyshark_unit.py b/tests/toolkit/test_pyshark_unit.py index 270d9ae24e..d4d246e0a1 100644 --- a/tests/toolkit/test_pyshark_unit.py +++ b/tests/toolkit/test_pyshark_unit.py @@ -517,7 +517,7 @@ def test_parse_flag_rejects_anything_else(self) -> None: with self.assertRaises(ProtocolError) as ctx: toolkit._parse_flag(value, 'tcp.flags_fin') self.assertEqual(str(ctx.exception), - f'invalid PyShark boolean field tcp.flags_fin: {value!r}') + f'PyShark: invalid boolean field tcp.flags_fin: {value!r}') def test_tcp_traceflow_reads_tshark_4_flags(self) -> None: from pcapkit.toolkit import pyshark as toolkit @@ -587,7 +587,8 @@ def test_unparsable_timestamp_raises(self) -> None: packet.frame_info.time_epoch = LayerField('Jul 14, 2017') with self.assertRaises(ProtocolError) as ctx: toolkit.tcp_traceflow(packet) - self.assertEqual(str(ctx.exception), "invalid PyShark frame.time_epoch: 'Jul 14, 2017'") + self.assertEqual(str(ctx.exception), + "PyShark: invalid frame.time_epoch: 'Jul 14, 2017'") def test_seq_and_ack_are_absolute(self) -> None: """tshark's ``tcp.seq``/``tcp.ack`` are relative by default; the @@ -776,7 +777,7 @@ def test_a_packet_without_a_frame_layer_raises(self) -> None: packet.layers = packet.layers[1:] with self.assertRaises(ProtocolError) as ctx: toolkit.tcp_traceflow(packet) - self.assertEqual(str(ctx.exception), 'PyShark packet 12 has no frame layer') + self.assertEqual(str(ctx.exception), 'PyShark: packet 12 has no frame layer') @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')