test/suite_io.py

   1 #
   2 # -*- coding: utf-8 -*-
   3 # Wireshark tests
   4 # By Gerald Combs <gerald@wireshark.org>
   5 #
   6 # Ported from a set of Bash scripts which were copyright 2005 Ulf Lamping
   7 #
   8 # SPDX-License-Identifier: GPL-2.0-or-later
   9 #
  10 '''File I/O tests'''
  11
  12 import io
  13 import os.path
  14 import subprocesstest
  15 import sys
  16 import unittest
  17 import fixtures
  18
  19 testout_pcap = 'testout.pcap'
  20 baseline_file = 'io-rawshark-dhcp-pcap.txt'
  21
  22
  23 @fixtures.fixture(scope='session')
  24 def io_baseline_str(dirs):
  25     with open(os.path.join(dirs.baseline_dir, baseline_file), 'r') as f:
  26         return f.read()
  27
  28
  29 def check_io_4_packets(self, capture_file, cmd=None, from_stdin=False, to_stdout=False):
  30     # Test direct->direct, stdin->direct, and direct->stdout file I/O.
  31     # Similar to suite_capture.check_capture_10_packets and
  32     # suite_capture.check_capture_stdin.
  33     self.assertIsNotNone(cmd)
  34     testout_file = self.filename_from_id(testout_pcap)
  35     if from_stdin and to_stdout:
  36         # XXX If we support this, should we bother with separate stdin->direct
  37         # and direct->stdout tests?
  38         self.fail('Stdin and stdout not supported in the same test.')
  39     elif from_stdin:
  40         # cat -B "${CAPTURE_DIR}dhcp.pcap" | $DUT -r - -w ./testout.pcap 2>./testout.txt
  41         cat_dhcp_cmd = subprocesstest.cat_dhcp_command('cat')
  42         stdin_cmd = '{0} | "{1}" -r - -w "{2}"'.format(cat_dhcp_cmd, cmd, testout_file)
  43         io_proc = self.runProcess(stdin_cmd, shell=True)
  44     elif to_stdout:
  45         # $DUT -r "${CAPTURE_DIR}dhcp.pcap" -w - > ./testout.pcap 2>./testout.txt
  46         stdout_cmd = '"{0}" -r "{1}" -w - > "{2}"'.format(cmd, capture_file('dhcp.pcap'), testout_file)
  47         io_proc = self.runProcess(stdout_cmd, shell=True)
  48     else: # direct->direct
  49         # $DUT -r "${CAPTURE_DIR}dhcp.pcap" -w ./testout.pcap > ./testout.txt 2>&1
  50         io_proc = self.runProcess((cmd,
  51             '-r', capture_file('dhcp.pcap'),
  52             '-w', testout_file,
  53         ))
  54     io_returncode = io_proc.returncode
  55     self.assertEqual(io_returncode, 0)
  56     self.assertTrue(os.path.isfile(testout_file))
  57     if (io_returncode == 0):
  58         self.checkPacketCount(4)
  59
  60
  61 @fixtures.mark_usefixtures('test_env')
  62 @fixtures.uses_fixtures
  63 class case_tshark_io(subprocesstest.SubprocessTestCase):
  64     def test_tshark_io_stdin_direct(self, cmd_tshark, capture_file):
  65         '''Read from stdin and write direct using TShark'''
  66         check_io_4_packets(self, capture_file, cmd=cmd_tshark, from_stdin=True)
  67
  68     def test_tshark_io_direct_stdout(self, cmd_tshark, capture_file):
  69         '''Read direct and write to stdout using TShark'''
  70         check_io_4_packets(self, capture_file, cmd=cmd_tshark, to_stdout=True)
  71
  72     def test_tshark_io_direct_direct(self, cmd_tshark, capture_file):
  73         '''Read direct and write direct using TShark'''
  74         check_io_4_packets(self, capture_file, cmd=cmd_tshark)
  75
  76
  77 @fixtures.mark_usefixtures('test_env')
  78 @fixtures.uses_fixtures
  79 class case_rawshark_io(subprocesstest.SubprocessTestCase):
  80     @unittest.skipUnless(sys.byteorder == 'little', 'Requires a little endian system')
  81     def test_rawshark_io_stdin(self, cmd_rawshark, capture_file, io_baseline_str):
  82         '''Read from stdin using Rawshark'''
  83         # tail -c +25 "${CAPTURE_DIR}dhcp.pcap" | $RAWSHARK -dencap:1 -R "udp.port==68" -nr - > $IO_RAWSHARK_DHCP_PCAP_TESTOUT 2> /dev/null
  84         # diff -u --strip-trailing-cr $IO_RAWSHARK_DHCP_PCAP_BASELINE $IO_RAWSHARK_DHCP_PCAP_TESTOUT > $DIFF_OUT 2>&1
  85         capture_file = capture_file('dhcp.pcap')
  86         testout_file = self.filename_from_id(testout_pcap)
  87         raw_dhcp_cmd = subprocesstest.cat_dhcp_command('raw')
  88         rawshark_cmd = '{0} | "{1}" -r - -n -dencap:1 -R "udp.port==68"'.format(raw_dhcp_cmd, cmd_rawshark)
  89         rawshark_proc = self.runProcess(rawshark_cmd, shell=True)
  90         rawshark_returncode = rawshark_proc.returncode
  91         self.assertEqual(rawshark_returncode, 0)
  92         if (rawshark_returncode == 0):
  93             self.assertTrue(self.diffOutput(rawshark_proc.stdout_str, io_baseline_str, 'rawshark', baseline_file))