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')