Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
#!/usr/bin/python
#
# Copyright (C) 2011 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
"""Script for testing ganeti.client.gnt_instance"""
import unittest
from ganeti import constants
from ganeti import utils
from ganeti import errors
from ganeti import objects
from ganeti.client import gnt_instance
import testutils
class TestConsole(unittest.TestCase):
def setUp(self):
self._output = []
self._cmds = []
self._next_cmd_exitcode = 0
def _Test(self, console, show_command, cluster_name):
return gnt_instance._DoConsole(console, show_command, cluster_name,
feedback_fn=self._Feedback,
_runcmd_fn=self._FakeRunCmd)
def _Feedback(self, msg, *args):
if args:
msg = msg % args
self._output.append(msg)
def _FakeRunCmd(self, cmd, interactive=None):
self.assertTrue(interactive)
self.assertTrue(isinstance(cmd, list))
self._cmds.append(cmd)
return utils.RunResult(self._next_cmd_exitcode, None, "", "", "cmd",
utils.process._TIMEOUT_NONE, 5)
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def testMessage(self):
cons = objects.InstanceConsole(instance="inst98.example.com",
kind=constants.CONS_MESSAGE,
message="Hello World")
self.assertEqual(self._Test(cons, False, "cluster.example.com"),
constants.EXIT_SUCCESS)
self.assertEqual(len(self._cmds), 0)
self.assertEqual(self._output, ["Hello World"])
def testVnc(self):
cons = objects.InstanceConsole(instance="inst1.example.com",
kind=constants.CONS_VNC,
host="node1.example.com",
port=5901,
display=1)
self.assertEqual(self._Test(cons, False, "cluster.example.com"),
constants.EXIT_SUCCESS)
self.assertEqual(len(self._cmds), 0)
self.assertEqual(len(self._output), 1)
self.assertTrue(" inst1.example.com " in self._output[0])
self.assertTrue(" node1.example.com:5901 " in self._output[0])
self.assertTrue("vnc://node1.example.com:5901/" in self._output[0])
def testSshShow(self):
cons = objects.InstanceConsole(instance="inst31.example.com",
kind=constants.CONS_SSH,
host="node93.example.com",
user="user_abc",
command="xm console x.y.z")
self.assertEqual(self._Test(cons, True, "cluster.example.com"),
constants.EXIT_SUCCESS)
self.assertEqual(len(self._cmds), 0)
self.assertEqual(len(self._output), 1)
self.assertTrue(" user_abc@node93.example.com " in self._output[0])
self.assertTrue("'xm console x.y.z'" in self._output[0])
def testSshRun(self):
cons = objects.InstanceConsole(instance="inst31.example.com",
kind=constants.CONS_SSH,
host="node93.example.com",
user="user_abc",
command=["xm", "console", "x.y.z"])
self.assertEqual(self._Test(cons, False, "cluster.example.com"),
constants.EXIT_SUCCESS)
self.assertEqual(len(self._cmds), 1)
self.assertEqual(len(self._output), 0)
# This is very important to prevent escapes from the console
self.assertTrue("-oEscapeChar=none" in self._cmds[0])
def testSshRunFail(self):
cons = objects.InstanceConsole(instance="inst31.example.com",
kind=constants.CONS_SSH,
host="node93.example.com",
user="user_abc",
command=["xm", "console", "x.y.z"])
self._next_cmd_exitcode = 100
self.assertRaises(errors.OpExecError, self._Test,
cons, False, "cluster.example.com")
self.assertEqual(len(self._cmds), 1)
self.assertEqual(len(self._output), 0)
if __name__ == "__main__":
testutils.GanetiTestProgram()