code.grnet.gr Git - ganeti-local/blob - test/ganeti.hypervisor.hv_kvm_unittest.py

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2010, 2011 Google Inc.
   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 as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  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.
  15 #
  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
  19 # 02110-1301, USA.
  20
  21
  22 """Script for testing the hypervisor.hv_kvm module"""
  23
  24 import threading
  25 import tempfile
  26 import unittest
  27 import socket
  28 import os
  29
  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
  36
  37 from ganeti.hypervisor import hv_kvm
  38
  39 import testutils
  40
  41
  42 class QmpStub(threading.Thread):
  43   """Stub for a QMP endpoint for a KVM instance
  44
  45   """
  46   _QMP_BANNER_DATA = {
  47     "QMP": {
  48       "version": {
  49         "package": "",
  50         "qemu": {
  51           "micro": 50,
  52           "minor": 13,
  53           "major": 0,
  54           },
  55         "capabilities": [],
  56         },
  57       }
  58     }
  59   _EMPTY_RESPONSE = {
  60     "return": [],
  61     }
  62
  63   def __init__(self, socket_filename, server_responses):
  64     """Creates a QMP stub
  65
  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
  72     """
  73     threading.Thread.__init__(self)
  74     self.socket_filename = socket_filename
  75     self.script = server_responses
  76
  77     self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  78     self.socket.bind(self.socket_filename)
  79     self.socket.listen(1)
  80
  81   def run(self):
  82     # Hypothesis: the messages we receive contain only a complete QMP message
  83     # encoded in JSON.
  84     conn, addr = self.socket.accept()
  85
  86     # Send the banner as the first thing
  87     conn.send(self.encode_string(self._QMP_BANNER_DATA))
  88
  89     # Expect qmp_capabilities and return an empty response
  90     conn.recv(4096)
  91     conn.send(self.encode_string(self._EMPTY_RESPONSE))
  92
  93     while True:
  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)
  97       msg = conn.recv(4096)
  98       if not msg:
  99         break
 100
 101       if not self.script:
 102         break
 103       response = self.script.pop(0)
 104       if isinstance(response, str):
 105         conn.send(response)
 106       elif isinstance(response, list):
 107         for chunk in response:
 108           conn.send(chunk)
 109       else:
 110         raise errors.ProgrammerError("Unknown response type for %s" % response)
 111
 112     conn.close()
 113
 114   def encode_string(self, message):
 115     return (serializer.DumpJson(message) +
 116             hv_kvm.QmpConnection._MESSAGE_END_TOKEN)
 117
 118
 119 class TestQmpMessage(testutils.GanetiTestCase):
 120   def testSerialization(self):
 121     test_data = {
 122       "execute": "command",
 123       "arguments": ["a", "b", "c"],
 124       }
 125     message = hv_kvm.QmpMessage(test_data)
 126
 127     for k, v in test_data.items():
 128       self.assertEqual(message[k], v)
 129
 130     serialized = str(message)
 131     self.assertEqual(len(serialized.splitlines()), 1,
 132                      msg="Got multi-line message")
 133
 134     rebuilt_message = hv_kvm.QmpMessage.BuildFromJsonString(serialized)
 135     self.assertEqual(rebuilt_message, message)
 136
 137
 138 class TestQmp(testutils.GanetiTestCase):
 139   def testQmp(self):
 140     requests = [
 141       {"execute": "query-kvm", "arguments": []},
 142       {"execute": "eject", "arguments": {"device": "ide1-cd0"}},
 143       {"execute": "query-status", "arguments": []},
 144       {"execute": "query-name", "arguments": []},
 145       ]
 146
 147     server_responses = [
 148       # One message, one send()
 149       '{"return": {"enabled": true, "present": true}}\r\n',
 150
 151       # Message sent using multiple send()
 152       ['{"retur', 'n": {}}\r\n'],
 153
 154       # Multiple messages sent using one send()
 155       '{"return": [{"name": "quit"}, {"name": "eject"}]}\r\n'
 156       '{"return": {"running": true, "singlestep": false}}\r\n',
 157       ]
 158
 159     expected_responses = [
 160       {"return": {"enabled": True, "present": True}},
 161       {"return": {}},
 162       {"return": [{"name": "quit"}, {"name": "eject"}]},
 163       {"return": {"running": True, "singlestep": False}},
 164       ]
 165
 166     # Set up the stub
 167     socket_file = tempfile.NamedTemporaryFile()
 168     os.remove(socket_file.name)
 169     qmp_stub = QmpStub(socket_file.name, server_responses)
 170     qmp_stub.start()
 171
 172     # Set up the QMP connection
 173     qmp_connection = hv_kvm.QmpConnection(socket_file.name)
 174     qmp_connection.connect()
 175
 176     # Format the script
 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)
 183
 184
 185 class TestConsole(unittest.TestCase):
 186   def _Test(self, instance, hvparams):
 187     cons = hv_kvm.KVMHypervisor.GetInstanceConsole(instance, hvparams, {})
 188     self.assertTrue(cons.Validate())
 189     return cons
 190
 191   def testSerial(self):
 192     instance = objects.Instance(name="kvm.example.com",
 193                                 primary_node="node6017")
 194     hvparams = {
 195       constants.HV_SERIAL_CONSOLE: True,
 196       constants.HV_VNC_BIND_ADDRESS: None,
 197       constants.HV_KVM_SPICE_BIND: None,
 198       }
 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)
 204
 205   def testVnc(self):
 206     instance = objects.Instance(name="kvm.example.com",
 207                                 primary_node="node7235",
 208                                 network_port=constants.VNC_BASE_PORT + 10)
 209     hvparams = {
 210       constants.HV_SERIAL_CONSOLE: False,
 211       constants.HV_VNC_BIND_ADDRESS: "192.0.2.1",
 212       constants.HV_KVM_SPICE_BIND: None,
 213       }
 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)
 219
 220   def testSpice(self):
 221     instance = objects.Instance(name="kvm.example.com",
 222                                 primary_node="node7235",
 223                                 network_port=11000)
 224     hvparams = {
 225       constants.HV_SERIAL_CONSOLE: False,
 226       constants.HV_VNC_BIND_ADDRESS: None,
 227       constants.HV_KVM_SPICE_BIND: "192.0.2.1",
 228       }
 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)
 233
 234   def testNoConsole(self):
 235     instance = objects.Instance(name="kvm.example.com",
 236                                 primary_node="node24325",
 237                                 network_port=0)
 238     hvparams = {
 239       constants.HV_SERIAL_CONSOLE: False,
 240       constants.HV_VNC_BIND_ADDRESS: None,
 241       constants.HV_KVM_SPICE_BIND: None,
 242       }
 243     cons = self._Test(instance, hvparams)
 244     self.assertEqual(cons.kind, constants.CONS_MESSAGE)
 245
 246
 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))
 258
 259
 260 if __name__ == "__main__":
 261   testutils.GanetiTestProgram()