test(integration): skip upstream-pinned PCAP-NG counts on a stand-in fixture (#1478)

tests/_support.py

@@ -3,6 +3,8 @@
 import abc
 import collections.abc
 import contextlib
+import functools
+import hashlib
 import importlib.util
 import inspect
 import math
@@ -230,6 +232,78 @@ def sample_path(name: str) -> str:
     return str(path)
 
 
+#: The generator that downloads the upstream PCAP-NG captures, and pins the
+#: SHA-256 of the exact bytes each download must have.
+PCAPNG_GENERATOR = ROOT / 'examples' / 'generators' / 'pcapng.py'
+
+
+@functools.lru_cache(maxsize=None)
+def upstream_digests() -> 'dict[str, str]':
+    """The pinned SHA-256 of every capture :file:`pcapng.py` downloads, by name.
+
+    Read out of the generator's own ``FIXTURES`` table, so the digest lives in
+    one place. The module is executed by path under a private name -- the
+    directory is not a package -- and imports nothing of :mod:`pcapkit` at the
+    top level, so this is cheap and safe from the unit tier.
+
+    """
+    spec = importlib.util.spec_from_file_location('_pcapng_generator', PCAPNG_GENERATOR)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f'Unable to load {PCAPNG_GENERATOR}')
+    module = importlib.util.module_from_spec(spec)
+    saved = list(sys.path)
+    try:
+        spec.loader.exec_module(module)
+    finally:
+        # The generator puts the tree root on sys.path for its own sake (#1343).
+        sys.path[:] = saved
+    return {fixture.name: fixture.sha256 for fixture in module.FIXTURES
+            if fixture.sha256 is not None}
+
+
+def stand_in_reason(path: str) -> 'Optional[str]':
+    """Say why ``path`` is not the upstream capture, or :data:`None` if it is.
+
+    When a download fails, :file:`examples/generators/pcapng.py` writes a
+    synthesised stand-in under the upstream name rather than failing (#1478).
+    The stand-in parses, but its block inventory and options differ, so an
+    assertion pinned to the upstream bytes has to skip on it rather than fail.
+    The pinned digest already tells the two apart, so no marker file is needed
+    and none can fall out of step with the bytes on disk.
+
+    Args:
+        path: A capture, as returned by :func:`sample_path`.
+
+    Returns:
+        A skip reason, or :data:`None` if ``path`` holds the pinned upstream bytes
+        or is not a downloaded capture at all.
+
+    """
+    name = pathlib.Path(path).name
+    expected = upstream_digests().get(name)
+    if expected is None:
+        return None
+    digest = hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest()
+    if digest == expected:
+        return None
+    return (f'{name} is a synthesised stand-in, not the upstream capture this test is '
+            f'pinned to (SHA-256 {digest[:12]}..., expected {expected[:12]}...); its '
+            f'download most likely failed -- re-run {REGENERATE_SAMPLES_CMD!r} with '
+            f'network access')
+
+
+def skip_if_stand_in(test: 'unittest.TestCase', path: str) -> None:
+    """Skip ``test`` -- or its current subtest -- if ``path`` is a stand-in.
+
+    See :func:`stand_in_reason`. Call it only ahead of assertions pinned to the
+    upstream bytes; whatever a stand-in also satisfies should stay unguarded.
+
+    """
+    reason = stand_in_reason(path)
+    if reason is not None:
+        test.skipTest(reason)
+
+
 def ensure_package(name: str, path: pathlib.Path) -> types.ModuleType:
     """Make ``name`` importable as a package, with a bare stub if it is not.
 

tests/integration/test_pcapng_end_to_end.py

@@ -12,12 +12,19 @@
 :file:`examples/generators/pcapng.py`, which also records the parser defects
 they provoke -- those show up as log noise here and are not assertions.
 
+The ``many_interfaces.pcapng`` counts are those of the upstream download. When
+the download fails the generator writes a 24-frame stand-in instead, and
+:func:`~tests._support.skip_if_stand_in` turns each assertion pinned to upstream
+into a skip rather than a failure (#1478). The ``dhcp_big_endian.pcapng``
+stand-in is built to upstream's four-frame inventory, so it is asserted either
+way.
+
 """
 from __future__ import annotations
 
 import unittest
 
-from tests._support import sample_path
+from tests._support import sample_path, skip_if_stand_in
 from tests.integration._helpers import (HAS_RUNTIME, EndToEndTestCase, read_json, report_stems,
                                         section_counts)
 
@@ -31,6 +38,10 @@
     'profile.pcapng': 40,
 }
 
+#: Fixtures whose counts above hold for the upstream download only, not for the
+#: stand-in the generator synthesises when that download fails (#1478).
+UPSTREAM_ONLY_COUNTS = frozenset({'many_interfaces.pcapng'})
+
 #: Block inventory of each fixture whose report can be read back. The
 #: ``test.pcapng`` report cannot; see :class:`PcapngUnescapedKeyTests`.
 PCAPNG_SECTIONS = {
@@ -63,7 +74,10 @@ class PcapngTreeReportTests(EndToEndTestCase):
     def test_every_fixture_dumps_a_report_with_its_expected_frame_count(self) -> None:
         for capture, frames in PCAPNG_FRAMES.items():
             with self.subTest(capture=capture):
-                extractor = self.extract(fin=sample_path(capture), fout=self.out(capture),
+                path = sample_path(capture)
+                if capture in UPSTREAM_ONLY_COUNTS:
+                    skip_if_stand_in(self, path)
+                extractor = self.extract(fin=path, fout=self.out(capture),
                                          format='tree', store=False)
 
                 report = self.tmp_path / f'{capture}.txt'
@@ -81,16 +95,20 @@ class PcapngJsonReportTests(EndToEndTestCase):
     def test_reports_hold_the_expected_blocks(self) -> None:
         for capture, sections in PCAPNG_SECTIONS.items():
             with self.subTest(capture=capture):
-                extractor = self.extract(fin=sample_path(capture), fout=self.out(capture),
+                path = sample_path(capture)
+                if capture in UPSTREAM_ONLY_COUNTS:
+                    skip_if_stand_in(self, path)
+                extractor = self.extract(fin=path, fout=self.out(capture),
                                          format='json', store=False)
                 report = read_json(extractor.output)
 
                 self.assertEqual(section_counts(report), sections)
                 self.assertEqual(extractor.length, PCAPNG_FRAMES[capture])
 
     def test_interface_descriptions_are_reported_one_per_interface(self) -> None:
-        extractor = self.extract(fin=sample_path('many_interfaces.pcapng'),
-                                 fout=self.out('many'), format='json', store=False)
+        path = sample_path('many_interfaces.pcapng')
+        skip_if_stand_in(self, path)
+        extractor = self.extract(fin=path, fout=self.out('many'), format='json', store=False)
         report = read_json(extractor.output)
 
         self.assertEqual(extractor.length, 64)

tests/protocols/misc/test_pcapng_unit.py

@@ -20,7 +20,7 @@
 import warnings
 from unittest import mock
 
-from tests._support import ROOT, reimport_once_per_class, sample_path
+from tests._support import ROOT, reimport_once_per_class, sample_path, skip_if_stand_in
 
 RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper')
 HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS)
@@ -3339,6 +3339,9 @@ def test_pcapng_big_endian_sample_section_options_are_intact(self) -> None:
         section = extractor.engine._ctx_list[0].section
 
         self.assertEqual(section.byteorder, 'big')
+        # The option layout below is upstream's; the stand-in written when the
+        # download fails is big-endian too, but carries its own options (#1478).
+        skip_if_stand_in(self, path)
         self.assertEqual([option.length for _, option in section.options.items(multi=True)],
                          [9, 12, 15, 7, 0])
         self.assertEqual(section.options[OptionType.opt_comment].comment, 'test001')

tests/test_support_helpers.py

@@ -53,21 +53,28 @@
 """
 from __future__ import annotations
 
+import hashlib
 import importlib
+import importlib.util
 import os
 import pathlib
+import shutil
 import signal
 import sys
+import tempfile
 import threading
 import time
 import types
 import unittest
+import urllib.error
 from unittest import mock
 
-from tests._support import (MAX_TIMEOUT, TIMEOUT_SCALE_ENV, bootstrap_core_modules, close_extractor,
-                            ensure_package, install_fake_protocol_module, isolate_modules,
-                            purge_modules, restore_modules, restore_modules_after, scale_timeout,
-                            snapshot_modules, time_limit)
+from tests._support import (MAX_TIMEOUT, PCAPNG_GENERATOR, TIMEOUT_SCALE_ENV,
+                            bootstrap_core_modules, close_extractor, ensure_package,
+                            install_fake_protocol_module, isolate_modules, purge_modules,
+                            restore_modules, restore_modules_after, scale_timeout,
+                            skip_if_stand_in, snapshot_modules, stand_in_reason, time_limit,
+                            upstream_digests)
 
 
 class Closeable:
@@ -862,5 +869,76 @@ def _method_names(case_class: type) -> 'list[str]':
             if name.startswith('test') and callable(getattr(case_class, name))]
 
 
+class StandInReasonTests(unittest.TestCase):
+    """:func:`~tests._support.stand_in_reason` tells upstream bytes from a stand-in.
+
+    Issue #1478: when the download of ``many_interfaces.pcapng`` failed, the
+    generator wrote a 24-frame stand-in and three integration assertions pinned
+    to upstream's 64 frames went red. The digest the generator already pins is
+    what tells the two apart, so that is what is tested here -- with the real
+    generator and a simulated failed download, not only with a fake table.
+
+    """
+
+    def setUp(self) -> None:
+        self.tmp = pathlib.Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+
+    def write(self, name: str, data: bytes) -> str:
+        path = self.tmp / name
+        path.write_bytes(data)
+        return str(path)
+
+    def test_digest_table_covers_both_downloads_and_leaves_sys_path_alone(self) -> None:
+        upstream_digests.cache_clear()
+        before = list(sys.path)
+        digests = upstream_digests()
+        self.assertEqual(sys.path, before)
+        self.assertEqual(sorted(digests), ['dhcp_big_endian.pcapng', 'many_interfaces.pcapng'])
+        for digest in digests.values():
+            self.assertRegex(digest, r'^[0-9a-f]{64}$')
+
+    def test_pinned_bytes_are_upstream(self) -> None:
+        path = self.write('many_interfaces.pcapng', b'upstream')
+        table = {'many_interfaces.pcapng': hashlib.sha256(b'upstream').hexdigest()}
+        with mock.patch('tests._support.upstream_digests', return_value=table):
+            self.assertIsNone(stand_in_reason(path))
+            skip_if_stand_in(self, path)  # does not skip
+
+    def test_other_bytes_are_a_stand_in(self) -> None:
+        path = self.write('many_interfaces.pcapng', b'synthesised')
+        table = {'many_interfaces.pcapng': hashlib.sha256(b'upstream').hexdigest()}
+        with mock.patch('tests._support.upstream_digests', return_value=table):
+            reason = stand_in_reason(path)
+            self.assertIsNotNone(reason)
+            self.assertIn('many_interfaces.pcapng is a synthesised stand-in', reason)
+            with self.assertRaises(unittest.SkipTest):
+                skip_if_stand_in(self, path)
+
+    def test_a_capture_that_is_never_downloaded_is_never_a_stand_in(self) -> None:
+        path = self.write('profile.pcapng', b'anything')
+        self.assertIsNone(stand_in_reason(path))
+
+    def test_failed_download_writes_a_capture_reported_as_a_stand_in(self) -> None:
+        spec = importlib.util.spec_from_file_location('_pcapng_generator_1478', PCAPNG_GENERATOR)
+        assert spec is not None and spec.loader is not None
+        generator = importlib.util.module_from_spec(spec)
+        before = list(sys.path)
+        self.addCleanup(sys.path.__setitem__, slice(None), before)
+        spec.loader.exec_module(generator)
+
+        fixtures = {fixture.name: fixture for fixture in generator.FIXTURES}
+        unreachable = urllib.error.URLError('simulated outage (#1478)')
+        for name in upstream_digests():
+            with self.subTest(name=name):
+                path = self.tmp / name
+                with mock.patch.object(generator, '_download', side_effect=unreachable), \
+                        mock.patch('builtins.print'):
+                    data, origin = generator._materialise(fixtures[name], path)
+                self.assertEqual(origin, 'synthesised (download unavailable)')
+                path.write_bytes(data)
+                self.assertIsNotNone(stand_in_reason(str(path)))
+
+
 if __name__ == '__main__':
     unittest.main()