fix(foundation): keep format='pcap' when an unavailable engine falls back to the default one (#1264)

pcapkit/foundation/extraction.py

@@ -810,6 +810,57 @@ def run(self) -> 'None':  # pylint: disable=inconsistent-return-statements
                 the built-in PCAP-NG engine reads a PCAP-NG input to PCAP output.
 
         :rtype: None
+        """
+        eng = self._select_engine()
+        if eng is not None:
+            logger.debug('using engine %s (%s)', eng.name, eng.module)
+            self._exeng = eng(self)
+            self._exeng.run()
+            self._open_output()
+
+            # start iteration
+            self.record_frames()
+            return
+
+        if self._magic in PCAP_Engine.MAGIC_NUMBER:
+            logger.debug('magic number %r identifies a PCAP file', self._magic)
+            self._exeng = cast('EngineBase[_P]', PCAP_Engine(self))
+        elif self._magic in PCAPNG_Engine.MAGIC_NUMBER:
+            logger.debug('magic number %r identifies a PCAP-NG file', self._magic)
+            self._exeng = cast('EngineBase[_P]', PCAPNG_Engine(self))
+        else:
+            raise FormatError(f'unknown file format: {self._magic!r}')
+
+        # start engine
+        self._exeng.run()
+        self._open_output()
+
+        # start iteration
+        self.record_frames()
+
+    def _select_engine(self) -> 'Optional[Type[EngineBase]]':
+        """Settle which extraction engine will actually run.
+
+        An engine that is unknown, not installed, or rules itself out via
+        :meth:`~pcapkit.foundation.engines.engine.EngineBase.unsupported_reason`
+        is replaced by the default one, and :attr:`self._exnam
+        <pcapkit.foundation.extraction.Extractor._exnam>` is updated to match.
+        :meth:`__init__` calls this before choosing the output format, so that
+        the choice reflects the engine that runs rather than the one requested.
+
+        Calling it again is silent: a fallback has already set the name to
+        ``'default'``, and an engine that passed the checks passes them again.
+
+        Warns:
+            pcapkit.utilities.warnings.EngineWarning: If the extraction engine is not
+                available: its dependency is not installed, its
+                :meth:`~pcapkit.foundation.engines.engine.EngineBase.unsupported_reason`
+                rules it out, or the supplied engine is unknown.
+
+        Returns:
+            The engine class to run, or :data:`None` for the default engine,
+            which :meth:`run` picks from the input's magic number.
+
         """
         logger.debug('requested extraction engine: %s', self._exnam)
 
@@ -833,14 +884,7 @@ def run(self) -> 'None':  # pylint: disable=inconsistent-return-statements
                      EngineWarning, stacklevel=stacklevel())
                 self._exnam = 'default'
             elif self.import_test(eng.module, name=eng.name) is not None:  # type: ignore[arg-type]
-                logger.debug('using engine %s (%s)', eng.name, eng.module)
-                self._exeng = eng(self)
-                self._exeng.run()
-                self._open_output()
-
-                # start iteration
-                self.record_frames()
-                return
+                return eng
             else:
                 # ``import_test`` has already warned that the package is absent;
                 # warning again here would report one problem twice.
@@ -850,22 +894,7 @@ def run(self) -> 'None':  # pylint: disable=inconsistent-return-statements
             warn(f'unsupported extraction engine: {self._exnam}; '
                  'using default engine instead', EngineWarning, stacklevel=stacklevel())
             self._exnam = 'default'  # using default/pcapkit engine
-
-        if self._magic in PCAP_Engine.MAGIC_NUMBER:
-            logger.debug('magic number %r identifies a PCAP file', self._magic)
-            self._exeng = cast('EngineBase[_P]', PCAP_Engine(self))
-        elif self._magic in PCAPNG_Engine.MAGIC_NUMBER:
-            logger.debug('magic number %r identifies a PCAP-NG file', self._magic)
-            self._exeng = cast('EngineBase[_P]', PCAPNG_Engine(self))
-        else:
-            raise FormatError(f'unknown file format: {self._magic!r}')
-
-        # start engine
-        self._exeng.run()
-        self._open_output()
-
-        # start iteration
-        self.record_frames()
+        return None
 
     @staticmethod
     def import_test(engine: 'str', *, name: 'Optional[str]' = None) -> 'Optional[ModuleType]':
@@ -1196,6 +1225,12 @@ def __init__(self,
                   one, i.e. DPKT, Scapy, PyShark, PyPCAP, pcap-ct or PyPCAPFile.
                 * If output file format is not supported.
 
+                In each case the engine is the one that will actually run: a
+                requested engine that is unavailable falls back to the default
+                engine first, which supports PCAP output.
+
+            pcapkit.utilities.warnings.EngineWarning: If the requested extraction
+                engine is not available, c.f. :meth:`run`.
             pcapkit.utilities.warnings.AttributeWarning: If ``trace_analyse`` is
                 requested while the extraction engine is PyShark.
 
@@ -1207,14 +1242,20 @@ def __init__(self,
         if format is None:
             format = 'tree'
 
+        # NOTE: the engine is settled first, because the format checks below and
+        # in the trace settings depend on which engine *runs*: an unavailable
+        # one falls back to the default engine, which writes PCAP perfectly well.
+        self._exnam = cast('Engines', (engine or 'default').lower())  # extract using engine
+        self._select_engine()
+
         # NOTE: the third-party engines' adapters hand the output writer each
         # frame as a plain :obj:`dict` built by ``packet2dict``, which carries
         # neither the frame's octets nor its record header, so there is nothing
         # for the PCAP writer to write. The flow tracer has the same limitation,
         # and is given the same replacement below.
-        if ((engine or 'default').lower() in ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile')
+        if (self._exnam in ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile')
                 and format in ('pcap', 'cap')):
-            warn(f"'Extractor(engine={engine})' does not support 'format={format}'; "
+            warn(f"'Extractor(engine={self._exnam})' does not support 'format={format}'; "
                  "using 'format=\"json\"' instead", FormatWarning, stacklevel=stacklevel())
             format = 'json'
 
@@ -1264,7 +1305,6 @@ def __init__(self,
 
         self._exptl = protocol or 'null'                              # extract til protocol
         self._exlyr = cast('Layers', (layer or 'none').lower())       # extract til layer
-        self._exnam = cast('Engines', (engine or 'default').lower())  # extract using engine
         self._exctx = ContextRegistry.make(context)                   # caller supplied context
 
         if reassembly:

tests/foundation/test_extraction.py

@@ -1067,13 +1067,17 @@ def _traced(self, temp: pathlib.Path, tag: str, **kwargs: object):
         """An ``Extractor`` built for tracing, with ``run`` patched out.
 
         Nothing is extracted, so no engine needs to be installed -- the format
-        substitution under test happens in the constructor. Returns the extractor
-        and the patched module-level ``warn``.
+        substitution under test happens in the constructor. ``_select_engine`` is
+        patched out too, so the requested engine is taken as the one that runs;
+        otherwise an engine missing here would fall back to the default one,
+        which keeps PCAP (#1264). Returns the extractor and the patched
+        module-level ``warn``.
 
         """
         from pcapkit.foundation.extraction import Extractor
 
-        with mock.patch.object(Extractor, 'run'):
+        with mock.patch.object(Extractor, 'run'), \
+                mock.patch.object(Extractor, '_select_engine', return_value=None):
             with mock.patch('pcapkit.foundation.extraction.warn') as warn:
                 extractor = Extractor(str(temp / 'sample.pcap'), str(temp / f'out-{tag}'),
                                       format='json', auto=False, nofile=True, trace=True,

tests/foundation/test_extraction_engine_fallback_format_unit.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+"""PCAP output survives a fallback from an unavailable engine to the default one.
+
+GitHub issue #1264: :class:`~pcapkit.foundation.extraction.Extractor` replaced
+``format='pcap'`` with ``'json'`` from the *requested* engine name, before
+:meth:`~pcapkit.foundation.extraction.Extractor.run` found the engine missing
+and ran the default :class:`~pcapkit.foundation.engines.pcap.PCAP` engine --
+which writes PCAP -- so ``out.json`` appeared where ``out.pcap`` was asked for.
+The trace format was replaced the same way.
+
+Each engine is made unavailable by patching its own check, so the result does
+not depend on what this host has installed: ``unsupported_reason`` covers the
+"not supported on this interpreter" path, and ``import_test`` the "not
+installed" one.
+
+"""
+
+import contextlib
+import importlib.util
+import os
+import sys
+import tempfile
+import unittest
+import warnings
+from unittest import mock
+
+from tests._support import reimport_once_per_class, sample_path
+
+#: Every engine whose frames cannot be written as PCAP.
+ENGINES = ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile')
+
+
+def _read(path: str) -> bytes:
+    with open(path, 'rb') as file:
+        return file.read()
+
+
+class TestEngineFallbackKeepsPCAPOutput(unittest.TestCase):
+    """The format check follows the engine that runs, not the one requested."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    @staticmethod
+    def _engine_class(name: str):
+        from pcapkit.corekit.module import ModuleDescriptor
+        from pcapkit.foundation.extraction import Extractor
+
+        eng = Extractor.__engine__[name]
+        return eng.klass if isinstance(eng, ModuleDescriptor) else eng
+
+    def _unavailable(self, name: str, how: str):
+        """Context manager making engine ``name`` unavailable ``how``."""
+        klass = self._engine_class(name)
+        if how == 'unsupported':
+            return mock.patch.object(klass, 'unsupported_reason', return_value='unit test')
+        # A ``None`` entry in :data:`sys.modules` makes the import raise
+        # :exc:`ImportError`, so the real ``import_test`` runs and warns.
+        stack = contextlib.ExitStack()
+        stack.enter_context(mock.patch.object(klass, 'unsupported_reason', return_value=None))
+        stack.enter_context(mock.patch.dict(sys.modules, {klass.module: None}))
+        return stack
+
+    def _extract(self, **kwargs):
+        from pcapkit.interface import extract
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            extractor = extract(store=False, **kwargs)
+        return extractor, [w.message for w in caught]
+
+    def test_pcap_output_is_kept_and_byte_exact(self) -> None:
+        from pcapkit.foundation.engines.pcap import PCAP
+        from pcapkit.utilities.warnings import EngineWarning, FormatWarning
+
+        fin = sample_path('in.pcap')
+        for name in ENGINES:
+            for how in ('unsupported', 'not-installed'):
+                for fmt in ('pcap', 'cap'):
+                    with self.subTest(engine=name, how=how, format=fmt), \
+                            tempfile.TemporaryDirectory() as tmp, self._unavailable(name, how):
+                        extractor, caught = self._extract(fin=fin, fout=os.path.join(tmp, 'out'),
+                                                          format=fmt, engine=name)
+                        self.assertIsInstance(extractor.engine, PCAP)
+                        self.assertEqual(extractor.format, 'pcap')
+                        self.assertEqual(os.listdir(tmp), ['out.pcap'])
+                        self.assertEqual(_read(os.path.join(tmp, 'out.pcap')), _read(fin))
+                        self.assertEqual([m for m in caught if isinstance(m, FormatWarning)], [])
+                        # Settling the engine early must not report it twice.
+                        engine_warnings = [str(m) for m in caught if isinstance(m, EngineWarning)]
+                        self.assertEqual(len(engine_warnings), 1, engine_warnings)
+                        self.assertIn('using default engine instead', engine_warnings[0])
+
+    def test_unknown_engine_keeps_pcap_output(self) -> None:
+        from pcapkit.utilities.warnings import EngineWarning
+
+        fin = sample_path('in.pcap')
+        with tempfile.TemporaryDirectory() as tmp:
+            extractor, caught = self._extract(fin=fin, fout=os.path.join(tmp, 'out'),
+                                              format='pcap', engine='no-such-engine')
+            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'])
+
+    def test_trace_format_is_kept(self) -> None:
+        from pcapkit.utilities.warnings import FormatWarning
+
+        fin = sample_path('in.pcap')
+        for name in ('dpkt', 'scapy', 'pyshark', 'pypcapfile'):
+            with self.subTest(engine=name), tempfile.TemporaryDirectory() as tmp, \
+                    self._unavailable(name, 'unsupported'):
+                extractor, caught = self._extract(fin=fin, fout=os.path.join(tmp, 'out'),
+                                                  format='json', engine=name, trace=True, tcp=True,
+                                                  trace_fout=os.path.join(tmp, 'trace'))
+                self.assertEqual([m for m in caught if isinstance(m, FormatWarning)], [])
+                flows = extractor.trace.tcp
+                self.assertTrue(flows)
+                for flow in flows:
+                    self.assertTrue(flow.fpout.endswith('.pcap'), flow.fpout)
+
+    def test_available_third_party_engine_still_falls_back_to_json(self) -> None:
+        # The other half of the contract: an engine that really runs keeps the
+        # existing replacement, and its warning names that engine.
+        from pcapkit.utilities.warnings import FormatWarning
+
+        if importlib.util.find_spec('dpkt') is None:
+            self.skipTest('dpkt not installed')
+        with tempfile.TemporaryDirectory() as tmp:
+            extractor, caught = self._extract(fin=sample_path('in.pcap'),
+                                              fout=os.path.join(tmp, 'out'), format='pcap', engine='DPKT')
+            self.assertEqual(extractor.format, 'json')
+            self.assertEqual(os.listdir(tmp), ['out.json'])
+        self.assertEqual([str(m) for m in caught if isinstance(m, FormatWarning)],
+                         ["'Extractor(engine=dpkt)' does not support 'format=pcap'; "
+                          "using 'format=\"json\"' instead"])
+
+
+if __name__ == '__main__':
+    unittest.main()