# 02110-1301, USA.
-"""Script for testing ganeti.hypervisor.hv_lxc"""
+"""Script for testing ganeti.hypervisor.hv_xen"""
import string # pylint: disable=W0402
import unittest
import shutil
import random
import os
+import mock
from ganeti import constants
from ganeti import objects
+from ganeti import pathutils
from ganeti import hypervisor
from ganeti import utils
from ganeti import errors
class TestConsole(unittest.TestCase):
def test(self):
- for cls in [hv_xen.XenPvmHypervisor, hv_xen.XenHvmHypervisor]:
+ hvparams = {constants.HV_XEN_CMD: constants.XEN_CMD_XL}
+ for cls in [hv_xen.XenPvmHypervisor(), hv_xen.XenHvmHypervisor()]:
instance = objects.Instance(name="xen.example.com",
- primary_node="node24828")
- cons = cls.GetInstanceConsole(instance, {}, {})
+ primary_node="node24828-uuid")
+ node = objects.Node(name="node24828", uuid="node24828-uuid")
+ cons = cls.GetInstanceConsole(instance, node, hvparams, {})
self.assertTrue(cons.Validate())
self.assertEqual(cons.kind, constants.CONS_SSH)
- self.assertEqual(cons.host, instance.primary_node)
+ self.assertEqual(cons.host, node.name)
self.assertEqual(cons.command[-1], instance.name)
constants.CPU_PINNING_ALL_XEN))
-class TestParseXmList(testutils.GanetiTestCase):
+class TestGetCommand(testutils.GanetiTestCase):
+ def testCommandExplicit(self):
+ """Test the case when the command is given as class parameter explicitly.
+
+ """
+ expected_cmd = "xl"
+ hv = hv_xen.XenHypervisor(_cmd=constants.XEN_CMD_XL)
+ self.assertEqual(hv._GetCommand(None), expected_cmd)
+
+ def testCommandInvalid(self):
+ """Test the case an invalid command is given as class parameter explicitly.
+
+ """
+ hv = hv_xen.XenHypervisor(_cmd="invalidcommand")
+ self.assertRaises(errors.ProgrammerError, hv._GetCommand, None)
+
+ def testCommandHvparams(self):
+ expected_cmd = "xl"
+ test_hvparams = {constants.HV_XEN_CMD: constants.XEN_CMD_XL}
+ hv = hv_xen.XenHypervisor()
+ self.assertEqual(hv._GetCommand(test_hvparams), expected_cmd)
+
+ def testCommandHvparamsInvalid(self):
+ test_hvparams = {}
+ hv = hv_xen.XenHypervisor()
+ self.assertRaises(errors.HypervisorError, hv._GetCommand, test_hvparams)
+
+ def testCommandHvparamsCmdInvalid(self):
+ test_hvparams = {constants.HV_XEN_CMD: "invalidcommand"}
+ hv = hv_xen.XenHypervisor()
+ self.assertRaises(errors.ProgrammerError, hv._GetCommand, test_hvparams)
+
+
+class TestParseInstanceList(testutils.GanetiTestCase):
def test(self):
data = testutils.ReadTestData("xen-xm-list-4.0.1-dom0-only.txt")
# Exclude node
- self.assertEqual(hv_xen._ParseXmList(data.splitlines(), False), [])
+ self.assertEqual(hv_xen._ParseInstanceList(data.splitlines(), False), [])
# Include node
- result = hv_xen._ParseXmList(data.splitlines(), True)
+ result = hv_xen._ParseInstanceList(data.splitlines(), True)
self.assertEqual(len(result), 1)
self.assertEqual(len(result[0]), 6)
for lines in tests:
try:
- hv_xen._ParseXmList(["Header would be here"] + lines, False)
+ hv_xen._ParseInstanceList(["Header would be here"] + lines, False)
except errors.HypervisorError, err:
- self.assertTrue("Can't parse output of xm list" in str(err))
+ self.assertTrue("Can't parse instance list" in str(err))
else:
self.fail("Exception was not raised")
-class TestGetXmList(testutils.GanetiTestCase):
+class TestGetInstanceList(testutils.GanetiTestCase):
def _Fail(self):
return utils.RunResult(constants.EXIT_FAILURE, None,
"stdout", "stderr", None,
def testTimeout(self):
fn = testutils.CallCounter(self._Fail)
try:
- hv_xen._GetXmList(fn, False, _timeout=0.1)
+ hv_xen._GetInstanceList(fn, False, _timeout=0.1)
except errors.HypervisorError, err:
self.assertTrue("timeout exceeded" in str(err))
else:
fn = testutils.CallCounter(compat.partial(self._Success, data))
- result = hv_xen._GetXmList(fn, True, _timeout=0.1)
+ result = hv_xen._GetInstanceList(fn, True, _timeout=0.1)
self.assertEqual(len(result), 4)
class TestMergeInstanceInfo(testutils.GanetiTestCase):
def testEmpty(self):
- self.assertEqual(hv_xen._MergeInstanceInfo({}, lambda _: []), {})
+ self.assertEqual(hv_xen._MergeInstanceInfo({}, []), {})
def _FakeXmList(self, include_node):
- self.assertTrue(include_node)
return [
(hv_xen._DOM0_NAME, NotImplemented, 4096, 7, NotImplemented,
NotImplemented),
]
def testMissingNodeInfo(self):
- result = hv_xen._MergeInstanceInfo({}, self._FakeXmList)
+ instance_list = self._FakeXmList(True)
+ result = hv_xen._MergeInstanceInfo({}, instance_list)
self.assertEqual(result, {
"memory_dom0": 4096,
- "dom0_cpus": 7,
+ "cpu_dom0": 7,
})
def testWithNodeInfo(self):
info = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
- result = hv_xen._GetNodeInfo(info, self._FakeXmList)
+ instance_list = self._FakeXmList(True)
+ result = hv_xen._GetNodeInfo(info, instance_list)
self.assertEqual(result, {
"cpu_nodes": 1,
"cpu_sockets": 2,
"cpu_total": 4,
- "dom0_cpus": 7,
+ "cpu_dom0": 7,
"hv_version": (4, 0),
"memory_dom0": 4096,
"memory_free": 8004,
def testManyDisks(self):
for offset in [0, 1, 10]:
- disks = [(objects.Disk(dev_type=constants.LD_LV), "/tmp/disk/%s" % idx)
+ disks = [(objects.Disk(dev_type=constants.DT_PLAIN), "/tmp/disk/%s" % idx)
for idx in range(len(hv_xen._DISK_LETTERS) + offset)]
if offset == 0:
def testTwoLvDisksWithMode(self):
disks = [
- (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDWR),
+ (objects.Disk(dev_type=constants.DT_PLAIN, mode=constants.DISK_RDWR),
"/tmp/diskFirst"),
- (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDONLY),
+ (objects.Disk(dev_type=constants.DT_PLAIN, mode=constants.DISK_RDONLY),
"/tmp/diskLast"),
]
def testFileDisks(self):
disks = [
- (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
+ (objects.Disk(dev_type=constants.DT_FILE, mode=constants.DISK_RDWR,
physical_id=[constants.FD_LOOP]),
"/tmp/diskFirst"),
- (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDONLY,
+ (objects.Disk(dev_type=constants.DT_FILE, mode=constants.DISK_RDONLY,
physical_id=[constants.FD_BLKTAP]),
"/tmp/diskTwo"),
- (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
+ (objects.Disk(dev_type=constants.DT_FILE, mode=constants.DISK_RDWR,
physical_id=[constants.FD_LOOP]),
"/tmp/diskThree"),
- (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
+ (objects.Disk(dev_type=constants.DT_FILE, mode=constants.DISK_RDWR,
physical_id=[constants.FD_BLKTAP]),
"/tmp/diskLast"),
]
def testInvalidFileDisk(self):
disks = [
- (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
+ (objects.Disk(dev_type=constants.DT_FILE, mode=constants.DISK_RDWR,
physical_id=["#unknown#"]),
"/tmp/diskinvalid"),
]
self.assertRaises(KeyError, hv_xen._GetConfigFileDiskData, disks, "sd")
-class TestXenHypervisorUnknownCommand(unittest.TestCase):
- def test(self):
+class TestXenHypervisorRunXen(unittest.TestCase):
+
+ XEN_SUB_CMD = "help"
+
+ def testCommandUnknown(self):
cmd = "#unknown command#"
self.assertFalse(cmd in constants.KNOWN_XEN_COMMANDS)
hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
_run_cmd_fn=NotImplemented,
_cmd=cmd)
- self.assertRaises(errors.ProgrammerError, hv._RunXen, [])
+ self.assertRaises(errors.ProgrammerError, hv._RunXen, [], None)
+
+ def testCommandNoHvparams(self):
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=NotImplemented)
+ hvparams = None
+ self.assertRaises(errors.HypervisorError, hv._RunXen, [self.XEN_SUB_CMD],
+ hvparams)
+
+ def testCommandFromHvparams(self):
+ expected_xen_cmd = "xl"
+ hvparams = {constants.HV_XEN_CMD: constants.XEN_CMD_XL}
+ mock_run_cmd = mock.Mock()
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ hv._RunXen([self.XEN_SUB_CMD], hvparams=hvparams)
+ mock_run_cmd.assert_called_with([expected_xen_cmd, self.XEN_SUB_CMD])
+
+
+class TestXenHypervisorGetInstanceList(unittest.TestCase):
+
+ RESULT_OK = utils.RunResult(0, None, "", "", "", None, None)
+ XEN_LIST = "list"
+
+ def testNoHvparams(self):
+ expected_xen_cmd = "xm"
+ mock_run_cmd = mock.Mock( return_value=self.RESULT_OK )
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ self.assertRaises(errors.HypervisorError, hv._GetInstanceList, True, None)
+
+ def testFromHvparams(self):
+ expected_xen_cmd = "xl"
+ hvparams = {constants.HV_XEN_CMD: constants.XEN_CMD_XL}
+ mock_run_cmd = mock.Mock( return_value=self.RESULT_OK )
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ hv._GetInstanceList(True, hvparams)
+ mock_run_cmd.assert_called_with([expected_xen_cmd, self.XEN_LIST])
+
+
+class TestXenHypervisorListInstances(unittest.TestCase):
+
+ RESULT_OK = utils.RunResult(0, None, "", "", "", None, None)
+ XEN_LIST = "list"
+
+ def testNoHvparams(self):
+ expected_xen_cmd = "xm"
+ mock_run_cmd = mock.Mock( return_value=self.RESULT_OK )
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ self.assertRaises(errors.HypervisorError, hv.ListInstances)
+
+ def testHvparamsXl(self):
+ expected_xen_cmd = "xl"
+ hvparams = {constants.HV_XEN_CMD: constants.XEN_CMD_XL}
+ mock_run_cmd = mock.Mock( return_value=self.RESULT_OK )
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ hv.ListInstances(hvparams=hvparams)
+ mock_run_cmd.assert_called_with([expected_xen_cmd, self.XEN_LIST])
+
+
+class TestXenHypervisorCheckToolstack(unittest.TestCase):
+
+ def setUp(self):
+ self.tmpdir = tempfile.mkdtemp()
+ self.cfg_name = "xen_config"
+ self.cfg_path = utils.PathJoin(self.tmpdir, self.cfg_name)
+ self.hv = hv_xen.XenHypervisor()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def testBinaryNotFound(self):
+ RESULT_FAILED = utils.RunResult(1, None, "", "", "", None, None)
+ mock_run_cmd = mock.Mock(return_value=RESULT_FAILED)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ result = hv._CheckToolstackBinary("xl")
+ self.assertFalse(result)
+
+ def testCheckToolstackXlConfigured(self):
+ RESULT_OK = utils.RunResult(0, None, "", "", "", None, None)
+ mock_run_cmd = mock.Mock(return_value=RESULT_OK)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ result = hv._CheckToolstackXlConfigured()
+ self.assertTrue(result)
+
+ def testCheckToolstackXlNotConfigured(self):
+ RESULT_FAILED = utils.RunResult(
+ 1, None, "",
+ "ERROR: A different toolstack (xm) have been selected!",
+ "", None, None)
+ mock_run_cmd = mock.Mock(return_value=RESULT_FAILED)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ result = hv._CheckToolstackXlConfigured()
+ self.assertFalse(result)
+
+ def testCheckToolstackXlFails(self):
+ RESULT_FAILED = utils.RunResult(
+ 1, None, "",
+ "ERROR: The pink bunny hid the binary.",
+ "", None, None)
+ mock_run_cmd = mock.Mock(return_value=RESULT_FAILED)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ self.assertRaises(errors.HypervisorError, hv._CheckToolstackXlConfigured)
class TestXenHypervisorWriteConfigFile(unittest.TestCase):
self.fail("Exception was not raised")
+class TestXenHypervisorVerify(unittest.TestCase):
+
+ def setUp(self):
+ output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
+ self._result_ok = utils.RunResult(0, None, output, "", "", None, None)
+
+ def testVerify(self):
+ hvparams = {constants.HV_XEN_CMD : constants.XEN_CMD_XL}
+ mock_run_cmd = mock.Mock(return_value=self._result_ok)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ hv._CheckToolstack = mock.Mock(return_value=True)
+ result = hv.Verify(hvparams)
+ self.assertTrue(result is None)
+
+ def testVerifyToolstackNotOk(self):
+ hvparams = {constants.HV_XEN_CMD : constants.XEN_CMD_XL}
+ mock_run_cmd = mock.Mock(return_value=self._result_ok)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ hv._CheckToolstack = mock.Mock()
+ hv._CheckToolstack.side_effect = errors.HypervisorError("foo")
+ result = hv.Verify(hvparams)
+ self.assertTrue(result is not None)
+
+ def testVerifyFailing(self):
+ result_failed = utils.RunResult(1, None, "", "", "", None, None)
+ mock_run_cmd = mock.Mock(return_value=result_failed)
+ hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
+ _run_cmd_fn=mock_run_cmd)
+ hv._CheckToolstack = mock.Mock(return_value=True)
+ result = hv.Verify()
+ self.assertTrue(result is not None)
+
+
class _TestXenHypervisor(object):
TARGET = NotImplemented
CMD = NotImplemented
HVNAME = NotImplemented
+ VALID_HVPARAMS = {constants.HV_XEN_CMD: constants.XEN_CMD_XL}
def setUp(self):
super(_TestXenHypervisor, self).setUp()
"testinstance.example.com",
])
- def testVerify(self):
- output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
- hv = self._GetHv(run_cmd=compat.partial(self._SuccessCommand,
- output))
- self.assertTrue(hv.Verify() is None)
-
- def testVerifyFailing(self):
- hv = self._GetHv(run_cmd=self._FailingCommand)
- self.assertTrue("failed:" in hv.Verify())
-
def _StartInstanceCommand(self, inst, paused, failcreate, cmd):
if cmd == [self.CMD, "info"]:
output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
self.fail("Unhandled command: %s" % (cmd, ))
return self._SuccessCommand(output, cmd)
- #return self._FailingCommand(cmd)
def _MakeInstance(self):
# Copy default parameters
hvp[constants.HV_VNC_PASSWORD_FILE] = self.vncpw_path
disks = [
- (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDWR),
+ (objects.Disk(dev_type=constants.DT_PLAIN, mode=constants.DISK_RDWR),
utils.PathJoin(self.tmpdir, "disk0")),
- (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDONLY),
+ (objects.Disk(dev_type=constants.DT_PLAIN, mode=constants.DISK_RDONLY),
utils.PathJoin(self.tmpdir, "disk1")),
]
def testStartInstance(self):
(inst, disks) = self._MakeInstance()
+ pathutils.LOG_XEN_DIR = self.tmpdir
for failcreate in [False, True]:
for paused in [False, True]:
if failcreate:
self.assertRaises(errors.HypervisorError, hv.StartInstance,
inst, disks, paused)
+ # Check whether a stale config file is left behind
+ self.assertFalse(os.path.exists(cfgfile))
else:
hv.StartInstance(inst, disks, paused)
-
- # Check if configuration was updated
- lines = utils.ReadFile(cfgfile).splitlines()
+ # Check if configuration was updated
+ lines = utils.ReadFile(cfgfile).splitlines()
if constants.HV_VNC_PASSWORD_FILE in inst.hvparams:
self.assertTrue(("vncpasswd = '%s'" % self.vncpw) in lines)
self.assertTrue(("extra = '%s'" % extra) in lines)
def _StopInstanceCommand(self, instance_name, force, fail, cmd):
- if ((force and cmd[:2] == [self.CMD, "destroy"]) or
- (not force and cmd[:2] == [self.CMD, "shutdown"])):
+ if (cmd == [self.CMD, "list"]):
+ output = "Name ID Mem VCPUs State Time(s)\n" \
+ "Domain-0 0 1023 1 r----- 142691.0\n" \
+ "%s 417 128 1 r----- 3.2\n" % instance_name
+ elif cmd[:2] == [self.CMD, "destroy"]:
self.assertEqual(cmd[2:], [instance_name])
output = ""
+ elif not force and cmd[:3] == [self.CMD, "shutdown", "-w"]:
+ self.assertEqual(cmd[3:], [instance_name])
+ output = ""
else:
self.fail("Unhandled command: %s" % (cmd, ))
if fail:
try:
- hv._StopInstance(name, force)
+ hv._StopInstance(name, force, None)
except errors.HypervisorError, err:
- self.assertTrue(str(err).startswith("Failed to stop instance"))
+ self.assertTrue(str(err).startswith("listing instances failed"),
+ msg=str(err))
else:
self.fail("Exception was not raised")
self.assertEqual(utils.ReadFile(cfgfile), cfgdata,
msg=("Configuration was removed when stopping"
" instance failed"))
else:
- hv._StopInstance(name, force)
+ hv._StopInstance(name, force, None)
self.assertFalse(os.path.exists(cfgfile))
def _MigrateNonRunningInstCmd(self, cmd):
for live in [False, True]:
try:
hv._MigrateInstance(NotImplemented, name, target, port, live,
- _ping_fn=NotImplemented)
+ self.VALID_HVPARAMS, _ping_fn=NotImplemented)
except errors.HypervisorError, err:
self.assertEqual(str(err), "Instance not running, cannot migrate")
else:
port = 28349
hv = self._GetHv(run_cmd=self._MigrateInstTargetUnreachCmd)
+ hvparams = {constants.HV_XEN_CMD: self.CMD}
for live in [False, True]:
if self.CMD == constants.XEN_CMD_XL:
else:
try:
hv._MigrateInstance(NotImplemented, name, target, port, live,
+ hvparams,
_ping_fn=compat.partial(self._FakeTcpPing,
(target, port), False))
except errors.HypervisorError, err:
target = constants.IP4_ADDRESS_LOCALHOST
port = 22364
+ hvparams = {constants.HV_XEN_CMD: self.CMD}
+
for live in [False, True]:
for fail in [False, True]:
ping_fn = \
if fail:
try:
hv._MigrateInstance(clustername, instname, target, port, live,
- _ping_fn=ping_fn)
+ hvparams, _ping_fn=ping_fn)
except errors.HypervisorError, err:
self.assertTrue(str(err).startswith("Failed to migrate instance"))
else:
self.fail("Exception was not raised")
else:
hv._MigrateInstance(clustername, instname, target, port, live,
- _ping_fn=ping_fn)
+ hvparams, _ping_fn=ping_fn)
if self.CMD == constants.XEN_CMD_XM:
expected_pings = 1
self.assertEqual(ping_fn.Count(), expected_pings)
+ def _GetNodeInfoCmd(self, fail, cmd):
+ if cmd == [self.CMD, "info"]:
+ if fail:
+ return self._FailingCommand(cmd)
+ else:
+ output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
+ elif cmd == [self.CMD, "list"]:
+ if fail:
+ self.fail("'xm list' shouldn't be called when 'xm info' failed")
+ else:
+ output = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
+ else:
+ self.fail("Unhandled command: %s" % (cmd, ))
+
+ return self._SuccessCommand(output, cmd)
+
+ def testGetNodeInfo(self):
+ run_cmd = compat.partial(self._GetNodeInfoCmd, False)
+ hv = self._GetHv(run_cmd=run_cmd)
+ result = hv.GetNodeInfo()
+
+ self.assertEqual(result["hv_version"], (4, 0))
+ self.assertEqual(result["memory_free"], 8004)
+
+ def testGetNodeInfoFailing(self):
+ run_cmd = compat.partial(self._GetNodeInfoCmd, True)
+ hv = self._GetHv(run_cmd=run_cmd)
+ self.assertTrue(hv.GetNodeInfo() is None)
+
def _MakeTestClass(cls, cmd):
"""Makes a class for testing.
projects / ganeti-local / blobdiff