Synnefo » snf-ganeti

#!/usr/bin/python

#

# Copyright (C) 2010, 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 the hypervisor.hv_kvm module"""

import threading

import tempfile

import unittest

import socket

import os

import struct

from ganeti import serializer

from ganeti import constants

from ganeti import compat

from ganeti import objects

from ganeti import errors

from ganeti import utils

from ganeti import pathutils

from ganeti.hypervisor import hv_kvm

import testutils

class QmpStub(threading.Thread):

  """Stub for a QMP endpoint for a KVM instance

  """

  _QMP_BANNER_DATA = {

    "QMP": {

      "version": {

        "package": "",

        "qemu": {

          "micro": 50,

          "minor": 13,

          "major": 0,

          },

        "capabilities": [],

        },

      }

    }

  _EMPTY_RESPONSE = {

    "return": [],

    }

  def __init__(self, socket_filename, server_responses):

    """Creates a QMP stub

    @type socket_filename: string

    @param socket_filename: filename of the UNIX socket that will be created

                            this class and used for the communication

    @type server_responses: list

    @param server_responses: list of responses that the server sends in response

                             to whatever it receives

    """

    threading.Thread.__init__(self)

    self.socket_filename = socket_filename

    self.script = server_responses

    self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

    self.socket.bind(self.socket_filename)

    self.socket.listen(1)

  def run(self):

    # Hypothesis: the messages we receive contain only a complete QMP message

    # encoded in JSON.

    conn, addr = self.socket.accept()

    # Send the banner as the first thing

    conn.send(self.encode_string(self._QMP_BANNER_DATA))

    # Expect qmp_capabilities and return an empty response

    conn.recv(4096)

    conn.send(self.encode_string(self._EMPTY_RESPONSE))

    while True:

      # We ignore the expected message, as the purpose of this object is not

      # to verify the correctness of the communication but to act as a

      # partner for the SUT (System Under Test, that is QmpConnection)

      msg = conn.recv(4096)

      if not msg:

        break

      if not self.script:

        break

      response = self.script.pop(0)

      if isinstance(response, str):

        conn.send(response)

      elif isinstance(response, list):

        for chunk in response:

          conn.send(chunk)

      else:

        raise errors.ProgrammerError("Unknown response type for %s" % response)

    conn.close()

  def encode_string(self, message):

    return (serializer.DumpJson(message) +

            hv_kvm.QmpConnection._MESSAGE_END_TOKEN)

class TestQmpMessage(testutils.GanetiTestCase):

  def testSerialization(self):

    test_data = {

      "execute": "command",

      "arguments": ["a", "b", "c"],

      }

    message = hv_kvm.QmpMessage(test_data)

    for k, v in test_data.items():

      self.assertEqual(message[k], v)

    serialized = str(message)

    self.assertEqual(len(serialized.splitlines()), 1,

                     msg="Got multi-line message")

    rebuilt_message = hv_kvm.QmpMessage.BuildFromJsonString(serialized)

    self.assertEqual(rebuilt_message, message)

    self.assertEqual(len(rebuilt_message), len(test_data))

  def testDelete(self):

    toDelete = "execute"

    test_data = {

      toDelete: "command",

      "arguments": ["a", "b", "c"],

      }

    message = hv_kvm.QmpMessage(test_data)

    oldLen = len(message)

    del(message[toDelete])

    newLen = len(message)

    self.assertEqual(oldLen - 1, newLen)

class TestQmp(testutils.GanetiTestCase):

  def testQmp(self):

    requests = [

      {"execute": "query-kvm", "arguments": []},

      {"execute": "eject", "arguments": {"device": "ide1-cd0"}},

      {"execute": "query-status", "arguments": []},

      {"execute": "query-name", "arguments": []},

      ]

    server_responses = [

      # One message, one send()

      '{"return": {"enabled": true, "present": true}}\r\n',

      # Message sent using multiple send()

      ['{"retur', 'n": {}}\r\n'],

      # Multiple messages sent using one send()

      '{"return": [{"name": "quit"}, {"name": "eject"}]}\r\n'

      '{"return": {"running": true, "singlestep": false}}\r\n',

      ]

    expected_responses = [

      {"return": {"enabled": True, "present": True}},

      {"return": {}},

      {"return": [{"name": "quit"}, {"name": "eject"}]},

      {"return": {"running": True, "singlestep": False}},

      ]

    # Set up the stub

    socket_file = tempfile.NamedTemporaryFile()

    os.remove(socket_file.name)

    qmp_stub = QmpStub(socket_file.name, server_responses)

    qmp_stub.start()

    # Set up the QMP connection

    qmp_connection = hv_kvm.QmpConnection(socket_file.name)

    qmp_connection.connect()

    # Format the script

    for request, expected_response in zip(requests, expected_responses):

      response = qmp_connection.Execute(request)

      msg = hv_kvm.QmpMessage(expected_response)

      self.assertEqual(len(str(msg).splitlines()), 1,

                       msg="Got multi-line message")

      self.assertEqual(response, msg)

class TestConsole(unittest.TestCase):

  def _Test(self, instance, node, hvparams):

    cons = hv_kvm.KVMHypervisor.GetInstanceConsole(instance, node, hvparams, {})

    self.assertTrue(cons.Validate())

    return cons

  def testSerial(self):

    instance = objects.Instance(name="kvm.example.com",

                                primary_node="node6017-uuid")

    node = objects.Node(name="node6017", uuid="node6017-uuid")

    hvparams = {

      constants.HV_SERIAL_CONSOLE: True,

      constants.HV_VNC_BIND_ADDRESS: None,

      constants.HV_KVM_SPICE_BIND: None,

      }

    cons = self._Test(instance, node, hvparams)

    self.assertEqual(cons.kind, constants.CONS_SSH)

    self.assertEqual(cons.host, node.name)

    self.assertEqual(cons.command[0], pathutils.KVM_CONSOLE_WRAPPER)

    self.assertEqual(cons.command[1], constants.SOCAT_PATH)

  def testVnc(self):

    instance = objects.Instance(name="kvm.example.com",

                                primary_node="node7235-uuid",

                                network_port=constants.VNC_BASE_PORT + 10)

    node = objects.Node(name="node7235", uuid="node7235-uuid")

    hvparams = {

      constants.HV_SERIAL_CONSOLE: False,

      constants.HV_VNC_BIND_ADDRESS: "192.0.2.1",

      constants.HV_KVM_SPICE_BIND: None,

      }

    cons = self._Test(instance, node, hvparams)

    self.assertEqual(cons.kind, constants.CONS_VNC)

    self.assertEqual(cons.host, "192.0.2.1")

    self.assertEqual(cons.port, constants.VNC_BASE_PORT + 10)

    self.assertEqual(cons.display, 10)

  def testSpice(self):

    instance = objects.Instance(name="kvm.example.com",

                                primary_node="node7235",

                                network_port=11000)

    node = objects.Node(name="node7235", uuid="node7235-uuid")

    hvparams = {

      constants.HV_SERIAL_CONSOLE: False,

      constants.HV_VNC_BIND_ADDRESS: None,

      constants.HV_KVM_SPICE_BIND: "192.0.2.1",

      }

    cons = self._Test(instance, node, hvparams)

    self.assertEqual(cons.kind, constants.CONS_SPICE)

    self.assertEqual(cons.host, "192.0.2.1")

    self.assertEqual(cons.port, 11000)

  def testNoConsole(self):

    instance = objects.Instance(name="kvm.example.com",

                                primary_node="node24325",

                                network_port=0)

    node = objects.Node(name="node24325", uuid="node24325-uuid")

    hvparams = {

      constants.HV_SERIAL_CONSOLE: False,

      constants.HV_VNC_BIND_ADDRESS: None,

      constants.HV_KVM_SPICE_BIND: None,

      }

    cons = self._Test(instance, node, hvparams)

    self.assertEqual(cons.kind, constants.CONS_MESSAGE)

class TestVersionChecking(testutils.GanetiTestCase):

  def testParseVersion(self):

    parse = hv_kvm.KVMHypervisor._ParseKVMVersion

    help_112 = testutils.ReadTestData("kvm_1.1.2_help.txt")

    help_10 = testutils.ReadTestData("kvm_1.0_help.txt")

    help_01590 = testutils.ReadTestData("kvm_0.15.90_help.txt")

    help_0125 = testutils.ReadTestData("kvm_0.12.5_help.txt")

    help_091 = testutils.ReadTestData("kvm_0.9.1_help.txt")

    self.assertEqual(parse(help_112), ("1.1.2", 1, 1, 2))

    self.assertEqual(parse(help_10), ("1.0", 1, 0, 0))

    self.assertEqual(parse(help_01590), ("0.15.90", 0, 15, 90))

    self.assertEqual(parse(help_0125), ("0.12.5", 0, 12, 5))

    self.assertEqual(parse(help_091), ("0.9.1", 0, 9, 1))

class TestSpiceParameterList(unittest.TestCase):

  def test(self):

    defaults = constants.HVC_DEFAULTS[constants.HT_KVM]

    params = \

      compat.UniqueFrozenset(getattr(constants, name)

                             for name in dir(constants)

                             if name.startswith("HV_KVM_SPICE_"))

    # Parameters whose default value evaluates to True and don't need to be set

    defaults_true = frozenset(filter(defaults.__getitem__, params))

    self.assertEqual(defaults_true, frozenset([

      constants.HV_KVM_SPICE_AUDIO_COMPR,

      constants.HV_KVM_SPICE_USE_VDAGENT,

      constants.HV_KVM_SPICE_TLS_CIPHERS,

      ]))

    # HV_KVM_SPICE_BIND decides whether the other parameters must be set if

    # their default evaluates to False

    assert constants.HV_KVM_SPICE_BIND in params

    assert constants.HV_KVM_SPICE_BIND not in defaults_true

    # Exclude some parameters

    params -= defaults_true | frozenset([

      constants.HV_KVM_SPICE_BIND,

      ])

    self.assertEqual(hv_kvm._SPICE_ADDITIONAL_PARAMS, params)

class TestHelpRegexps(testutils.GanetiTestCase):

  def testBootRe(self):

    """Check _BOOT_RE

    It has too match -drive.*boot=on|off except if there is another dash-option

    at the beginning of the line.

    """

    boot_re = hv_kvm.KVMHypervisor._BOOT_RE

    help_112 = testutils.ReadTestData("kvm_1.1.2_help.txt")

    help_10 = testutils.ReadTestData("kvm_1.0_help.txt")

    help_01590 = testutils.ReadTestData("kvm_0.15.90_help.txt")

    help_0125 = testutils.ReadTestData("kvm_0.12.5_help.txt")

    help_091 = testutils.ReadTestData("kvm_0.9.1_help.txt")

    help_091_fake = testutils.ReadTestData("kvm_0.9.1_help_boot_test.txt")

    self.assertTrue(boot_re.search(help_091))

    self.assertTrue(boot_re.search(help_0125))

    self.assertFalse(boot_re.search(help_091_fake))

    self.assertFalse(boot_re.search(help_112))

    self.assertFalse(boot_re.search(help_10))

    self.assertFalse(boot_re.search(help_01590))

class TestGetTunFeatures(unittest.TestCase):

  def testWrongIoctl(self):

    tmpfile = tempfile.NamedTemporaryFile()

    # A file does not have the right ioctls, so this must always fail

    result = hv_kvm._GetTunFeatures(tmpfile.fileno())

    self.assertTrue(result is None)

  def _FakeIoctl(self, features, fd, request, buf):

    self.assertEqual(request, hv_kvm.TUNGETFEATURES)

    (reqno, ) = struct.unpack("I", buf)

    self.assertEqual(reqno, 0)

    return struct.pack("I", features)

  def test(self):

    tmpfile = tempfile.NamedTemporaryFile()

    fd = tmpfile.fileno()

    for features in [0, hv_kvm.IFF_VNET_HDR]:

      fn = compat.partial(self._FakeIoctl, features)

      result = hv_kvm._GetTunFeatures(fd, _ioctl=fn)

      self.assertEqual(result, features)

class TestProbeTapVnetHdr(unittest.TestCase):

  def _FakeTunFeatures(self, expected_fd, flags, fd):

    self.assertEqual(fd, expected_fd)

    return flags

  def test(self):

    tmpfile = tempfile.NamedTemporaryFile()

    fd = tmpfile.fileno()

    for flags in [0, hv_kvm.IFF_VNET_HDR]:

      fn = compat.partial(self._FakeTunFeatures, fd, flags)

      result = hv_kvm._ProbeTapVnetHdr(fd, _features_fn=fn)

      if flags == 0:

        self.assertFalse(result)

      else:

        self.assertTrue(result)

  def testUnsupported(self):

    tmpfile = tempfile.NamedTemporaryFile()

    fd = tmpfile.fileno()

    self.assertFalse(hv_kvm._ProbeTapVnetHdr(fd, _features_fn=lambda _: None))

if __name__ == "__main__":

  testutils.GanetiTestProgram()