test/suite_sharkd.py

   1 #
   2 # -*- coding: utf-8 -*-
   3 # Wireshark tests
   4 # By Gerald Combs <gerald@wireshark.org>
   5 #
   6 # SPDX-License-Identifier: GPL-2.0-or-later
   7 #
   8 '''sharkd tests'''
   9
  10 import config
  11 import json
  12 import os.path
  13 import subprocess
  14 import subprocesstest
  15 import sys
  16 import unittest
  17
  18 dhcp_pcap = os.path.join(config.capture_dir, 'dhcp.pcap')
  19
  20 class case_sharkd(subprocesstest.SubprocessTestCase):
  21     def test_sharkd_hello_no_pcap(self):
  22         '''sharkd hello message, no capture file'''
  23         sharkd_proc = self.startProcess((config.cmd_sharkd, '-'),
  24             stdin=subprocess.PIPE
  25         )
  26
  27         sharkd_commands = '{"req":"status"}\n'
  28         if sys.version_info[0] >= 3:
  29             sharkd_commands = sharkd_commands.encode('UTF-8')
  30         sharkd_proc.stdin.write(sharkd_commands)
  31         self.waitProcess(sharkd_proc)
  32
  33         self.assertEqual(self.countOutput('Hello in child.', count_stdout=False, count_stderr=True), 1, 'No hello message.')
  34
  35         try:
  36             jdata = json.loads(sharkd_proc.stdout_str)
  37             self.assertEqual(jdata['duration'], 0.0, 'Missing duration.')
  38         except:
  39             self.fail('Invalid JSON: "{}"'.format(sharkd_proc.stdout_str))
  40
  41     def test_sharkd_hello_dhcp_pcap(self):
  42         '''sharkd hello message, simple capture file'''
  43         sharkd_proc = self.startProcess((config.cmd_sharkd, '-'),
  44             stdin=subprocess.PIPE
  45         )
  46
  47         sharkd_commands = ''
  48         sharkd_commands = '{"req":"load","file":' + json.JSONEncoder().encode(dhcp_pcap) + '}\n'
  49         sharkd_commands += '{"req":"status"}\n'
  50         sharkd_commands += '{"req":"frames"}\n'
  51         if sys.version_info[0] >= 3:
  52             sharkd_commands = sharkd_commands.encode('UTF-8')
  53
  54         sharkd_proc.stdin.write(sharkd_commands)
  55         self.waitProcess(sharkd_proc)
  56
  57         has_dhcp = False
  58         for line in sharkd_proc.stdout_str.splitlines():
  59             line = line.strip()
  60             if not line: continue
  61             try:
  62                 jdata = json.loads(line)
  63             except:
  64                 self.fail('Invalid JSON for "{}"'.format(line))
  65
  66             try:
  67                 if 'DHCP' in jdata[0]['c']:
  68                     has_dhcp = True
  69             except:
  70                 pass
  71
  72         self.assertTrue(has_dhcp, 'Failed to find DHCP in JSON output')