4 # Copyright (C) 2011, 2013 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 ganeti.hypervisor.hv_lxc"""
24 import string # pylint: disable=W0402
31 from ganeti import constants
32 from ganeti import objects
33 from ganeti import pathutils
34 from ganeti import hypervisor
35 from ganeti import utils
36 from ganeti import errors
37 from ganeti import compat
39 from ganeti.hypervisor import hv_xen
44 # Map from hypervisor class to hypervisor name
45 HVCLASS_TO_HVNAME = utils.InvertDict(hypervisor._HYPERVISOR_MAP)
48 class TestConsole(unittest.TestCase):
50 for cls in [hv_xen.XenPvmHypervisor, hv_xen.XenHvmHypervisor]:
51 instance = objects.Instance(name="xen.example.com",
52 primary_node="node24828")
53 cons = cls.GetInstanceConsole(instance, {}, {})
54 self.assertTrue(cons.Validate())
55 self.assertEqual(cons.kind, constants.CONS_SSH)
56 self.assertEqual(cons.host, instance.primary_node)
57 self.assertEqual(cons.command[-1], instance.name)
60 class TestCreateConfigCpus(unittest.TestCase):
62 for cpu_mask in [None, ""]:
63 self.assertEqual(hv_xen._CreateConfigCpus(cpu_mask),
67 self.assertEqual(hv_xen._CreateConfigCpus(constants.CPU_PINNING_ALL),
71 self.assertEqual(hv_xen._CreateConfigCpus("9"), "cpu = \"9\"")
73 def testMultiple(self):
74 self.assertEqual(hv_xen._CreateConfigCpus("0-2,4,5-5:3:all"),
75 ("cpus = [ \"0,1,2,4,5\", \"3\", \"%s\" ]" %
76 constants.CPU_PINNING_ALL_XEN))
79 class TestParseXmList(testutils.GanetiTestCase):
81 data = testutils.ReadTestData("xen-xm-list-4.0.1-dom0-only.txt")
84 self.assertEqual(hv_xen._ParseXmList(data.splitlines(), False), [])
87 result = hv_xen._ParseXmList(data.splitlines(), True)
88 self.assertEqual(len(result), 1)
89 self.assertEqual(len(result[0]), 6)
92 self.assertEqual(result[0][0], hv_xen._DOM0_NAME)
95 self.assertEqual(result[0][1], 0)
98 self.assertEqual(result[0][2], 1023)
101 self.assertEqual(result[0][3], 1)
104 self.assertEqual(result[0][4], "r-----")
107 self.assertAlmostEqual(result[0][5], 121152.6)
109 def testWrongLineFormat(self):
111 ["three fields only"],
112 ["name InvalidID 128 1 r----- 12345"],
117 hv_xen._ParseXmList(["Header would be here"] + lines, False)
118 except errors.HypervisorError, err:
119 self.assertTrue("Can't parse output of xm list" in str(err))
121 self.fail("Exception was not raised")
124 class TestGetXmList(testutils.GanetiTestCase):
126 return utils.RunResult(constants.EXIT_FAILURE, None,
127 "stdout", "stderr", None,
128 NotImplemented, NotImplemented)
130 def testTimeout(self):
131 fn = testutils.CallCounter(self._Fail)
133 hv_xen._GetXmList(fn, False, _timeout=0.1)
134 except errors.HypervisorError, err:
135 self.assertTrue("timeout exceeded" in str(err))
137 self.fail("Exception was not raised")
139 self.assertTrue(fn.Count() < 10,
140 msg="'xm list' was called too many times")
142 def _Success(self, stdout):
143 return utils.RunResult(constants.EXIT_SUCCESS, None, stdout, "", None,
144 NotImplemented, NotImplemented)
146 def testSuccess(self):
147 data = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
149 fn = testutils.CallCounter(compat.partial(self._Success, data))
151 result = hv_xen._GetXmList(fn, True, _timeout=0.1)
153 self.assertEqual(len(result), 4)
155 self.assertEqual(map(compat.fst, result), [
157 "server01.example.com",
158 "web3106215069.example.com",
159 "testinstance.example.com",
162 self.assertEqual(fn.Count(), 1)
165 class TestParseNodeInfo(testutils.GanetiTestCase):
167 self.assertEqual(hv_xen._ParseNodeInfo(""), {})
169 def testUnknownInput(self):
172 "something else goes",
175 self.assertEqual(hv_xen._ParseNodeInfo(data), {})
177 def testBasicInfo(self):
178 data = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
179 result = hv_xen._ParseNodeInfo(data)
180 self.assertEqual(result, {
184 "hv_version": (4, 0),
186 "memory_total": 16378,
190 class TestMergeInstanceInfo(testutils.GanetiTestCase):
192 self.assertEqual(hv_xen._MergeInstanceInfo({}, lambda _: []), {})
194 def _FakeXmList(self, include_node):
195 self.assertTrue(include_node)
197 (hv_xen._DOM0_NAME, NotImplemented, 4096, 7, NotImplemented,
199 ("inst1.example.com", NotImplemented, 2048, 4, NotImplemented,
203 def testMissingNodeInfo(self):
204 result = hv_xen._MergeInstanceInfo({}, self._FakeXmList)
205 self.assertEqual(result, {
210 def testWithNodeInfo(self):
211 info = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
212 result = hv_xen._GetNodeInfo(info, self._FakeXmList)
213 self.assertEqual(result, {
218 "hv_version": (4, 0),
222 "memory_total": 16378,
226 class TestGetConfigFileDiskData(unittest.TestCase):
227 def testLetterCount(self):
228 self.assertEqual(len(hv_xen._DISK_LETTERS), 26)
230 def testNoDisks(self):
231 self.assertEqual(hv_xen._GetConfigFileDiskData([], "hd"), [])
233 def testManyDisks(self):
234 for offset in [0, 1, 10]:
235 disks = [(objects.Disk(dev_type=constants.LD_LV), "/tmp/disk/%s" % idx)
236 for idx in range(len(hv_xen._DISK_LETTERS) + offset)]
239 result = hv_xen._GetConfigFileDiskData(disks, "hd")
240 self.assertEqual(result, [
241 "'phy:/tmp/disk/%s,hd%s,r'" % (idx, string.ascii_lowercase[idx])
242 for idx in range(len(hv_xen._DISK_LETTERS) + offset)
246 hv_xen._GetConfigFileDiskData(disks, "hd")
247 except errors.HypervisorError, err:
248 self.assertEqual(str(err), "Too many disks")
250 self.fail("Exception was not raised")
252 def testTwoLvDisksWithMode(self):
254 (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDWR),
256 (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDONLY),
260 result = hv_xen._GetConfigFileDiskData(disks, "hd")
261 self.assertEqual(result, [
262 "'phy:/tmp/diskFirst,hda,w'",
263 "'phy:/tmp/diskLast,hdb,r'",
266 def testFileDisks(self):
268 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
269 physical_id=[constants.FD_LOOP]),
271 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDONLY,
272 physical_id=[constants.FD_BLKTAP]),
274 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
275 physical_id=[constants.FD_LOOP]),
277 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
278 physical_id=[constants.FD_BLKTAP]),
282 result = hv_xen._GetConfigFileDiskData(disks, "sd")
283 self.assertEqual(result, [
284 "'file:/tmp/diskFirst,sda,w'",
285 "'tap:aio:/tmp/diskTwo,sdb,r'",
286 "'file:/tmp/diskThree,sdc,w'",
287 "'tap:aio:/tmp/diskLast,sdd,w'",
290 def testInvalidFileDisk(self):
292 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
293 physical_id=["#unknown#"]),
297 self.assertRaises(KeyError, hv_xen._GetConfigFileDiskData, disks, "sd")
300 class TestXenHypervisorUnknownCommand(unittest.TestCase):
302 cmd = "#unknown command#"
303 self.assertFalse(cmd in constants.KNOWN_XEN_COMMANDS)
304 hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
305 _run_cmd_fn=NotImplemented,
307 self.assertRaises(errors.ProgrammerError, hv._RunXen, [])
310 class TestXenHypervisorWriteConfigFile(unittest.TestCase):
312 self.tmpdir = tempfile.mkdtemp()
315 shutil.rmtree(self.tmpdir)
317 def testWriteError(self):
318 cfgdir = utils.PathJoin(self.tmpdir, "foobar")
320 hv = hv_xen.XenHypervisor(_cfgdir=cfgdir,
321 _run_cmd_fn=NotImplemented,
324 self.assertFalse(os.path.exists(cfgdir))
327 hv._WriteConfigFile("name", "data")
328 except errors.HypervisorError, err:
329 self.assertTrue(str(err).startswith("Cannot write Xen instance"))
331 self.fail("Exception was not raised")
334 class _TestXenHypervisor(object):
335 TARGET = NotImplemented
337 HVNAME = NotImplemented
340 super(_TestXenHypervisor, self).setUp()
342 self.tmpdir = tempfile.mkdtemp()
344 self.vncpw = "".join(random.sample(string.ascii_letters, 10))
346 self.vncpw_path = utils.PathJoin(self.tmpdir, "vncpw")
347 utils.WriteFile(self.vncpw_path, data=self.vncpw)
350 super(_TestXenHypervisor, self).tearDown()
352 shutil.rmtree(self.tmpdir)
354 def _GetHv(self, run_cmd=NotImplemented):
355 return self.TARGET(_cfgdir=self.tmpdir, _run_cmd_fn=run_cmd, _cmd=self.CMD)
357 def _SuccessCommand(self, stdout, cmd):
358 self.assertEqual(cmd[0], self.CMD)
360 return utils.RunResult(constants.EXIT_SUCCESS, None, stdout, "", None,
361 NotImplemented, NotImplemented)
363 def _FailingCommand(self, cmd):
364 self.assertEqual(cmd[0], self.CMD)
366 return utils.RunResult(constants.EXIT_FAILURE, None,
367 "", "This command failed", None,
368 NotImplemented, NotImplemented)
370 def _FakeTcpPing(self, expected, result, target, port, **kwargs):
371 self.assertEqual((target, port), expected)
374 def testReadingNonExistentConfigFile(self):
378 hv._ReadConfigFile("inst15780.example.com")
379 except errors.HypervisorError, err:
380 self.assertTrue(str(err).startswith("Failed to load Xen config file:"))
382 self.fail("Exception was not raised")
384 def testRemovingAutoConfigFile(self):
385 name = "inst8206.example.com"
386 cfgfile = utils.PathJoin(self.tmpdir, name)
387 autodir = utils.PathJoin(self.tmpdir, "auto")
388 autocfgfile = utils.PathJoin(autodir, name)
392 utils.WriteFile(autocfgfile, data="")
396 self.assertTrue(os.path.isfile(autocfgfile))
397 hv._WriteConfigFile(name, "content")
398 self.assertFalse(os.path.exists(autocfgfile))
399 self.assertEqual(utils.ReadFile(cfgfile), "content")
401 def _XenList(self, cmd):
402 self.assertEqual(cmd, [self.CMD, "list"])
404 # TODO: Use actual data from "xl" command
405 output = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
407 return self._SuccessCommand(output, cmd)
409 def testGetInstanceInfo(self):
410 hv = self._GetHv(run_cmd=self._XenList)
412 (name, instid, memory, vcpus, state, runtime) = \
413 hv.GetInstanceInfo("server01.example.com")
415 self.assertEqual(name, "server01.example.com")
416 self.assertEqual(instid, 1)
417 self.assertEqual(memory, 1024)
418 self.assertEqual(vcpus, 1)
419 self.assertEqual(state, "-b----")
420 self.assertAlmostEqual(runtime, 167643.2)
422 def testGetInstanceInfoDom0(self):
423 hv = self._GetHv(run_cmd=self._XenList)
425 # TODO: Not sure if this is actually used anywhere (can't find it), but the
426 # code supports querying for Dom0
427 (name, instid, memory, vcpus, state, runtime) = \
428 hv.GetInstanceInfo(hv_xen._DOM0_NAME)
430 self.assertEqual(name, "Domain-0")
431 self.assertEqual(instid, 0)
432 self.assertEqual(memory, 1023)
433 self.assertEqual(vcpus, 1)
434 self.assertEqual(state, "r-----")
435 self.assertAlmostEqual(runtime, 154706.1)
437 def testGetInstanceInfoUnknown(self):
438 hv = self._GetHv(run_cmd=self._XenList)
440 result = hv.GetInstanceInfo("unknown.example.com")
441 self.assertTrue(result is None)
443 def testGetAllInstancesInfo(self):
444 hv = self._GetHv(run_cmd=self._XenList)
446 result = hv.GetAllInstancesInfo()
448 self.assertEqual(map(compat.fst, result), [
449 "server01.example.com",
450 "web3106215069.example.com",
451 "testinstance.example.com",
454 def testListInstances(self):
455 hv = self._GetHv(run_cmd=self._XenList)
457 self.assertEqual(hv.ListInstances(), [
458 "server01.example.com",
459 "web3106215069.example.com",
460 "testinstance.example.com",
463 def testVerify(self):
464 output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
465 hv = self._GetHv(run_cmd=compat.partial(self._SuccessCommand,
467 self.assertTrue(hv.Verify() is None)
469 def testVerifyFailing(self):
470 hv = self._GetHv(run_cmd=self._FailingCommand)
471 self.assertTrue("failed:" in hv.Verify())
473 def _StartInstanceCommand(self, inst, paused, failcreate, cmd):
474 if cmd == [self.CMD, "info"]:
475 output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
476 elif cmd == [self.CMD, "list"]:
477 output = testutils.ReadTestData("xen-xm-list-4.0.1-dom0-only.txt")
478 elif cmd[:2] == [self.CMD, "create"]:
480 cfgfile = utils.PathJoin(self.tmpdir, inst.name)
483 self.assertEqual(args, ["-p", cfgfile])
485 self.assertEqual(args, [cfgfile])
488 return self._FailingCommand(cmd)
492 self.fail("Unhandled command: %s" % (cmd, ))
494 return self._SuccessCommand(output, cmd)
496 def _MakeInstance(self):
497 # Copy default parameters
498 bep = objects.FillDict(constants.BEC_DEFAULTS, {})
499 hvp = objects.FillDict(constants.HVC_DEFAULTS[self.HVNAME], {})
501 # Override default VNC password file path
502 if constants.HV_VNC_PASSWORD_FILE in hvp:
503 hvp[constants.HV_VNC_PASSWORD_FILE] = self.vncpw_path
506 (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDWR),
507 utils.PathJoin(self.tmpdir, "disk0")),
508 (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDONLY),
509 utils.PathJoin(self.tmpdir, "disk1")),
512 inst = objects.Instance(name="server01.example.com",
513 hvparams=hvp, beparams=bep,
514 osparams={}, nics=[], os="deb1",
515 disks=map(compat.fst, disks))
520 def testStartInstance(self):
521 (inst, disks) = self._MakeInstance()
522 pathutils.LOG_XEN_DIR = self.tmpdir
524 for failcreate in [False, True]:
525 for paused in [False, True]:
526 run_cmd = compat.partial(self._StartInstanceCommand,
527 inst, paused, failcreate)
529 hv = self._GetHv(run_cmd=run_cmd)
531 # Ensure instance is not listed
532 self.assertTrue(inst.name not in hv.ListInstances())
534 # Remove configuration
535 cfgfile = utils.PathJoin(self.tmpdir, inst.name)
536 utils.RemoveFile(cfgfile)
539 self.assertRaises(errors.HypervisorError, hv.StartInstance,
541 # Check whether a stale config file is left behind
542 self.assertFalse(os.path.exists(cfgfile))
544 hv.StartInstance(inst, disks, paused)
545 # Check if configuration was updated
546 lines = utils.ReadFile(cfgfile).splitlines()
548 if constants.HV_VNC_PASSWORD_FILE in inst.hvparams:
549 self.assertTrue(("vncpasswd = '%s'" % self.vncpw) in lines)
551 extra = inst.hvparams[constants.HV_KERNEL_ARGS]
552 self.assertTrue(("extra = '%s'" % extra) in lines)
554 def _StopInstanceCommand(self, instance_name, force, fail, cmd):
555 if ((force and cmd[:2] == [self.CMD, "destroy"]) or
556 (not force and cmd[:2] == [self.CMD, "shutdown"])):
557 self.assertEqual(cmd[2:], [instance_name])
560 self.fail("Unhandled command: %s" % (cmd, ))
563 # Simulate a failing command
564 return self._FailingCommand(cmd)
566 return self._SuccessCommand(output, cmd)
568 def testStopInstance(self):
569 name = "inst4284.example.com"
570 cfgfile = utils.PathJoin(self.tmpdir, name)
571 cfgdata = "config file content\n"
573 for force in [False, True]:
574 for fail in [False, True]:
575 utils.WriteFile(cfgfile, data=cfgdata)
577 run_cmd = compat.partial(self._StopInstanceCommand, name, force, fail)
579 hv = self._GetHv(run_cmd=run_cmd)
581 self.assertTrue(os.path.isfile(cfgfile))
585 hv._StopInstance(name, force)
586 except errors.HypervisorError, err:
587 self.assertTrue(str(err).startswith("Failed to stop instance"))
589 self.fail("Exception was not raised")
590 self.assertEqual(utils.ReadFile(cfgfile), cfgdata,
591 msg=("Configuration was removed when stopping"
594 hv._StopInstance(name, force)
595 self.assertFalse(os.path.exists(cfgfile))
597 def _MigrateNonRunningInstCmd(self, cmd):
598 if cmd == [self.CMD, "list"]:
599 output = testutils.ReadTestData("xen-xm-list-4.0.1-dom0-only.txt")
601 self.fail("Unhandled command: %s" % (cmd, ))
603 return self._SuccessCommand(output, cmd)
605 def testMigrateInstanceNotRunning(self):
606 name = "nonexistinginstance.example.com"
607 target = constants.IP4_ADDRESS_LOCALHOST
610 hv = self._GetHv(run_cmd=self._MigrateNonRunningInstCmd)
612 for live in [False, True]:
614 hv._MigrateInstance(NotImplemented, name, target, port, live,
615 _ping_fn=NotImplemented)
616 except errors.HypervisorError, err:
617 self.assertEqual(str(err), "Instance not running, cannot migrate")
619 self.fail("Exception was not raised")
621 def _MigrateInstTargetUnreachCmd(self, cmd):
622 if cmd == [self.CMD, "list"]:
623 output = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
625 self.fail("Unhandled command: %s" % (cmd, ))
627 return self._SuccessCommand(output, cmd)
629 def testMigrateTargetUnreachable(self):
630 name = "server01.example.com"
631 target = constants.IP4_ADDRESS_LOCALHOST
634 hv = self._GetHv(run_cmd=self._MigrateInstTargetUnreachCmd)
636 for live in [False, True]:
637 if self.CMD == constants.XEN_CMD_XL:
638 # TODO: Detect unreachable targets
642 hv._MigrateInstance(NotImplemented, name, target, port, live,
643 _ping_fn=compat.partial(self._FakeTcpPing,
644 (target, port), False))
645 except errors.HypervisorError, err:
646 wanted = "Remote host %s not" % target
647 self.assertTrue(str(err).startswith(wanted))
649 self.fail("Exception was not raised")
651 def _MigrateInstanceCmd(self, cluster_name, instance_name, target, port,
653 if cmd == [self.CMD, "list"]:
654 output = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
655 elif cmd[:2] == [self.CMD, "migrate"]:
656 if self.CMD == constants.XEN_CMD_XM:
657 args = ["-p", str(port)]
662 elif self.CMD == constants.XEN_CMD_XL:
664 "-s", constants.XL_SSH_CMD % cluster_name,
665 "-C", utils.PathJoin(self.tmpdir, instance_name),
669 self.fail("Unknown Xen command '%s'" % self.CMD)
671 args.extend([instance_name, target])
672 self.assertEqual(cmd[2:], args)
675 return self._FailingCommand(cmd)
679 self.fail("Unhandled command: %s" % (cmd, ))
681 return self._SuccessCommand(output, cmd)
683 def testMigrateInstance(self):
684 clustername = "cluster.example.com"
685 instname = "server01.example.com"
686 target = constants.IP4_ADDRESS_LOCALHOST
689 for live in [False, True]:
690 for fail in [False, True]:
692 testutils.CallCounter(compat.partial(self._FakeTcpPing,
693 (target, port), True))
696 compat.partial(self._MigrateInstanceCmd,
697 clustername, instname, target, port, live,
700 hv = self._GetHv(run_cmd=run_cmd)
704 hv._MigrateInstance(clustername, instname, target, port, live,
706 except errors.HypervisorError, err:
707 self.assertTrue(str(err).startswith("Failed to migrate instance"))
709 self.fail("Exception was not raised")
711 hv._MigrateInstance(clustername, instname, target, port, live,
714 if self.CMD == constants.XEN_CMD_XM:
719 self.assertEqual(ping_fn.Count(), expected_pings)
721 def _GetNodeInfoCmd(self, fail, cmd):
722 if cmd == [self.CMD, "info"]:
724 return self._FailingCommand(cmd)
726 output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
727 elif cmd == [self.CMD, "list"]:
729 self.fail("'xm list' shouldn't be called when 'xm info' failed")
731 output = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
733 self.fail("Unhandled command: %s" % (cmd, ))
735 return self._SuccessCommand(output, cmd)
737 def testGetNodeInfo(self):
738 run_cmd = compat.partial(self._GetNodeInfoCmd, False)
739 hv = self._GetHv(run_cmd=run_cmd)
740 result = hv.GetNodeInfo()
742 self.assertEqual(result["hv_version"], (4, 0))
743 self.assertEqual(result["memory_free"], 8004)
745 def testGetNodeInfoFailing(self):
746 run_cmd = compat.partial(self._GetNodeInfoCmd, True)
747 hv = self._GetHv(run_cmd=run_cmd)
748 self.assertTrue(hv.GetNodeInfo() is None)
751 def _MakeTestClass(cls, cmd):
752 """Makes a class for testing.
754 The returned class has structure as shown in the following pseudo code:
756 class Test{cls.__name__}{cmd}(_TestXenHypervisor, unittest.TestCase):
759 HVNAME = {Hypervisor name retrieved using class}
762 @param cls: Hypervisor class to be tested
764 @param cmd: Hypervisor command
766 @return: Class name and class object (not instance)
769 name = "Test%sCmd%s" % (cls.__name__, cmd.title())
770 bases = (_TestXenHypervisor, unittest.TestCase)
771 hvname = HVCLASS_TO_HVNAME[cls]
773 return (name, type(name, bases, dict(TARGET=cls, CMD=cmd, HVNAME=hvname)))
776 # Create test classes programmatically instead of manually to reduce the risk
777 # of forgetting some combinations
778 for cls in [hv_xen.XenPvmHypervisor, hv_xen.XenHvmHypervisor]:
779 for cmd in constants.KNOWN_XEN_COMMANDS:
780 (name, testcls) = _MakeTestClass(cls, cmd)
782 assert name not in locals()
784 locals()[name] = testcls
787 if __name__ == "__main__":
788 testutils.GanetiTestProgram()