test/suite_capture.py

   1 #
   2 # Wireshark tests
   3 # By Gerald Combs <gerald@wireshark.org>
   4 #
   5 # Ported from a set of Bash scripts which were copyright 2005 Ulf Lamping
   6 #
   7 # SPDX-License-Identifier: GPL-2.0-or-later
   8 #
   9 '''Capture tests'''
  10
  11 import glob
  12 import hashlib
  13 import os
  14 import socket
  15 import subprocess
  16 import subprocesstest
  17 from subprocesstest import cat_dhcp_command, cat_cap_file_command, count_output, grep_output, check_packet_count
  18 import sys
  19 import threading
  20 import time
  21 import uuid
  22 import sysconfig
  23 import pytest
  24
  25 capture_duration = 5
  26
  27 testout_pcap = 'testout.pcap'
  28 testout_pcapng = 'testout.pcapng'
  29 snapshot_len = 96
  30
  31 class UdpTrafficGenerator(threading.Thread):
  32     def __init__(self):
  33         super().__init__(daemon=True)
  34         self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  35         self.stopped = False
  36
  37     def run(self):
  38         while not self.stopped:
  39             time.sleep(.05)
  40             self.sock.sendto(b'Wireshark test\n', ('127.0.0.1', 9))
  41
  42     def stop(self):
  43         if not self.stopped:
  44             self.stopped = True
  45             self.join()
  46
  47
  48 @pytest.fixture
  49 def traffic_generator():
  50     '''
  51     Traffic generator factory. Invoking it returns a tuple (start_func, cfilter)
  52     where cfilter is a capture filter to match the generated traffic.
  53     start_func can be invoked to start generating traffic and returns a function
  54     which can be used to stop traffic generation early.
  55     Currently generates a bunch of UDP traffic to localhost.
  56     '''
  57     threads = []
  58     def start_processes():
  59         thread = UdpTrafficGenerator()
  60         thread.start()
  61         threads.append(thread)
  62         return thread.stop
  63     try:
  64         yield start_processes, 'udp port 9'
  65     finally:
  66         for thread in threads:
  67             thread.stop()
  68
  69
  70 @pytest.fixture(scope='session')
  71 def wireshark_k(wireshark_command):
  72     return tuple(list(wireshark_command) + ['-k'])
  73
  74
  75 def capture_command(*args, shell=False, quoted=False):
  76     cmd_args = list(args)
  77     if type(cmd_args[0]) != str:
  78         # Assume something like ['wireshark', '-k']
  79         cmd_args = list(cmd_args[0]) + list(cmd_args)[1:]
  80     if shell:
  81         cmd_args[0] = f'"{cmd_args[0]}"'
  82         cmd_args = ' '.join(cmd_args)
  83     return cmd_args
  84
  85
  86 @pytest.fixture
  87 def check_capture_10_packets(capture_interface, cmd_capinfos, traffic_generator, result_file):
  88     start_traffic, cfilter = traffic_generator
  89     def check_capture_10_packets_real(self, cmd=None, to_stdout=False, env=None):
  90         assert cmd is not None
  91         testout_file = result_file(testout_pcap)
  92         stop_traffic = start_traffic()
  93         if to_stdout:
  94             subprocesstest.check_run(capture_command(cmd,
  95                 '-i', '"{}"'.format(capture_interface),
  96                 '-p',
  97                 '-w', '-',
  98                 '-c', '10',
  99                 '-a', 'duration:{}'.format(capture_duration),
 100                 '-f', '"{}"'.format(cfilter),
 101                 '>', testout_file,
 102                 shell=True
 103             ),
 104             shell=True, env=env)
 105         else:
 106             subprocesstest.check_run(capture_command(cmd,
 107                 '-i', capture_interface,
 108                 '-p',
 109                 '-w', testout_file,
 110                 '-c', '10',
 111                 '-a', 'duration:{}'.format(capture_duration),
 112                 '-f', cfilter,
 113             ), env=env)
 114         stop_traffic()
 115         check_packet_count(cmd_capinfos, 10, testout_file)
 116     return check_capture_10_packets_real
 117
 118
 119 @pytest.fixture
 120 def check_capture_fifo(cmd_capinfos, result_file):
 121     if sys.platform == 'win32':
 122         pytest.skip('Test requires OS fifo support.')
 123
 124     def check_capture_fifo_real(self, cmd=None, env=None):
 125         assert cmd is not None
 126         testout_file = result_file(testout_pcap)
 127         fifo_file = result_file('testout.fifo')
 128         try:
 129             # If a previous test left its fifo laying around, e.g. from a failure, remove it.
 130             os.unlink(fifo_file)
 131         except Exception:
 132             pass
 133         os.mkfifo(fifo_file)
 134         slow_dhcp_cmd = cat_dhcp_command('slow')
 135         fifo_proc = subprocess.Popen(
 136             ('{0} > {1}'.format(slow_dhcp_cmd, fifo_file)),
 137             shell=True)
 138         subprocesstest.check_run(capture_command(cmd,
 139             '-i', fifo_file,
 140             '-p',
 141             '-w', testout_file,
 142             '-a', 'duration:{}'.format(capture_duration),
 143         ), env=env)
 144         fifo_proc.kill()
 145         assert os.path.isfile(testout_file)
 146         check_packet_count(cmd_capinfos, 8, testout_file)
 147     return check_capture_fifo_real
 148
 149
 150 @pytest.fixture
 151 def check_capture_stdin(cmd_capinfos, result_file):
 152     # Capturing always requires dumpcap, hence the dependency on it.
 153     def check_capture_stdin_real(self, cmd=None, env=None):
 154         # Similar to suite_io.check_io_4_packets.
 155         assert cmd is not None
 156         testout_file = result_file(testout_pcap)
 157         slow_dhcp_cmd = cat_dhcp_command('slow')
 158         capture_cmd = capture_command(cmd,
 159             '-i', '-',
 160             '-w', f'"{testout_file}"',
 161             '-a', 'duration:{}'.format(capture_duration),
 162             shell=True
 163         )
 164         is_gui = type(cmd) != str and '-k' in cmd[0]
 165         if is_gui:
 166             capture_cmd += ' --log-level=info'
 167         if sysconfig.get_platform().startswith('mingw'):
 168             pytest.skip('FIXME Pipes are broken with the MSYS2 shell')
 169         pipe_proc = subprocesstest.check_run(slow_dhcp_cmd + ' | ' + capture_cmd, shell=True, capture_output=True, env=env)
 170         if is_gui:
 171             # Wireshark uses stdout and not stderr for diagnostic messages
 172             # XXX: Confirm this
 173             assert grep_output(pipe_proc.stdout, 'Wireshark is up and ready to go'), 'No startup message.'
 174             assert grep_output(pipe_proc.stdout, 'Capture started'), 'No capture start message.'
 175             assert grep_output(pipe_proc.stdout, 'Capture stopped'), 'No capture stop message.'
 176         assert os.path.isfile(testout_file)
 177         check_packet_count(cmd_capinfos, 8, testout_file)
 178     return check_capture_stdin_real
 179
 180
 181 @pytest.fixture
 182 def check_capture_read_filter(capture_interface, traffic_generator, cmd_capinfos, result_file):
 183     start_traffic, cfilter = traffic_generator
 184     def check_capture_read_filter_real(self, cmd=None, env=None):
 185         assert cmd is not None
 186         testout_file = result_file(testout_pcap)
 187         stop_traffic = start_traffic()
 188         subprocesstest.check_run(capture_command(cmd,
 189             '-i', capture_interface,
 190             '-p',
 191             '-w', testout_file,
 192             '-2',
 193             '-R', 'dcerpc.cn_call_id==123456', # Something unlikely.
 194             '-c', '10',
 195             '-a', 'duration:{}'.format(capture_duration),
 196             '-f', cfilter,
 197         ), env=env)
 198         stop_traffic()
 199         check_packet_count(cmd_capinfos, 0, testout_file)
 200     return check_capture_read_filter_real
 201
 202 @pytest.fixture
 203 def check_capture_snapshot_len(capture_interface, cmd_tshark, traffic_generator, cmd_capinfos, result_file):
 204     start_traffic, cfilter = traffic_generator
 205     def check_capture_snapshot_len_real(self, cmd=None, env=None):
 206         assert cmd is not None
 207         stop_traffic = start_traffic()
 208         testout_file = result_file(testout_pcap)
 209         subprocesstest.check_run(capture_command(cmd,
 210             '-i', capture_interface,
 211             '-p',
 212             '-w', testout_file,
 213             '-s', str(snapshot_len),
 214             '-a', 'duration:{}'.format(capture_duration),
 215             '-f', cfilter,
 216         ), env=env)
 217         stop_traffic()
 218         assert os.path.isfile(testout_file)
 219
 220         # Use tshark to filter out all packets larger than 68 bytes.
 221         testout2_file = result_file('testout2.pcap')
 222
 223         subprocesstest.check_run((cmd_tshark,
 224             '-r', testout_file,
 225             '-w', testout2_file,
 226             '-Y', 'frame.cap_len>{}'.format(snapshot_len),
 227         ), env=env)
 228         check_packet_count(cmd_capinfos, 0, testout2_file)
 229     return check_capture_snapshot_len_real
 230
 231
 232 @pytest.fixture
 233 def check_dumpcap_autostop_stdin(cmd_dumpcap, cmd_capinfos, result_file):
 234     def check_dumpcap_autostop_stdin_real(self, packets=None, filesize=None, env=None):
 235         # Similar to check_capture_stdin.
 236         testout_file = result_file(testout_pcap)
 237         cat100_dhcp_cmd = cat_dhcp_command('cat100')
 238         condition='oops:invalid'
 239
 240         if packets is not None:
 241             condition = 'packets:{}'.format(packets)
 242         elif filesize is not None:
 243             condition = 'filesize:{}'.format(filesize)
 244         else:
 245             raise AssertionError('Need one of packets or filesize')
 246
 247         cmd_ = '"{}"'.format(cmd_dumpcap)
 248         capture_cmd = ' '.join((cmd_,
 249             '-i', '-',
 250             '-w', testout_file,
 251             '-a', condition,
 252         ))
 253         if sysconfig.get_platform().startswith('mingw'):
 254             pytest.skip('FIXME Pipes are broken with the MSYS2 shell')
 255         subprocesstest.check_run(cat100_dhcp_cmd + ' | ' + capture_cmd, shell=True, env=env)
 256         assert os.path.isfile(testout_file)
 257
 258         if packets is not None:
 259             check_packet_count(cmd_capinfos, packets, testout_file)
 260         elif filesize is not None:
 261             capturekb = os.path.getsize(testout_file) / 1000
 262             assert capturekb >= filesize
 263     return check_dumpcap_autostop_stdin_real
 264
 265
 266 @pytest.fixture
 267 def check_dumpcap_ringbuffer_stdin(cmd_dumpcap, cmd_capinfos, result_file):
 268     def check_dumpcap_ringbuffer_stdin_real(self, packets=None, filesize=None, env=None):
 269         # Similar to check_capture_stdin.
 270         rb_unique = 'dhcp_rb_' + uuid.uuid4().hex[:6] # Random ID
 271         testout_file = result_file('testout.{}.pcapng'.format(rb_unique))
 272         testout_glob = result_file('testout.{}_*.pcapng'.format(rb_unique))
 273         cat100_dhcp_cmd = cat_dhcp_command('cat100')
 274         condition='oops:invalid'
 275
 276         if packets is not None:
 277             condition = 'packets:{}'.format(packets)
 278         elif filesize is not None:
 279             condition = 'filesize:{}'.format(filesize)
 280         else:
 281             raise AssertionError('Need one of packets or filesize')
 282
 283         cmd_ = '"{}"'.format(cmd_dumpcap)
 284         capture_cmd = ' '.join((cmd_,
 285             '-i', '-',
 286             '-w', testout_file,
 287             '-a', 'files:2',
 288             '-b', condition,
 289         ))
 290         if sysconfig.get_platform().startswith('mingw'):
 291             pytest.skip('FIXME Pipes are broken with the MSYS2 shell')
 292         subprocesstest.check_run(cat100_dhcp_cmd + ' | ' + capture_cmd, shell=True, env=env)
 293
 294         rb_files = glob.glob(testout_glob)
 295         assert len(rb_files) == 2
 296
 297         for rbf in rb_files:
 298             assert os.path.isfile(rbf)
 299             if packets is not None:
 300                 check_packet_count(cmd_capinfos, packets, rbf)
 301             elif filesize is not None:
 302                 capturekb = os.path.getsize(rbf) / 1000
 303                 assert capturekb >= filesize
 304     return check_dumpcap_ringbuffer_stdin_real
 305
 306
 307 @pytest.fixture
 308 def check_dumpcap_pcapng_sections(cmd_dumpcap, cmd_tshark, cmd_capinfos, capture_file, result_file):
 309     if sys.platform == 'win32':
 310         pytest.skip('Test requires OS fifo support.')
 311     def check_dumpcap_pcapng_sections_real(self, multi_input=False, multi_output=False, env=None):
 312         # Make sure we always test multiple SHBs in an input.
 313         in_files_l = [ [
 314             capture_file('many_interfaces.pcapng.1'),
 315             capture_file('many_interfaces.pcapng.2')
 316             ] ]
 317         if multi_input:
 318             in_files_l.append([ capture_file('many_interfaces.pcapng.3') ])
 319         fifo_files = []
 320         fifo_procs = []
 321         # Default values for our validity tests
 322         check_val_d = {
 323             'filename': None,
 324             'packet_count': 0,
 325             'idb_count': 0,
 326             'ua_pt1_count': 0,
 327             'ua_pt2_count': 0,
 328             'ua_pt3_count': 0,
 329             'ua_dc_count': 0,
 330         }
 331         check_vals = [ check_val_d ]
 332
 333         for in_files in in_files_l:
 334             fifo_file = result_file('dumpcap_pcapng_sections_{}.fifo'.format(len(fifo_files) + 1))
 335             fifo_files.append(fifo_file)
 336             # If a previous test left its fifo laying around, e.g. from a failure, remove it.
 337             try:
 338                 os.unlink(fifo_file)
 339             except Exception: pass
 340             os.mkfifo(fifo_file)
 341             cat_cmd = cat_cap_file_command(in_files)
 342             fifo_procs.append(subprocess.Popen(('{0} > {1}'.format(cat_cmd, fifo_file)), shell=True))
 343
 344         if multi_output:
 345             rb_unique = 'sections_rb_' + uuid.uuid4().hex[:6] # Random ID
 346             testout_file = result_file('testout.{}.pcapng'.format(rb_unique))
 347             testout_glob = result_file('testout.{}_*.pcapng'.format(rb_unique))
 348             check_vals.append(check_val_d.copy())
 349             # check_vals[]['filename'] will be filled in below
 350         else:
 351             testout_file = result_file(testout_pcapng)
 352             check_vals[0]['filename'] = testout_file
 353
 354         # Capture commands
 355         if not multi_input and not multi_output:
 356             # Passthrough SHBs, single output file
 357             capture_cmd_args = (
 358                 '-i', fifo_files[0],
 359                 '-w', testout_file
 360             )
 361             check_vals[0]['packet_count'] = 79
 362             check_vals[0]['idb_count'] = 22
 363             check_vals[0]['ua_pt1_count'] = 1
 364             check_vals[0]['ua_pt2_count'] = 1
 365         elif not multi_input and multi_output:
 366             # Passthrough SHBs, multiple output files
 367             capture_cmd_args = (
 368                 '-i', fifo_files[0],
 369                 '-w', testout_file,
 370                 '-a', 'files:2',
 371                 '-b', 'packets:53'
 372             )
 373             check_vals[0]['packet_count'] = 53
 374             check_vals[0]['idb_count'] = 11
 375             check_vals[0]['ua_pt1_count'] = 1
 376             check_vals[1]['packet_count'] = 26
 377             check_vals[1]['idb_count'] = 22
 378             check_vals[1]['ua_pt1_count'] = 1
 379             check_vals[1]['ua_pt2_count'] = 1
 380         elif multi_input and not multi_output:
 381             # Dumpcap SHBs, single output file
 382             capture_cmd_args = (
 383                 '-i', fifo_files[0],
 384                 '-i', fifo_files[1],
 385                 '-w', testout_file
 386             )
 387             check_vals[0]['packet_count'] = 88
 388             check_vals[0]['idb_count'] = 33
 389             check_vals[0]['ua_dc_count'] = 1
 390         else:
 391             # Dumpcap SHBs, multiple output files
 392             capture_cmd_args = (
 393                 '-i', fifo_files[0],
 394                 '-i', fifo_files[1],
 395                 '-w', testout_file,
 396                 '-a', 'files:2',
 397                 '-b', 'packets:53'
 398             )
 399             check_vals[0]['packet_count'] = 53
 400             check_vals[0]['idb_count'] = 11
 401             check_vals[0]['ua_dc_count'] = 1
 402             check_vals[1]['packet_count'] = 35
 403             check_vals[1]['idb_count'] = 33
 404             check_vals[1]['ua_dc_count'] = 1
 405
 406         capture_cmd = capture_command(cmd_dumpcap, *capture_cmd_args)
 407
 408         subprocesstest.check_run(capture_cmd, env=env)
 409         for fifo_proc in fifo_procs: fifo_proc.kill()
 410
 411         rb_files = []
 412         if multi_output:
 413             rb_files = sorted(glob.glob(testout_glob))
 414             assert len(rb_files) == 2
 415             check_vals[0]['filename'] = rb_files[0]
 416             check_vals[1]['filename'] = rb_files[1]
 417
 418         for rbf in rb_files:
 419             assert os.path.isfile(rbf)
 420
 421         # Output tests
 422
 423         if not multi_input and not multi_output:
 424             # Check strict bit-for-bit passthrough.
 425             in_hash = hashlib.sha256()
 426             out_hash = hashlib.sha256()
 427             for in_file in in_files_l[0]:
 428                 in_cap_file = capture_file(in_file)
 429                 with open(in_cap_file, 'rb') as f:
 430                     in_hash.update(f.read())
 431             with open(testout_file, 'rb') as f:
 432                 out_hash.update(f.read())
 433             assert in_hash.hexdigest() == out_hash.hexdigest()
 434
 435         # many_interfaces.pcapng.1 : 64 packets written by "Passthrough test #1"
 436         # many_interfaces.pcapng.2 : 15 packets written by "Passthrough test #2"
 437         # many_interfaces.pcapng.3 : 9 packets written by "Passthrough test #3"
 438         # Each has 11 interfaces.
 439         idb_compare_eq = True
 440         if multi_input and multi_output:
 441             # Having multiple inputs forces the use of threads. In our
 442             # case this means that non-packet block counts in the first
 443             # file in is nondeterministic.
 444             idb_compare_eq = False
 445         for check_val in check_vals:
 446             check_packet_count(cmd_capinfos, check_val['packet_count'], check_val['filename'])
 447
 448             tshark_proc = subprocesstest.check_run(capture_command(cmd_tshark,
 449                 '-r', check_val['filename'],
 450                 '-V',
 451                 '-X', 'read_format:MIME Files Format'
 452             ), capture_output=True, env=env)
 453             # XXX Are there any other sanity checks we should run?
 454             if idb_compare_eq:
 455                 assert count_output(tshark_proc.stdout, r'Block \d+: Interface Description Block \d+') \
 456                         == check_val['idb_count']
 457             else:
 458                 assert count_output(tshark_proc.stdout, r'Block \d+: Interface Description Block \d+') \
 459                         >= check_val['idb_count']
 460                 idb_compare_eq = True
 461             assert count_output(tshark_proc.stdout, r'Option: User Application = Passthrough test #1') \
 462                         == check_val['ua_pt1_count']
 463             assert count_output(tshark_proc.stdout, r'Option: User Application = Passthrough test #2') \
 464                         == check_val['ua_pt2_count']
 465             assert count_output(tshark_proc.stdout, r'Option: User Application = Passthrough test #3') \
 466                         == check_val['ua_pt3_count']
 467             assert count_output(tshark_proc.stdout, r'Option: User Application = Dumpcap \(Wireshark\)') \
 468                         == check_val['ua_dc_count']
 469     return check_dumpcap_pcapng_sections_real
 470
 471
 472 class TestWiresharkCapture:
 473     def test_wireshark_capture_10_packets_to_file(self, request, wireshark_k, check_capture_10_packets, make_screenshot_on_error, test_env):
 474         '''Capture 10 packets from the network to a file using Wireshark'''
 475         disabled = request.config.getoption('--disable-gui', default=False)
 476         if disabled:
 477             pytest.skip('GUI tests are disabled via --disable-gui')
 478         with make_screenshot_on_error():
 479             check_capture_10_packets(self, cmd=wireshark_k, env=test_env)
 480
 481     # Wireshark doesn't currently support writing to stdout while capturing.
 482     # def test_wireshark_capture_10_packets_to_stdout(self, wireshark_k, check_capture_10_packets):
 483     #     '''Capture 10 packets from the network to stdout using Wireshark'''
 484     #     check_capture_10_packets(self, cmd=wireshark_k, to_stdout=True)
 485
 486     def test_wireshark_capture_from_fifo(self, request, wireshark_k, check_capture_fifo, make_screenshot_on_error, test_env):
 487         '''Capture from a fifo using Wireshark'''
 488         disabled = request.config.getoption('--disable-gui', default=False)
 489         if disabled:
 490             pytest.skip('GUI tests are disabled via --disable-gui')
 491         with make_screenshot_on_error():
 492             check_capture_fifo(self, cmd=wireshark_k, env=test_env)
 493
 494     def test_wireshark_capture_from_stdin(self, request, wireshark_k, check_capture_stdin, make_screenshot_on_error, test_env):
 495         '''Capture from stdin using Wireshark'''
 496         disabled = request.config.getoption('--disable-gui', default=False)
 497         if disabled:
 498             pytest.skip('GUI tests are disabled via --disable-gui')
 499         with make_screenshot_on_error():
 500             check_capture_stdin(self, cmd=wireshark_k, env=test_env)
 501
 502     def test_wireshark_capture_snapshot_len(self, request, wireshark_k, check_capture_snapshot_len, make_screenshot_on_error, test_env):
 503         '''Capture truncated packets using Wireshark'''
 504         disabled = request.config.getoption('--disable-gui', default=False)
 505         if disabled:
 506             pytest.skip('GUI tests are disabled via --disable-gui')
 507         with make_screenshot_on_error():
 508             check_capture_snapshot_len(self, cmd=wireshark_k, env=test_env)
 509
 510
 511 class TestTsharkCapture:
 512     def test_tshark_capture_10_packets_to_file(self, cmd_tshark, check_capture_10_packets, test_env):
 513         '''Capture 10 packets from the network to a file using TShark'''
 514         check_capture_10_packets(self, cmd=cmd_tshark, env=test_env)
 515
 516     def test_tshark_capture_10_packets_to_stdout(self, cmd_tshark, check_capture_10_packets, test_env):
 517         '''Capture 10 packets from the network to stdout using TShark'''
 518         check_capture_10_packets(self, cmd=cmd_tshark, to_stdout=True, env=test_env)
 519
 520     def test_tshark_capture_from_fifo(self, cmd_tshark, check_capture_fifo, test_env):
 521         '''Capture from a fifo using TShark'''
 522         check_capture_fifo(self, cmd=cmd_tshark, env=test_env)
 523
 524     def test_tshark_capture_from_stdin(self, cmd_tshark, check_capture_stdin, test_env):
 525         '''Capture from stdin using TShark'''
 526         check_capture_stdin(self, cmd=cmd_tshark, env=test_env)
 527
 528     def test_tshark_capture_snapshot_len(self, cmd_tshark, check_capture_snapshot_len, test_env):
 529         '''Capture truncated packets using TShark'''
 530         check_capture_snapshot_len(self, cmd=cmd_tshark, env=test_env)
 531
 532
 533 class TestDumpcapCapture:
 534     def test_dumpcap_capture_10_packets_to_file(self, cmd_dumpcap, check_capture_10_packets, base_env):
 535         '''Capture 10 packets from the network to a file using Dumpcap'''
 536         check_capture_10_packets(self, cmd=cmd_dumpcap, env=base_env)
 537
 538     def test_dumpcap_capture_10_packets_to_stdout(self, cmd_dumpcap, check_capture_10_packets, base_env):
 539         '''Capture 10 packets from the network to stdout using Dumpcap'''
 540         check_capture_10_packets(self, cmd=cmd_dumpcap, to_stdout=True, env=base_env)
 541
 542     def test_dumpcap_capture_from_fifo(self, cmd_dumpcap, check_capture_fifo, base_env):
 543         '''Capture from a fifo using Dumpcap'''
 544         check_capture_fifo(self, cmd=cmd_dumpcap, env=base_env)
 545
 546     def test_dumpcap_capture_from_stdin(self, cmd_dumpcap, check_capture_stdin, base_env):
 547         '''Capture from stdin using Dumpcap'''
 548         check_capture_stdin(self, cmd=cmd_dumpcap, env=base_env)
 549
 550     def test_dumpcap_capture_snapshot_len(self, check_capture_snapshot_len, cmd_dumpcap, base_env):
 551         '''Capture truncated packets using Dumpcap'''
 552         check_capture_snapshot_len(self, cmd=cmd_dumpcap, env=base_env)
 553
 554
 555 class TestDumpcapAutostop:
 556     # duration, filesize, packets, files
 557     def test_dumpcap_autostop_filesize(self, check_dumpcap_autostop_stdin, base_env):
 558         '''Capture from stdin using Dumpcap until we reach a file size limit'''
 559         check_dumpcap_autostop_stdin(self, filesize=15, env=base_env)
 560
 561     def test_dumpcap_autostop_packets(self, check_dumpcap_autostop_stdin, base_env):
 562         '''Capture from stdin using Dumpcap until we reach a packet limit'''
 563         check_dumpcap_autostop_stdin(self, packets=97, env=base_env) # Last prime before 100. Arbitrary.
 564
 565
 566 class TestDumpcapRingbuffer:
 567     # duration, interval, filesize, packets, files
 568     def test_dumpcap_ringbuffer_filesize(self, check_dumpcap_ringbuffer_stdin, base_env):
 569         '''Capture from stdin using Dumpcap and write multiple files until we reach a file size limit'''
 570         check_dumpcap_ringbuffer_stdin(self, filesize=15, env=base_env)
 571
 572     def test_dumpcap_ringbuffer_packets(self, check_dumpcap_ringbuffer_stdin, base_env):
 573         '''Capture from stdin using Dumpcap and write multiple files until we reach a packet limit'''
 574         check_dumpcap_ringbuffer_stdin(self, packets=47, env=base_env) # Last prime before 50. Arbitrary.
 575
 576
 577 class TestDumpcapPcapngSections:
 578     def test_dumpcap_pcapng_single_in_single_out(self, check_dumpcap_pcapng_sections, base_env):
 579         '''Capture from a single pcapng source using Dumpcap and write a single file'''
 580         if sys.byteorder == 'big':
 581             pytest.skip('this test is supported on little endian only')
 582         check_dumpcap_pcapng_sections(self, env=base_env)
 583
 584     def test_dumpcap_pcapng_single_in_multi_out(self, check_dumpcap_pcapng_sections, base_env):
 585         '''Capture from a single pcapng source using Dumpcap and write two files'''
 586         if sys.byteorder == 'big':
 587             pytest.skip('this test is supported on little endian only')
 588         check_dumpcap_pcapng_sections(self, multi_output=True, env=base_env)
 589
 590     def test_dumpcap_pcapng_multi_in_single_out(self, check_dumpcap_pcapng_sections, base_env):
 591         '''Capture from two pcapng sources using Dumpcap and write a single file'''
 592         if sys.byteorder == 'big':
 593             pytest.skip('this test is supported on little endian only')
 594         check_dumpcap_pcapng_sections(self, multi_input=True, env=base_env)
 595
 596     def test_dumpcap_pcapng_multi_in_multi_out(self, check_dumpcap_pcapng_sections, base_env):
 597         '''Capture from two pcapng sources using Dumpcap and write two files'''
 598         if sys.byteorder == 'big':
 599             pytest.skip('this test is supported on little endian only')
 600         check_dumpcap_pcapng_sections(self, multi_input=True, multi_output=True, env=base_env)