4 # Copyright (C) 2010, 2011 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Script for testing the hypervisor.hv_kvm module"""
30 from ganeti import serializer
31 from ganeti import constants
32 from ganeti import compat
33 from ganeti import objects
34 from ganeti import errors
35 from ganeti import utils
37 from ganeti.hypervisor import hv_kvm
42 class QmpStub(threading.Thread):
43 """Stub for a QMP endpoint for a KVM instance
63 def __init__(self, socket_filename, server_responses):
66 @type socket_filename: string
67 @param socket_filename: filename of the UNIX socket that will be created
68 this class and used for the communication
69 @type server_responses: list
70 @param server_responses: list of responses that the server sends in response
71 to whatever it receives
73 threading.Thread.__init__(self)
74 self.socket_filename = socket_filename
75 self.script = server_responses
77 self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
78 self.socket.bind(self.socket_filename)
82 # Hypothesis: the messages we receive contain only a complete QMP message
84 conn, addr = self.socket.accept()
86 # Send the banner as the first thing
87 conn.send(self.encode_string(self._QMP_BANNER_DATA))
89 # Expect qmp_capabilities and return an empty response
91 conn.send(self.encode_string(self._EMPTY_RESPONSE))
94 # We ignore the expected message, as the purpose of this object is not
95 # to verify the correctness of the communication but to act as a
96 # partner for the SUT (System Under Test, that is QmpConnection)
103 response = self.script.pop(0)
104 if isinstance(response, str):
106 elif isinstance(response, list):
107 for chunk in response:
110 raise errors.ProgrammerError("Unknown response type for %s" % response)
114 def encode_string(self, message):
115 return (serializer.DumpJson(message) +
116 hv_kvm.QmpConnection._MESSAGE_END_TOKEN)
119 class TestQmpMessage(testutils.GanetiTestCase):
120 def testSerialization(self):
122 "execute": "command",
123 "arguments": ["a", "b", "c"],
125 message = hv_kvm.QmpMessage(test_data)
127 for k, v in test_data.items():
128 self.assertEqual(message[k], v)
130 serialized = str(message)
131 self.assertEqual(len(serialized.splitlines()), 1,
132 msg="Got multi-line message")
134 rebuilt_message = hv_kvm.QmpMessage.BuildFromJsonString(serialized)
135 self.assertEqual(rebuilt_message, message)
138 class TestQmp(testutils.GanetiTestCase):
141 {"execute": "query-kvm", "arguments": []},
142 {"execute": "eject", "arguments": {"device": "ide1-cd0"}},
143 {"execute": "query-status", "arguments": []},
144 {"execute": "query-name", "arguments": []},
148 # One message, one send()
149 '{"return": {"enabled": true, "present": true}}\r\n',
151 # Message sent using multiple send()
152 ['{"retur', 'n": {}}\r\n'],
154 # Multiple messages sent using one send()
155 '{"return": [{"name": "quit"}, {"name": "eject"}]}\r\n'
156 '{"return": {"running": true, "singlestep": false}}\r\n',
159 expected_responses = [
160 {"return": {"enabled": True, "present": True}},
162 {"return": [{"name": "quit"}, {"name": "eject"}]},
163 {"return": {"running": True, "singlestep": False}},
167 socket_file = tempfile.NamedTemporaryFile()
168 os.remove(socket_file.name)
169 qmp_stub = QmpStub(socket_file.name, server_responses)
172 # Set up the QMP connection
173 qmp_connection = hv_kvm.QmpConnection(socket_file.name)
174 qmp_connection.connect()
177 for request, expected_response in zip(requests, expected_responses):
178 response = qmp_connection.Execute(request)
179 msg = hv_kvm.QmpMessage(expected_response)
180 self.assertEqual(len(str(msg).splitlines()), 1,
181 msg="Got multi-line message")
182 self.assertEqual(response, msg)
185 class TestConsole(unittest.TestCase):
186 def _Test(self, instance, hvparams):
187 cons = hv_kvm.KVMHypervisor.GetInstanceConsole(instance, hvparams, {})
188 self.assertTrue(cons.Validate())
191 def testSerial(self):
192 instance = objects.Instance(name="kvm.example.com",
193 primary_node="node6017")
195 constants.HV_SERIAL_CONSOLE: True,
196 constants.HV_VNC_BIND_ADDRESS: None,
197 constants.HV_KVM_SPICE_BIND: None,
199 cons = self._Test(instance, hvparams)
200 self.assertEqual(cons.kind, constants.CONS_SSH)
201 self.assertEqual(cons.host, instance.primary_node)
202 self.assertEqual(cons.command[0], constants.KVM_CONSOLE_WRAPPER)
203 self.assertEqual(cons.command[1], constants.SOCAT_PATH)
206 instance = objects.Instance(name="kvm.example.com",
207 primary_node="node7235",
208 network_port=constants.VNC_BASE_PORT + 10)
210 constants.HV_SERIAL_CONSOLE: False,
211 constants.HV_VNC_BIND_ADDRESS: "192.0.2.1",
212 constants.HV_KVM_SPICE_BIND: None,
214 cons = self._Test(instance, hvparams)
215 self.assertEqual(cons.kind, constants.CONS_VNC)
216 self.assertEqual(cons.host, "192.0.2.1")
217 self.assertEqual(cons.port, constants.VNC_BASE_PORT + 10)
218 self.assertEqual(cons.display, 10)
221 instance = objects.Instance(name="kvm.example.com",
222 primary_node="node7235",
225 constants.HV_SERIAL_CONSOLE: False,
226 constants.HV_VNC_BIND_ADDRESS: None,
227 constants.HV_KVM_SPICE_BIND: "192.0.2.1",
229 cons = self._Test(instance, hvparams)
230 self.assertEqual(cons.kind, constants.CONS_SPICE)
231 self.assertEqual(cons.host, "192.0.2.1")
232 self.assertEqual(cons.port, 11000)
234 def testNoConsole(self):
235 instance = objects.Instance(name="kvm.example.com",
236 primary_node="node24325",
239 constants.HV_SERIAL_CONSOLE: False,
240 constants.HV_VNC_BIND_ADDRESS: None,
241 constants.HV_KVM_SPICE_BIND: None,
243 cons = self._Test(instance, hvparams)
244 self.assertEqual(cons.kind, constants.CONS_MESSAGE)
247 class TestVersionChecking(testutils.GanetiTestCase):
248 def testParseVersion(self):
249 parse = hv_kvm.KVMHypervisor._ParseKVMVersion
250 help_10 = utils.ReadFile(self._TestDataFilename("kvm_1.0_help.txt"))
251 help_01590 = utils.ReadFile(self._TestDataFilename("kvm_0.15.90_help.txt"))
252 help_0125 = utils.ReadFile(self._TestDataFilename("kvm_0.12.5_help.txt"))
253 help_091 = utils.ReadFile(self._TestDataFilename("kvm_0.9.1_help.txt"))
254 self.assertEqual(parse(help_10), ("1.0", 1, 0, 0))
255 self.assertEqual(parse(help_01590), ("0.15.90", 0, 15, 90))
256 self.assertEqual(parse(help_0125), ("0.12.5", 0, 12, 5))
257 self.assertEqual(parse(help_091), ("0.9.1", 0, 9, 1))
260 if __name__ == "__main__":
261 testutils.GanetiTestProgram()