fix(foundation): keep a user's verbose callback on every engine (#1874)

.github/workflows/unit-tests.yml

@@ -1030,11 +1030,11 @@ jobs:
           fi
 
           case "$ENGINE" in
-            Scapy) TEST_PATHS="tests/toolkit/test_scapy_unit.py tests/foundation/engines/test_scapy_engine.py tests/protocols/transport/test_sctp_unit.py" ;;
-            PyShark) TEST_PATHS="tests/toolkit/test_pyshark_unit.py tests/foundation/engines/test_pyshark_engine.py" ;;
-            PyPCAPFile) TEST_PATHS="tests/toolkit/test_pypcapfile_unit.py tests/foundation/engines/test_pypcapfile_engine.py" ;;
-            PyPCAP) TEST_PATHS="tests/toolkit/test_pypcap_unit.py tests/foundation/engines/test_pypcap_engine.py tests/foundation/engines/test_pcap_backend.py" ;;
-            PCAP_CT) TEST_PATHS="tests/toolkit/test_pcap_ct_unit.py tests/foundation/engines/test_pcap_ct_engine.py tests/foundation/engines/test_pcap_backend.py" ;;
+            Scapy) TEST_PATHS="tests/toolkit/test_scapy_unit.py tests/foundation/engines/test_scapy_engine.py tests/protocols/transport/test_sctp_unit.py tests/foundation/engines/test_engine_verbose_callback_1874_unit.py" ;;
+            PyShark) TEST_PATHS="tests/toolkit/test_pyshark_unit.py tests/foundation/engines/test_pyshark_engine.py tests/foundation/engines/test_engine_verbose_callback_1874_unit.py" ;;
+            PyPCAPFile) TEST_PATHS="tests/toolkit/test_pypcapfile_unit.py tests/foundation/engines/test_pypcapfile_engine.py tests/foundation/engines/test_engine_verbose_callback_1874_unit.py" ;;
+            PyPCAP) TEST_PATHS="tests/toolkit/test_pypcap_unit.py tests/foundation/engines/test_pypcap_engine.py tests/foundation/engines/test_pcap_backend.py tests/foundation/engines/test_engine_verbose_callback_1874_unit.py" ;;
+            PCAP_CT) TEST_PATHS="tests/toolkit/test_pcap_ct_unit.py tests/foundation/engines/test_pcap_ct_engine.py tests/foundation/engines/test_pcap_backend.py tests/foundation/engines/test_engine_verbose_callback_1874_unit.py" ;;
           esac
           echo "Running: $TEST_PATHS"
           # shellcheck disable=SC2086

docs/source/ext.rst

@@ -512,9 +512,13 @@ The following code snippet shows how to create a new engine class:
            # check if there's any extraction protocol and/or layer thresholds
            # set by the user (some engines may not support this feature); or
            # to set up the verbose handler for the underlying engine module,
-           # as in the following code snippet.
+           # as in the following code snippet. Install it only when the user
+           # asked for the default output, i.e., `verbose=True`, which sets
+           # `ext._vfunc_default`: a callable given as `verbose=` is the
+           # user's own handler, already stored as `ext._vfunc`, so it must
+           # be kept rather than replaced.
 
-           if ext._flag_v:  # check if verbose mode is enabled
+           if ext._vfunc_default:  # check if the default verbose output is requested
                # NOTE: PyPCAPKit has its own bundled supporting functions for
                # third-party libraries, e.g., the `packet2chain` function is
                # to convert a given Packet instance of the Scapy library to a

docs/source/pcapkit/foundation/extraction.rst

@@ -89,6 +89,7 @@ and extracts parameters from a PCAP file.
 
    .. autoattribute:: _eof_mark
    .. autoattribute:: _vfunc
+   .. autoattribute:: _vfunc_default
 
    .. autoattribute:: _exnam
    .. autoattribute:: _exeng

pcapkit/foundation/engines/dpkt.py

@@ -454,11 +454,11 @@ def run(self) -> 'None':
                  "'context=' is ignored",
                  AttributeWarning, stacklevel=stacklevel())
 
-        # setup verbose handler
-        if ext._flag_v:
+        # setup verbose handler, unless the caller gave one of their own (#1874)
+        if ext._vfunc_default:
             from pcapkit.toolkit.dpkt import packet2chain  # isort:skip
             ext._vfunc = lambda e, f: print(
-                f'Frame {e._frnum:>3d}: {packet2chain(f)}'  # pylint: disable=protected-access
+                f'Frame {e._frnum:>3d}: {packet2chain(cast("DPKTPacket", f))}'  # pylint: disable=protected-access
             )
 
         # extract global header

pcapkit/foundation/engines/pcap_ct.py

@@ -399,11 +399,11 @@ def run(self) -> 'None':
         self._handle = cast('Handle', self._expkg.pcap(name=ext._ifnm, promisc=False))
         self._dlink = Enum_LinkType.get(self._handle.datalink())
 
-        # setup verbose handler
-        if ext._flag_v:
+        # setup verbose handler, unless the caller gave one of their own (#1874)
+        if ext._vfunc_default:
             from pcapkit.toolkit.pcap_ct import packet2chain  # isort:skip
             ext._vfunc = lambda e, f: print(
-                f'Frame {e._frnum:>3d}: {packet2chain(f[1], data_link=self._dlink)}'  # pylint: disable=protected-access
+                f'Frame {e._frnum:>3d}: {packet2chain(cast("RawFrame", f)[1], data_link=self._dlink)}'  # pylint: disable=protected-access
             )  # pylint: disable=logging-fstring-interpolation
 
         # extract & analyse file

pcapkit/foundation/engines/pypcap.py

@@ -312,11 +312,11 @@ def run(self) -> 'None':
         handle = cast('Handle', self._expkg.pcap(name=ext._ifnm, promisc=False))
         self._dlink = Enum_LinkType.get(handle.datalink())
 
-        # setup verbose handler
-        if ext._flag_v:
+        # setup verbose handler, unless the caller gave one of their own (#1874)
+        if ext._vfunc_default:
             from pcapkit.toolkit.pypcap import packet2chain  # isort:skip
             ext._vfunc = lambda e, f: print(
-                f'Frame {e._frnum:>3d}: {packet2chain(f[1], data_link=self._dlink)}'  # pylint: disable=protected-access
+                f'Frame {e._frnum:>3d}: {packet2chain(cast("RawFrame", f)[1], data_link=self._dlink)}'  # pylint: disable=protected-access
             )  # pylint: disable=logging-fstring-interpolation
 
         # extract & analyse file

pcapkit/foundation/engines/pypcapfile.py

@@ -220,11 +220,11 @@ def run(self) -> 'None':
         self._dlink = Enum_LinkType.get(sfile.header.ll_type)
         self._declf = self._get_decoder(sfile.header.ll_type)
 
-        # setup verbose handler
-        if ext._flag_v:
+        # setup verbose handler, unless the caller gave one of their own (#1874)
+        if ext._vfunc_default:
             from pcapkit.toolkit.pypcapfile import packet2chain  # isort:skip
             ext._vfunc = lambda e, f: print(
-                f'Frame {e._frnum:>3d}: {packet2chain(f, data_link=self._dlink)}'  # pylint: disable=protected-access
+                f'Frame {e._frnum:>3d}: {packet2chain(cast("PCAPFilePacket", f), data_link=self._dlink)}'  # pylint: disable=protected-access
             )  # pylint: disable=logging-fstring-interpolation
 
         # extract & analyse file

pcapkit/foundation/engines/pyshark.py

@@ -254,12 +254,13 @@ def run(self) -> 'None':
                  f"so 'ipv4={ext._ipv4}', 'ipv6={ext._ipv6}' and 'tcp={ext._tcp}' will be ignored",
                  AttributeWarning, stacklevel=stacklevel())
 
-        # setup verbose handler; a packet always has a frame layer, since
-        # read_frame skips every record that does not (#1515)
-        if ext._flag_v:
+        # setup verbose handler, unless the caller gave one of their own
+        # (#1874); a packet always has a frame layer, since read_frame skips
+        # every record that does not (#1515)
+        if ext._vfunc_default:
             ext._vfunc = lambda e, f: print(
                 f'Frame {e._frnum:>3d}: '  # pylint: disable=protected-access
-                f'{cast("BaseLayer", _frame_layer(f)).protocols}'
+                f'{cast("BaseLayer", _frame_layer(cast("PySharkPacket", f))).protocols}'
             )
 
         # extract global header

pcapkit/foundation/engines/scapy.py

@@ -383,11 +383,11 @@ def run(self) -> 'None':
                  "'context=' is ignored",
                  AttributeWarning, stacklevel=stacklevel())
 
-        # setup verbose handler
-        if ext._flag_v:
+        # setup verbose handler, unless the caller gave one of their own (#1874)
+        if ext._vfunc_default:
             from pcapkit.toolkit.scapy import packet2chain  # isort:skip
             ext._vfunc = lambda e, f: print(
-                f'Frame {e._frnum:>3d}: {packet2chain(f)}'  # pylint: disable=protected-access
+                f'Frame {e._frnum:>3d}: {packet2chain(cast("ScapyPacket", f))}'  # pylint: disable=protected-access
             )
 
         # extract global header

pcapkit/foundation/extraction.py

@@ -85,6 +85,9 @@
                    PCAPFilePacket, tuple[float, bytes]]
 
     Protocols = Union[str, ProtocolBase, Type[ProtocolBase]]
+    # NOTE: every caller discards the callback's result, so it is typed ``Any``:
+    # any two-argument callback is accepted, whatever it returns, e.g.
+    # ``lambda e, f: sys.stdout.write(str(f))``.
     VerboseHandler = Callable[['Extractor', Packet], Any]
 
 __all__ = ['Extractor']
@@ -124,7 +127,9 @@ class Extractor(Generic[_P]):
        so a positional argument after ``reasm_store`` has to move to a
        keyword. Flows are traced bidirectionally by default. :attr:`engine`
        is typed :class:`~pcapkit.foundation.engines.engine.EngineBase`, the
-       base every engine derives from (:issue:`1022`).
+       base every engine derives from (:issue:`1022`). A callback given as
+       ``verbose`` is called on every engine, where the third-party engines
+       replaced it with their own printer (:issue:`1874`).
 
     """
     if TYPE_CHECKING:
@@ -161,8 +166,10 @@ class Extractor(Generic[_P]):
         _flag_r: 'bool'
         #: Trace flag. It indicates if the flow tracing is enabled.
         _flag_t: 'bool'
-        #: Verbose flag. This is used to determine if the verbose callback
-        #: function should be called at each frame.
+        #: Verbose flag. It indicates if verbose output is enabled, whether by
+        #: ``verbose=True`` or by a callback, which
+        #: :attr:`~pcapkit.foundation.extraction.Extractor._vfunc_default`
+        #: tells apart.
         _flag_v: 'bool'
         #: No EOF flag. It is useful when the input is a live capture on a pipe or
         #: on standard input, where reaching the end of what has arrived so far
@@ -178,7 +185,14 @@ class Extractor(Generic[_P]):
         _flag_s: 'bool'
 
         #: Verbose callback function.
-        #_vfunc: 'VerboseHandler'
+        _vfunc: 'VerboseHandler'
+        #: Default verbose printer flag. It is set only by ``verbose=True``,
+        #: which asks for the default output, and an engine whose frames are
+        #: not PyPCAPKit's own replaces
+        #: :attr:`~pcapkit.foundation.extraction.Extractor._vfunc` with its own
+        #: printer only when it is set. A callback given as ``verbose`` leaves
+        #: it unset, so every engine keeps that callback (:issue:`1874`).
+        _vfunc_default: 'bool'
 
         #: Frame number.
         _frnum: 'int'
@@ -1279,19 +1293,23 @@ def __init__(self,
         # verbose callback function
         if isinstance(verbose, bool):
             self._flag_v = verbose
+            self._vfunc_default = verbose
             if verbose:
                 # NOTE: ``verbose=True`` and the CLI's ``-v`` are a request for
                 # user-facing output on stdout, not diagnostics -- the frame
                 # chains are a feature of the tool, so they stay on ``print``
                 # rather than becoming log records a consumer has to configure
-                # a handler to see (and on a different stream at that).
+                # a handler to see (and on a different stream at that). Of the
+                # built-in engines only PCAP and PCAPNG keep this callback, so
+                # its frame is theirs; the others install their own.
                 self._vfunc = lambda e, f: print(
-                    f'Frame {e._frnum:>3d}: {f.protochain}'  # pylint: disable=protected-access
+                    f'Frame {e._frnum:>3d}: {cast("Frame | PCAPNG", f).protochain}'  # pylint: disable=protected-access
                 )
             else:
                 self._vfunc = lambda e, f: None
         else:
             self._flag_v = True
+            self._vfunc_default = False
             self._vfunc = verbose
 
         self._frnum = 0   # frame number