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