fix(interface): word reassembly, trace, output-format and engine-fallback messages in the house form (#2303)

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

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

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

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

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

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
+<pcapkit.foundation.extraction.Extractor.make_name>` and
+:meth:`TraceFlowBase.make_fout
+<pcapkit.foundation.traceflow.traceflow.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()