From 2167d38c5707f0c15c5de302c9e41ef6fb3ce02a Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Fri, 9 Oct 2026 09:55:48 -0400 Subject: [PATCH] test(integration): skip upstream-pinned PCAP-NG counts on a stand-in fixture (#1478) When the download of many_interfaces.pcapng or dhcp_big_endian.pcapng fails, examples/generators/pcapng.py writes a synthesised stand-in under the same name, and assertions pinned to the upstream bytes went red on a network blip. - tests/_support.py: stand_in_reason() compares a capture's SHA-256 against the digest the generator already pins in FIXTURES; skip_if_stand_in() skips the test or subtest when they differ. - test_pcapng_end_to_end.py: guard the many_interfaces.pcapng counts; the dhcp_big_endian stand-in matches upstream's inventory and stays asserted. - test_pcapng_unit.py: guard the big-endian SHB option layout check. - test_support_helpers.py: cover the helpers, including a simulated failed download through the real generator. --- tests/_support.py | 74 ++++++++++++++++++ tests/integration/test_pcapng_end_to_end.py | 28 +++++-- tests/protocols/misc/test_pcapng_unit.py | 5 +- tests/test_support_helpers.py | 86 ++++++++++++++++++++- 4 files changed, 183 insertions(+), 10 deletions(-) diff --git a/tests/_support.py b/tests/_support.py index 348473647b..65266432f0 100644 --- a/tests/_support.py +++ b/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. diff --git a/tests/integration/test_pcapng_end_to_end.py b/tests/integration/test_pcapng_end_to_end.py index 49ce475f85..336cb7b607 100644 --- a/tests/integration/test_pcapng_end_to_end.py +++ b/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,7 +95,10 @@ 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) @@ -89,8 +106,9 @@ def test_reports_hold_the_expected_blocks(self) -> None: 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) diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index cc0d87a0a8..bfead3593c 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/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') diff --git a/tests/test_support_helpers.py b/tests/test_support_helpers.py index 5b0442bc2f..59331e7ac9 100644 --- a/tests/test_support_helpers.py +++ b/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()