Return to cmdparse.py CVS log

Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / mtr / test

mtr 0.94

    1: #!/usr/bin/env python
    2: #
    3: #   mtr  --  a network diagnostic tool
    4: #   Copyright (C) 2016  Matt Kimball
    5: #
    6: #   This program is free software; you can redistribute it and/or modify
    7: #   it under the terms of the GNU General Public License version 2 as
    8: #   published by the Free Software Foundation.
    9: #
   10: #   This program is distributed in the hope that it will be useful,
   11: #   but WITHOUT ANY WARRANTY; without even the implied warranty of
   12: #   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   13: #   GNU General Public License for more details.
   14: #
   15: #   You should have received a copy of the GNU General Public License along
   16: #   with this program; if not, write to the Free Software Foundation, Inc.,
   17: #   51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
   18: #
   19: 
   20: '''Test mtr-packet's command parsing.'''
   21: 
   22: 
   23: import time
   24: import unittest
   25: 
   26: import mtrpacket
   27: 
   28: 
   29: class TestCommandParse(mtrpacket.MtrPacketTest):
   30:     '''Test cases with malformed commands and version checks'''
   31: 
   32:     def test_unknown_command(self):
   33:         'Test sending a command unknown to mtr-packet'
   34: 
   35:         self.write_command('13 argle-bargle')
   36:         self.assertEqual(self.read_reply(), '13 unknown-command')
   37: 
   38:     def test_malformed_command(self):
   39:         'Test sending a malformed command request to mtr-packet'
   40: 
   41:         self.write_command('malformed')
   42:         self.assertEqual(self.read_reply(), '0 command-parse-error')
   43: 
   44:     def test_exit_on_stdin_closed(self):
   45:         '''Test that the packet process terminates after stdin is closed
   46: 
   47:         Test that, when outstanding requests are complete, the process
   48:         terminates following stdin being closed.'''
   49: 
   50:         self.write_command('15 send-probe ip-4 8.8.254.254 timeout 1')
   51:         self.packet_process.stdin.close()
   52:         time.sleep(2)
   53:         self.read_reply()
   54:         exit_code = self.packet_process.poll()
   55:         self.assertIsNotNone(exit_code)
   56: 
   57:     def test_invalid_argument(self):
   58:         'Test sending invalid arguments with probe requests'
   59: 
   60:         bad_commands = [
   61:             '22 send-probe',
   62:             '23 send-probe ip-4 str-value',
   63:             '24 send-probe ip-4 8.8.8.8 timeout str-value',
   64:             '25 send-probe ip-4 8.8.8.8 ttl str-value',
   65:         ]
   66: 
   67:         for cmd in bad_commands:
   68:             self.write_command(cmd)
   69:             reply = self.parse_reply()
   70:             self.assertEqual(reply.command_name, 'invalid-argument')
   71: 
   72:     def test_versioning(self):
   73:         'Test version checks and feature support checks'
   74: 
   75:         feature_tests = [
   76:             ('31 check-support feature ip-4', 'ok'),
   77:             ('32 check-support feature send-probe', 'ok'),
   78:             ('33 check-support feature bogus-feature', 'no')
   79:         ]
   80: 
   81:         self.write_command('30 check-support feature version')
   82:         reply = self.parse_reply()
   83:         self.assertEqual(reply.token, 30)
   84:         self.assertEqual(reply.command_name, 'feature-support')
   85:         self.assertIn('support', reply.argument)
   86: 
   87:         for (request, expected) in feature_tests:
   88:             self.write_command(request)
   89:             reply = self.parse_reply()
   90:             self.assertEqual(reply.command_name, 'feature-support')
   91:             self.assertIn('support', reply.argument)
   92:             self.assertEqual(reply.argument['support'], expected)
   93: 
   94:     def test_command_overflow(self):
   95:         'Test overflowing the incoming command buffer'
   96: 
   97:         big_buffer = 'x' * (64 * 1024)
   98:         self.write_command(big_buffer)
   99: 
  100:         reply = self.read_reply()
  101:         self.assertEqual(reply, '0 command-buffer-overflow')
  102: 
  103: 
  104: if __name__ == '__main__':
  105:     mtrpacket.check_running_as_root()
  106:     unittest.main()