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