From f90b62aa001eedc535710d06d8165121f0ade82c Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 26 Oct 2026 17:01:46 -0400 Subject: [PATCH] fix(interface): word reassembly, trace, output-format and engine-fallback messages in the house form (#2303) reassemble and trace raised "Unsupported ... protocol: foo", capitalised and without the value's repr; follow_tcp_stream warned "fallback to default engine" where Extractor says "using default engine instead", and left the engine's name unquoted when it has no reassembly adapter; TraceFlowBase.make_fout warned a capitalised "Unsupported output format: ..."; and Extractor.make_name raised "unknown output format: foo" with the value unquoted -- which, for an empty format, ends the message at the colon with nothing after it. Lowercase each message and quote the rejected value with !r, and use "using default engine instead" for every engine fallback. Test: tests/interface/test_message_form_2303_unit.py pins all seven messages. --- pcapkit/foundation/extraction.py | 6 +- pcapkit/foundation/traceflow/traceflow.py | 2 +- pcapkit/interface/core.py | 4 +- pcapkit/interface/misc.py | 4 +- ..._extraction_engine_fallback_format_unit.py | 2 +- .../interface/test_message_form_2303_unit.py | 136 ++++++++++++++++++ 6 files changed, 145 insertions(+), 9 deletions(-) create mode 100644 tests/interface/test_message_form_2303_unit.py diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index b94e95a1c0..39404fb107 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -933,7 +933,7 @@ def _select_engine(self) -> 'Optional[Type[EngineBase]]': self._exnam = 'default' # using default/pcapkit engine if self._exnam not in ('default', 'pcapkit'): - warn(f'unsupported extraction engine: {self._exnam}; ' + warn(f'unsupported extraction engine: {self._exnam!r}; ' 'using default engine instead', EngineWarning, stacklevel=stacklevel()) self._exnam = 'default' # using default/pcapkit engine return None @@ -1038,7 +1038,7 @@ def make_name(cls, fin: 'str | IO[bytes]' = 'in.pcap', fout: 'str' = 'out', entry = cls.__output__.get(fmt) registered = entry[1] if entry is not None else None if registered is None: - raise FormatError(f'unknown output format: {fmt}') + raise FormatError(f'unknown output format: {fmt!r}') # NOTE: ``__output__`` spells its extensions with the leading dot, # and so does every ``ext=`` handed to @@ -1502,7 +1502,7 @@ def __init__(self, if not self._flag_q: output, ext = self.__output__[fmt] if ext is None: - warn(f'Unsupported output format: {fmt}; disabled file output feature', + warn(f'unsupported output format: {fmt!r}; disabled file output feature', FormatWarning, stacklevel=stacklevel()) if isinstance(output, ModuleDescriptor): output = output.klass diff --git a/pcapkit/foundation/traceflow/traceflow.py b/pcapkit/foundation/traceflow/traceflow.py index 8226e8ba6d..f433b159c0 100644 --- a/pcapkit/foundation/traceflow/traceflow.py +++ b/pcapkit/foundation/traceflow/traceflow.py @@ -320,7 +320,7 @@ def make_fout(cls, fout: 'str' = './tmp', fmt: 'str' = 'pcap') -> 'tuple[Type[Du """ output, ext = cls.__output__[fmt] if ext is None: - warn(f'Unsupported output format: {fmt}; disabled file output feature', + warn(f'unsupported output format: {fmt!r}; disabled file output feature', FormatWarning, stacklevel=stacklevel()) if isinstance(output, ModuleDescriptor): output = output.klass diff --git a/pcapkit/interface/core.py b/pcapkit/interface/core.py index 36667a9e61..35e9a1e4d5 100644 --- a/pcapkit/interface/core.py +++ b/pcapkit/interface/core.py @@ -224,7 +224,7 @@ def reassemble(protocol: 'str | Type[ProtocolBase]', strict: 'bool' = False, return IPv6_Reassembly(strict=strict, timeout=timeout) if protocol == 'TCP': return TCP_Reassembly(strict=strict, timeout=timeout) - raise FormatError(f'Unsupported reassembly protocol: {protocol}') + raise FormatError(f'unsupported reassembly protocol: {protocol!r}') def trace(protocol: 'str | Type[ProtocolBase]', fout: 'Optional[str]', @@ -261,4 +261,4 @@ def trace(protocol: 'str | Type[ProtocolBase]', fout: 'Optional[str]', if protocol == 'TCP': return TCP_TraceFlow(fout=fout, format=format, byteorder=byteorder, nanosecond=nanosecond, bidirectional=bidirectional) - raise FormatError(f'Unsupported flow tracing protocol: {protocol}') + raise FormatError(f'unsupported flow tracing protocol: {protocol!r}') diff --git a/pcapkit/interface/misc.py b/pcapkit/interface/misc.py index f9e7980d1e..35fa6494f6 100644 --- a/pcapkit/interface/misc.py +++ b/pcapkit/interface/misc.py @@ -122,7 +122,7 @@ def follow_tcp_stream(fin: 'Optional[str]' = None, verbose: 'bool' = False, # all and disable TCP flow tracing outright, so ``extraction.trace`` below # would raise instead of yielding streams. if engine is not None and engine.lower() in ('pyshark', 'pypcap', 'pcap_ct'): - warn(f'unsupported extraction engine: {engine}; fallback to default engine', + warn(f'unsupported extraction engine: {engine!r}; using default engine instead', EngineWarning, stacklevel=stacklevel()) engine = None @@ -182,7 +182,7 @@ def follow_tcp_stream(fin: 'Optional[str]' = None, verbose: 'bool' = False, # to the pcapkit adapter is exactly the #399 failure mode -- a wrong or empty # result indistinguishable from a real one -- so warn and return no streams # rather than reassemble frames whose shape we cannot parse. - warn(f'unsupported extraction engine for TCP stream following: {exeng.name}; ' + warn(f'unsupported extraction engine for TCP stream following: {exeng.name!r}; ' 'returning no streams', EngineWarning, stacklevel=stacklevel()) return () diff --git a/tests/foundation/test_extraction_engine_fallback_format_unit.py b/tests/foundation/test_extraction_engine_fallback_format_unit.py index ca58522723..c80014d8ad 100644 --- a/tests/foundation/test_extraction_engine_fallback_format_unit.py +++ b/tests/foundation/test_extraction_engine_fallback_format_unit.py @@ -102,7 +102,7 @@ def test_unknown_engine_keeps_pcap_output(self) -> None: self.assertEqual(extractor.format, 'pcap') self.assertEqual(_read(os.path.join(tmp, 'out.pcap')), _read(fin)) self.assertEqual([str(m) for m in caught if isinstance(m, EngineWarning)], - ['unsupported extraction engine: no-such-engine; using default engine instead']) + ["unsupported extraction engine: 'no-such-engine'; using default engine instead"]) def test_trace_format_is_kept(self) -> None: from pcapkit.utilities.warnings import FormatWarning diff --git a/tests/interface/test_message_form_2303_unit.py b/tests/interface/test_message_form_2303_unit.py new file mode 100644 index 0000000000..acca380ebb --- /dev/null +++ b/tests/interface/test_message_form_2303_unit.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +"""Reassembly, trace, engine-fallback and output-format messages in the house form. C.f. #2303. + +Every rejection message here is lowercase and quotes the rejected value with +:func:`repr`, and every engine fallback is worded ``using default engine +instead``. Guards against :func:`~pcapkit.interface.core.reassemble` and +:func:`~pcapkit.interface.core.trace` raising a capitalised +``Unsupported ... protocol: foo``; against +:func:`~pcapkit.interface.misc.follow_tcp_stream` wording its engine fallback +otherwise than :class:`~pcapkit.foundation.extraction.Extractor` does, or +naming the engine it has no reassembly adapter for unquoted; and against +:meth:`Extractor.make_name +` and +:meth:`TraceFlowBase.make_fout +` leaving the +rejected format bare -- which for the empty format ends the message at the +colon with nothing after it. Each message is pinned whole. + +""" + +import os +import tempfile +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class, sample_path + + +class TestMessageForm(unittest.TestCase): + """Each message is lowercase and quotes the rejected value with :func:`repr`.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_reassemble_unsupported_protocol(self) -> None: + from pcapkit.interface.core import reassemble + from pcapkit.utilities.exceptions import FormatError + + with self.assertRaises(FormatError) as context: + reassemble('foo') + self.assertEqual(str(context.exception), "unsupported reassembly protocol: 'foo'") + + def test_trace_unsupported_protocol(self) -> None: + from pcapkit.interface.core import trace + from pcapkit.utilities.exceptions import FormatError + + with self.assertRaises(FormatError) as context: + trace('foo', fout=None, format=None) + self.assertEqual(str(context.exception), "unsupported flow tracing protocol: 'foo'") + + def test_follow_tcp_stream_engine_fallback(self) -> None: + from pcapkit.interface.misc import follow_tcp_stream + from pcapkit.utilities.warnings import EngineWarning + + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + follow_tcp_stream(fin=sample_path('in.pcap'), fout=tmp, engine='pyshark') + self.assertEqual([str(w.message) for w in caught if issubclass(w.category, EngineWarning)], + ["unsupported extraction engine: 'pyshark'; using default engine instead"]) + + def test_follow_tcp_stream_engine_without_a_reassembly_adapter(self) -> None: + from pcapkit.foundation.engines.engine import Engine + from pcapkit.foundation.extraction import Extractor + from pcapkit.interface.misc import follow_tcp_stream + from pcapkit.utilities.warnings import EngineWarning + + class UnitEngine(Engine[object]): # no ``engine=``, so not registered here + __engine_name__ = 'UnitEngine' + __engine_module__ = __name__ + + def run(self) -> 'None': + pass + + def read_frame(self) -> 'object': + # ``Extractor.record_frames`` reads until EOF, and + # ``follow_tcp_stream`` runs the extraction itself, so there is + # no ``auto=False`` to stop it short of this. + raise EOFError + + # ``__engine__`` is one global table shared by every engine class, so + # the registration is rolled back. + with tempfile.TemporaryDirectory() as tmp, mock.patch.dict(Extractor.__engine__), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + Extractor.register_engine('unit-engine', UnitEngine) + streams = follow_tcp_stream(fin=sample_path('in.pcap'), fout=tmp, + engine='unit-engine') + self.assertEqual(streams, ()) + self.assertEqual([str(w.message) for w in caught if issubclass(w.category, EngineWarning)], + ["unsupported extraction engine for TCP stream following: " + "'UnitEngine'; returning no streams"]) + + def test_extractor_engine_fallback(self) -> None: + from pcapkit.foundation.extraction import Extractor + from pcapkit.utilities.warnings import EngineWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + Extractor(fin=sample_path('in.pcap'), nofile=True, store=False, engine='no-such-engine') + self.assertEqual([str(w.message) for w in caught if issubclass(w.category, EngineWarning)], + ["unsupported extraction engine: 'no-such-engine'; using default engine instead"]) + + def test_make_name_unknown_output_format(self) -> None: + from pcapkit.foundation.extraction import Extractor + from pcapkit.utilities.exceptions import FormatError + + # The only output-format rejection an ``Extractor`` reaches. The + # ``FormatWarning`` in ``Extractor.__init__`` fires on ``ext is None``, + # which ``make_name`` -- called earlier in the same constructor, and + # only when ``nofile=False``, which is also the warning's condition -- + # has already raised on, so that warning is unreachable. + for fmt, pinned in (('no-such-format', "unknown output format: 'no-such-format'"), + ('', "unknown output format: ''")): + with self.subTest(format=fmt), tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(FormatError) as context: + Extractor(fin=sample_path('in.pcap'), fout=os.path.join(tmp, 'out'), + format=fmt, auto=False) + self.assertEqual(str(context.exception), pinned) + + def test_make_fout_unsupported_output_format(self) -> None: + from pcapkit.foundation.traceflow.tcp import TCP + from pcapkit.utilities.warnings import FormatWarning + + # ``__output__`` is a defaultdict, so the lookup inserts ``'foo'``; put + # the registry back afterwards. + with tempfile.TemporaryDirectory() as tmp, mock.patch.dict(TCP.__output__), \ + warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + TCP.make_fout(tmp, 'foo') + self.assertEqual([str(w.message) for w in caught if issubclass(w.category, FormatWarning)], + ["unsupported output format: 'foo'; disabled file output feature"]) + + +if __name__ == '__main__': + unittest.main()