Adding upstream version 4.2.2.upstream/4.2.2

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 599 insertions, 0 deletions

diff --git a/test/suite_capture.py b/test/suite_capture.py
new file mode 100644
index 00000000..ffcd6cd5
--- /dev/null
+++ b/test/suite_capture.py
@@ -0,0 +1,599 @@
+#
+# Wireshark tests
+# By Gerald Combs <gerald@wireshark.org>
+#
+# Ported from a set of Bash scripts which were copyright 2005 Ulf Lamping
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+'''Capture tests'''
+
+import glob
+import hashlib
+import os
+import socket
+import subprocess
+import subprocesstest
+from subprocesstest import cat_dhcp_command, cat_cap_file_command, count_output, grep_output, check_packet_count
+import sys
+import threading
+import time
+import uuid
+import sysconfig
+import pytest
+
+capture_duration = 5
+
+testout_pcap = 'testout.pcap'
+testout_pcapng = 'testout.pcapng'
+snapshot_len = 96
+
+class UdpTrafficGenerator(threading.Thread):
+    def __init__(self):
+        super().__init__(daemon=True)
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.stopped = False
+
+    def run(self):
+        while not self.stopped:
+            time.sleep(.05)
+            self.sock.sendto(b'Wireshark test\n', ('127.0.0.1', 9))
+
+    def stop(self):
+        if not self.stopped:
+            self.stopped = True
+            self.join()
+
+
+@pytest.fixture
+def traffic_generator():
+    '''
+    Traffic generator factory. Invoking it returns a tuple (start_func, cfilter)
+    where cfilter is a capture filter to match the generated traffic.
+    start_func can be invoked to start generating traffic and returns a function
+    which can be used to stop traffic generation early.
+    Currently generates a bunch of UDP traffic to localhost.
+    '''
+    threads = []
+    def start_processes():
+        thread = UdpTrafficGenerator()
+        thread.start()
+        threads.append(thread)
+        return thread.stop
+    try:
+        yield start_processes, 'udp port 9'
+    finally:
+        for thread in threads:
+            thread.stop()
+
+
+@pytest.fixture(scope='session')
+def wireshark_k(wireshark_command):
+    return tuple(list(wireshark_command) + ['-k'])
+
+
+def capture_command(*args, shell=False):
+    cmd_args = list(args)
+    if type(cmd_args[0]) != str:
+        # Assume something like ['wireshark', '-k']
+        cmd_args = list(cmd_args[0]) + list(cmd_args)[1:]
+    if shell:
+        cmd_args = ' '.join(cmd_args)
+    return cmd_args
+
+
+@pytest.fixture
+def check_capture_10_packets(capture_interface, cmd_capinfos, traffic_generator, result_file):
+    start_traffic, cfilter = traffic_generator
+    def check_capture_10_packets_real(self, cmd=None, to_stdout=False, env=None):
+        assert cmd is not None
+        testout_file = result_file(testout_pcap)
+        stop_traffic = start_traffic()
+        if to_stdout:
+            subprocesstest.check_run(capture_command(cmd,
+                '-i', '"{}"'.format(capture_interface),
+                '-p',
+                '-w', '-',
+                '-c', '10',
+                '-a', 'duration:{}'.format(capture_duration),
+                '-f', '"{}"'.format(cfilter),
+                '>', testout_file,
+                shell=True
+            ),
+            shell=True, env=env)
+        else:
+            subprocesstest.check_run(capture_command(cmd,
+                '-i', capture_interface,
+                '-p',
+                '-w', testout_file,
+                '-c', '10',
+                '-a', 'duration:{}'.format(capture_duration),
+                '-f', cfilter,
+            ), env=env)
+        stop_traffic()
+        check_packet_count(cmd_capinfos, 10, testout_file)
+    return check_capture_10_packets_real
+
+
+@pytest.fixture
+def check_capture_fifo(cmd_capinfos, result_file):
+    if sys.platform == 'win32':
+        pytest.skip('Test requires OS fifo support.')
+
+    def check_capture_fifo_real(self, cmd=None, env=None):
+        assert cmd is not None
+        testout_file = result_file(testout_pcap)
+        fifo_file = result_file('testout.fifo')
+        try:
+            # If a previous test left its fifo laying around, e.g. from a failure, remove it.
+            os.unlink(fifo_file)
+        except Exception:
+            pass
+        os.mkfifo(fifo_file)
+        slow_dhcp_cmd = cat_dhcp_command('slow')
+        fifo_proc = subprocess.Popen(
+            ('{0} > {1}'.format(slow_dhcp_cmd, fifo_file)),
+            shell=True)
+        subprocesstest.check_run(capture_command(cmd,
+            '-i', fifo_file,
+            '-p',
+            '-w', testout_file,
+            '-a', 'duration:{}'.format(capture_duration),
+        ), env=env)
+        fifo_proc.kill()
+        assert os.path.isfile(testout_file)
+        check_packet_count(cmd_capinfos, 8, testout_file)
+    return check_capture_fifo_real
+
+
+@pytest.fixture
+def check_capture_stdin(cmd_capinfos, result_file):
+    # Capturing always requires dumpcap, hence the dependency on it.
+    def check_capture_stdin_real(self, cmd=None, env=None):
+        # Similar to suite_io.check_io_4_packets.
+        assert cmd is not None
+        testout_file = result_file(testout_pcap)
+        slow_dhcp_cmd = cat_dhcp_command('slow')
+        capture_cmd = capture_command(cmd,
+            '-i', '-',
+            '-w', testout_file,
+            '-a', 'duration:{}'.format(capture_duration),
+            shell=True
+        )
+        is_gui = type(cmd) != str and '-k' in cmd[0]
+        if is_gui:
+            capture_cmd += ' --log-level=info'
+        if sysconfig.get_platform().startswith('mingw'):
+            pytest.skip('FIXME Pipes are broken with the MSYS2 shell')
+        pipe_proc = subprocesstest.check_run(slow_dhcp_cmd + ' | ' + capture_cmd, shell=True, capture_output=True, env=env)
+        if is_gui:
+            # Wireshark uses stdout and not stderr for diagnostic messages
+            # XXX: Confirm this
+            assert grep_output(pipe_proc.stdout, 'Wireshark is up and ready to go'), 'No startup message.'
+            assert grep_output(pipe_proc.stdout, 'Capture started'), 'No capture start message.'
+            assert grep_output(pipe_proc.stdout, 'Capture stopped'), 'No capture stop message.'
+        assert os.path.isfile(testout_file)
+        check_packet_count(cmd_capinfos, 8, testout_file)
+    return check_capture_stdin_real
+
+
+@pytest.fixture
+def check_capture_read_filter(capture_interface, traffic_generator, cmd_capinfos, result_file):
+    start_traffic, cfilter = traffic_generator
+    def check_capture_read_filter_real(self, cmd=None, env=None):
+        assert cmd is not None
+        testout_file = result_file(testout_pcap)
+        stop_traffic = start_traffic()
+        subprocesstest.check_run(capture_command(cmd,
+            '-i', capture_interface,
+            '-p',
+            '-w', testout_file,
+            '-2',
+            '-R', 'dcerpc.cn_call_id==123456', # Something unlikely.
+            '-c', '10',
+            '-a', 'duration:{}'.format(capture_duration),
+            '-f', cfilter,
+        ), env=env)
+        stop_traffic()
+        check_packet_count(cmd_capinfos, 0, testout_file)
+    return check_capture_read_filter_real
+
+@pytest.fixture
+def check_capture_snapshot_len(capture_interface, cmd_tshark, traffic_generator, cmd_capinfos, result_file):
+    start_traffic, cfilter = traffic_generator
+    def check_capture_snapshot_len_real(self, cmd=None, env=None):
+        assert cmd is not None
+        stop_traffic = start_traffic()
+        testout_file = result_file(testout_pcap)
+        subprocesstest.check_run(capture_command(cmd,
+            '-i', capture_interface,
+            '-p',
+            '-w', testout_file,
+            '-s', str(snapshot_len),
+            '-a', 'duration:{}'.format(capture_duration),
+            '-f', cfilter,
+        ), env=env)
+        stop_traffic()
+        assert os.path.isfile(testout_file)
+
+        # Use tshark to filter out all packets larger than 68 bytes.
+        testout2_file = result_file('testout2.pcap')
+
+        subprocesstest.check_run((cmd_tshark,
+            '-r', testout_file,
+            '-w', testout2_file,
+            '-Y', 'frame.cap_len>{}'.format(snapshot_len),
+        ), env=env)
+        check_packet_count(cmd_capinfos, 0, testout2_file)
+    return check_capture_snapshot_len_real
+
+
+@pytest.fixture
+def check_dumpcap_autostop_stdin(cmd_dumpcap, cmd_capinfos, result_file):
+    def check_dumpcap_autostop_stdin_real(self, packets=None, filesize=None, env=None):
+        # Similar to check_capture_stdin.
+        testout_file = result_file(testout_pcap)
+        cat100_dhcp_cmd = cat_dhcp_command('cat100')
+        condition='oops:invalid'
+
+        if packets is not None:
+            condition = 'packets:{}'.format(packets)
+        elif filesize is not None:
+            condition = 'filesize:{}'.format(filesize)
+        else:
+            raise AssertionError('Need one of packets or filesize')
+
+        cmd_ = '"{}"'.format(cmd_dumpcap)
+        capture_cmd = ' '.join((cmd_,
+            '-i', '-',
+            '-w', testout_file,
+            '-a', condition,
+        ))
+        if sysconfig.get_platform().startswith('mingw'):
+            pytest.skip('FIXME Pipes are broken with the MSYS2 shell')
+        subprocesstest.check_run(cat100_dhcp_cmd + ' | ' + capture_cmd, shell=True, env=env)
+        assert os.path.isfile(testout_file)
+
+        if packets is not None:
+            check_packet_count(cmd_capinfos, packets, testout_file)
+        elif filesize is not None:
+            capturekb = os.path.getsize(testout_file) / 1000
+            assert capturekb >= filesize
+    return check_dumpcap_autostop_stdin_real
+
+
+@pytest.fixture
+def check_dumpcap_ringbuffer_stdin(cmd_dumpcap, cmd_capinfos, result_file):
+    def check_dumpcap_ringbuffer_stdin_real(self, packets=None, filesize=None, env=None):
+        # Similar to check_capture_stdin.
+        rb_unique = 'dhcp_rb_' + uuid.uuid4().hex[:6] # Random ID
+        testout_file = result_file('testout.{}.pcapng'.format(rb_unique))
+        testout_glob = result_file('testout.{}_*.pcapng'.format(rb_unique))
+        cat100_dhcp_cmd = cat_dhcp_command('cat100')
+        condition='oops:invalid'
+
+        if packets is not None:
+            condition = 'packets:{}'.format(packets)
+        elif filesize is not None:
+            condition = 'filesize:{}'.format(filesize)
+        else:
+            raise AssertionError('Need one of packets or filesize')
+
+        cmd_ = '"{}"'.format(cmd_dumpcap)
+        capture_cmd = ' '.join((cmd_,
+            '-i', '-',
+            '-w', testout_file,
+            '-a', 'files:2',
+            '-b', condition,
+        ))
+        if sysconfig.get_platform().startswith('mingw'):
+            pytest.skip('FIXME Pipes are broken with the MSYS2 shell')
+        subprocesstest.check_run(cat100_dhcp_cmd + ' | ' + capture_cmd, shell=True, env=env)
+
+        rb_files = glob.glob(testout_glob)
+        assert len(rb_files) == 2
+
+        for rbf in rb_files:
+            assert os.path.isfile(rbf)
+            if packets is not None:
+                check_packet_count(cmd_capinfos, packets, rbf)
+            elif filesize is not None:
+                capturekb = os.path.getsize(rbf) / 1000
+                assert capturekb >= filesize
+    return check_dumpcap_ringbuffer_stdin_real
+
+
+@pytest.fixture
+def check_dumpcap_pcapng_sections(cmd_dumpcap, cmd_tshark, cmd_capinfos, capture_file, result_file):
+    if sys.platform == 'win32':
+        pytest.skip('Test requires OS fifo support.')
+    def check_dumpcap_pcapng_sections_real(self, multi_input=False, multi_output=False, env=None):
+        # Make sure we always test multiple SHBs in an input.
+        in_files_l = [ [
+            capture_file('many_interfaces.pcapng.1'),
+            capture_file('many_interfaces.pcapng.2')
+            ] ]
+        if multi_input:
+            in_files_l.append([ capture_file('many_interfaces.pcapng.3') ])
+        fifo_files = []
+        fifo_procs = []
+        # Default values for our validity tests
+        check_val_d = {
+            'filename': None,
+            'packet_count': 0,
+            'idb_count': 0,
+            'ua_pt1_count': 0,
+            'ua_pt2_count': 0,
+            'ua_pt3_count': 0,
+            'ua_dc_count': 0,
+        }
+        check_vals = [ check_val_d ]
+
+        for in_files in in_files_l:
+            fifo_file = result_file('dumpcap_pcapng_sections_{}.fifo'.format(len(fifo_files) + 1))
+            fifo_files.append(fifo_file)
+            # If a previous test left its fifo laying around, e.g. from a failure, remove it.
+            try:
+                os.unlink(fifo_file)
+            except Exception: pass
+            os.mkfifo(fifo_file)
+            cat_cmd = cat_cap_file_command(in_files)
+            fifo_procs.append(subprocess.Popen(('{0} > {1}'.format(cat_cmd, fifo_file)), shell=True))
+
+        if multi_output:
+            rb_unique = 'sections_rb_' + uuid.uuid4().hex[:6] # Random ID
+            testout_file = result_file('testout.{}.pcapng'.format(rb_unique))
+            testout_glob = result_file('testout.{}_*.pcapng'.format(rb_unique))
+            check_vals.append(check_val_d.copy())
+            # check_vals[]['filename'] will be filled in below
+        else:
+            testout_file = result_file(testout_pcapng)
+            check_vals[0]['filename'] = testout_file
+
+        # Capture commands
+        if not multi_input and not multi_output:
+            # Passthrough SHBs, single output file
+            capture_cmd_args = (
+                '-i', fifo_files[0],
+                '-w', testout_file
+            )
+            check_vals[0]['packet_count'] = 79
+            check_vals[0]['idb_count'] = 22
+            check_vals[0]['ua_pt1_count'] = 1
+            check_vals[0]['ua_pt2_count'] = 1
+        elif not multi_input and multi_output:
+            # Passthrough SHBs, multiple output files
+            capture_cmd_args = (
+                '-i', fifo_files[0],
+                '-w', testout_file,
+                '-a', 'files:2',
+                '-b', 'packets:53'
+            )
+            check_vals[0]['packet_count'] = 53
+            check_vals[0]['idb_count'] = 11
+            check_vals[0]['ua_pt1_count'] = 1
+            check_vals[1]['packet_count'] = 26
+            check_vals[1]['idb_count'] = 22
+            check_vals[1]['ua_pt1_count'] = 1
+            check_vals[1]['ua_pt2_count'] = 1
+        elif multi_input and not multi_output:
+            # Dumpcap SHBs, single output file
+            capture_cmd_args = (
+                '-i', fifo_files[0],
+                '-i', fifo_files[1],
+                '-w', testout_file
+            )
+            check_vals[0]['packet_count'] = 88
+            check_vals[0]['idb_count'] = 33
+            check_vals[0]['ua_dc_count'] = 1
+        else:
+            # Dumpcap SHBs, multiple output files
+            capture_cmd_args = (
+                '-i', fifo_files[0],
+                '-i', fifo_files[1],
+                '-w', testout_file,
+                '-a', 'files:2',
+                '-b', 'packets:53'
+            )
+            check_vals[0]['packet_count'] = 53
+            check_vals[0]['idb_count'] = 11
+            check_vals[0]['ua_dc_count'] = 1
+            check_vals[1]['packet_count'] = 35
+            check_vals[1]['idb_count'] = 33
+            check_vals[1]['ua_dc_count'] = 1
+
+        capture_cmd = capture_command(cmd_dumpcap, *capture_cmd_args)
+
+        subprocesstest.check_run(capture_cmd, env=env)
+        for fifo_proc in fifo_procs: fifo_proc.kill()
+
+        rb_files = []
+        if multi_output:
+            rb_files = sorted(glob.glob(testout_glob))
+            assert len(rb_files) == 2
+            check_vals[0]['filename'] = rb_files[0]
+            check_vals[1]['filename'] = rb_files[1]
+
+        for rbf in rb_files:
+            assert os.path.isfile(rbf)
+
+        # Output tests
+
+        if not multi_input and not multi_output:
+            # Check strict bit-for-bit passthrough.
+            in_hash = hashlib.sha256()
+            out_hash = hashlib.sha256()
+            for in_file in in_files_l[0]:
+                in_cap_file = capture_file(in_file)
+                with open(in_cap_file, 'rb') as f:
+                    in_hash.update(f.read())
+            with open(testout_file, 'rb') as f:
+                out_hash.update(f.read())
+            assert in_hash.hexdigest() == out_hash.hexdigest()
+
+        # many_interfaces.pcapng.1 : 64 packets written by "Passthrough test #1"
+        # many_interfaces.pcapng.2 : 15 packets written by "Passthrough test #2"
+        # many_interfaces.pcapng.3 : 9 packets written by "Passthrough test #3"
+        # Each has 11 interfaces.
+        idb_compare_eq = True
+        if multi_input and multi_output:
+            # Having multiple inputs forces the use of threads. In our
+            # case this means that non-packet block counts in the first
+            # file in is nondeterministic.
+            idb_compare_eq = False
+        for check_val in check_vals:
+            check_packet_count(cmd_capinfos, check_val['packet_count'], check_val['filename'])
+
+            tshark_proc = subprocesstest.check_run(capture_command(cmd_tshark,
+                '-r', check_val['filename'],
+                '-V',
+                '-X', 'read_format:MIME Files Format'
+            ), capture_output=True, env=env)
+            # XXX Are there any other sanity checks we should run?
+            if idb_compare_eq:
+                assert count_output(tshark_proc.stdout, r'Block \d+: Interface Description Block \d+') \
+                        == check_val['idb_count']
+            else:
+                assert count_output(tshark_proc.stdout, r'Block \d+: Interface Description Block \d+') \
+                        >= check_val['idb_count']
+                idb_compare_eq = True
+            assert count_output(tshark_proc.stdout, r'Option: User Application = Passthrough test #1') \
+                        == check_val['ua_pt1_count']
+            assert count_output(tshark_proc.stdout, r'Option: User Application = Passthrough test #2') \
+                        == check_val['ua_pt2_count']
+            assert count_output(tshark_proc.stdout, r'Option: User Application = Passthrough test #3') \
+                        == check_val['ua_pt3_count']
+            assert count_output(tshark_proc.stdout, r'Option: User Application = Dumpcap \(Wireshark\)') \
+                        == check_val['ua_dc_count']
+    return check_dumpcap_pcapng_sections_real
+
+
+class TestWiresharkCapture:
+    def test_wireshark_capture_10_packets_to_file(self, request, wireshark_k, check_capture_10_packets, make_screenshot_on_error, test_env):
+        '''Capture 10 packets from the network to a file using Wireshark'''
+        disabled = request.config.getoption('--disable-gui', default=False)
+        if disabled:
+            pytest.skip('GUI tests are disabled via --disable-gui')
+        with make_screenshot_on_error():
+            check_capture_10_packets(self, cmd=wireshark_k, env=test_env)
+
+    # Wireshark doesn't currently support writing to stdout while capturing.
+    # def test_wireshark_capture_10_packets_to_stdout(self, wireshark_k, check_capture_10_packets):
+    #     '''Capture 10 packets from the network to stdout using Wireshark'''
+    #     check_capture_10_packets(self, cmd=wireshark_k, to_stdout=True)
+
+    def test_wireshark_capture_from_fifo(self, request, wireshark_k, check_capture_fifo, make_screenshot_on_error, test_env):
+        '''Capture from a fifo using Wireshark'''
+        disabled = request.config.getoption('--disable-gui', default=False)
+        if disabled:
+            pytest.skip('GUI tests are disabled via --disable-gui')
+        with make_screenshot_on_error():
+            check_capture_fifo(self, cmd=wireshark_k, env=test_env)
+
+    def test_wireshark_capture_from_stdin(self, request, wireshark_k, check_capture_stdin, make_screenshot_on_error, test_env):
+        '''Capture from stdin using Wireshark'''
+        disabled = request.config.getoption('--disable-gui', default=False)
+        if disabled:
+            pytest.skip('GUI tests are disabled via --disable-gui')
+        with make_screenshot_on_error():
+            check_capture_stdin(self, cmd=wireshark_k, env=test_env)
+
+    def test_wireshark_capture_snapshot_len(self, request, wireshark_k, check_capture_snapshot_len, make_screenshot_on_error, test_env):
+        '''Capture truncated packets using Wireshark'''
+        disabled = request.config.getoption('--disable-gui', default=False)
+        if disabled:
+            pytest.skip('GUI tests are disabled via --disable-gui')
+        with make_screenshot_on_error():
+            check_capture_snapshot_len(self, cmd=wireshark_k, env=test_env)
+
+
+class TestTsharkCapture:
+    def test_tshark_capture_10_packets_to_file(self, cmd_tshark, check_capture_10_packets, test_env):
+        '''Capture 10 packets from the network to a file using TShark'''
+        check_capture_10_packets(self, cmd=cmd_tshark, env=test_env)
+
+    def test_tshark_capture_10_packets_to_stdout(self, cmd_tshark, check_capture_10_packets, test_env):
+        '''Capture 10 packets from the network to stdout using TShark'''
+        check_capture_10_packets(self, cmd=cmd_tshark, to_stdout=True, env=test_env)
+
+    def test_tshark_capture_from_fifo(self, cmd_tshark, check_capture_fifo, test_env):
+        '''Capture from a fifo using TShark'''
+        check_capture_fifo(self, cmd=cmd_tshark, env=test_env)
+
+    def test_tshark_capture_from_stdin(self, cmd_tshark, check_capture_stdin, test_env):
+        '''Capture from stdin using TShark'''
+        check_capture_stdin(self, cmd=cmd_tshark, env=test_env)
+
+    def test_tshark_capture_snapshot_len(self, cmd_tshark, check_capture_snapshot_len, test_env):
+        '''Capture truncated packets using TShark'''
+        check_capture_snapshot_len(self, cmd=cmd_tshark, env=test_env)
+
+
+class TestDumpcapCapture:
+    def test_dumpcap_capture_10_packets_to_file(self, cmd_dumpcap, check_capture_10_packets, base_env):
+        '''Capture 10 packets from the network to a file using Dumpcap'''
+        check_capture_10_packets(self, cmd=cmd_dumpcap, env=base_env)
+
+    def test_dumpcap_capture_10_packets_to_stdout(self, cmd_dumpcap, check_capture_10_packets, base_env):
+        '''Capture 10 packets from the network to stdout using Dumpcap'''
+        check_capture_10_packets(self, cmd=cmd_dumpcap, to_stdout=True, env=base_env)
+
+    def test_dumpcap_capture_from_fifo(self, cmd_dumpcap, check_capture_fifo, base_env):
+        '''Capture from a fifo using Dumpcap'''
+        check_capture_fifo(self, cmd=cmd_dumpcap, env=base_env)
+
+    def test_dumpcap_capture_from_stdin(self, cmd_dumpcap, check_capture_stdin, base_env):
+        '''Capture from stdin using Dumpcap'''
+        check_capture_stdin(self, cmd=cmd_dumpcap, env=base_env)
+
+    def test_dumpcap_capture_snapshot_len(self, check_capture_snapshot_len, cmd_dumpcap, base_env):
+        '''Capture truncated packets using Dumpcap'''
+        check_capture_snapshot_len(self, cmd=cmd_dumpcap, env=base_env)
+
+
+class TestDumpcapAutostop:
+    # duration, filesize, packets, files
+    def test_dumpcap_autostop_filesize(self, check_dumpcap_autostop_stdin, base_env):
+        '''Capture from stdin using Dumpcap until we reach a file size limit'''
+        check_dumpcap_autostop_stdin(self, filesize=15, env=base_env)
+
+    def test_dumpcap_autostop_packets(self, check_dumpcap_autostop_stdin, base_env):
+        '''Capture from stdin using Dumpcap until we reach a packet limit'''
+        check_dumpcap_autostop_stdin(self, packets=97, env=base_env) # Last prime before 100. Arbitrary.
+
+
+class TestDumpcapRingbuffer:
+    # duration, interval, filesize, packets, files
+    def test_dumpcap_ringbuffer_filesize(self, check_dumpcap_ringbuffer_stdin, base_env):
+        '''Capture from stdin using Dumpcap and write multiple files until we reach a file size limit'''
+        check_dumpcap_ringbuffer_stdin(self, filesize=15, env=base_env)
+
+    def test_dumpcap_ringbuffer_packets(self, check_dumpcap_ringbuffer_stdin, base_env):
+        '''Capture from stdin using Dumpcap and write multiple files until we reach a packet limit'''
+        check_dumpcap_ringbuffer_stdin(self, packets=47, env=base_env) # Last prime before 50. Arbitrary.
+
+
+class TestDumpcapPcapngSections:
+    def test_dumpcap_pcapng_single_in_single_out(self, check_dumpcap_pcapng_sections, base_env):
+        '''Capture from a single pcapng source using Dumpcap and write a single file'''
+        if sys.byteorder == 'big':
+            pytest.skip('this test is supported on little endian only')
+        check_dumpcap_pcapng_sections(self, env=base_env)
+
+    def test_dumpcap_pcapng_single_in_multi_out(self, check_dumpcap_pcapng_sections, base_env):
+        '''Capture from a single pcapng source using Dumpcap and write two files'''
+        if sys.byteorder == 'big':
+            pytest.skip('this test is supported on little endian only')
+        check_dumpcap_pcapng_sections(self, multi_output=True, env=base_env)
+
+    def test_dumpcap_pcapng_multi_in_single_out(self, check_dumpcap_pcapng_sections, base_env):
+        '''Capture from two pcapng sources using Dumpcap and write a single file'''
+        if sys.byteorder == 'big':
+            pytest.skip('this test is supported on little endian only')
+        check_dumpcap_pcapng_sections(self, multi_input=True, env=base_env)
+
+    def test_dumpcap_pcapng_multi_in_multi_out(self, check_dumpcap_pcapng_sections, base_env):
+        '''Capture from two pcapng sources using Dumpcap and write two files'''
+        if sys.byteorder == 'big':
+            pytest.skip('this test is supported on little endian only')
+        check_dumpcap_pcapng_sections(self, multi_input=True, multi_output=True, env=base_env)