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 23import time 24import unittest 25 26import mtrpacket 27 28 29class 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 104if __name__ == '__main__': 105 mtrpacket.check_running_as_root() 106 unittest.main() 107