hv_xen: Test verifying hypervisor
[ganeti-github.git] / test / py / ganeti.hypervisor.hv_xen_unittest.py
1 #!/usr/bin/python
2 #
3
4 # Copyright (C) 2011, 2013 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 ganeti.hypervisor.hv_lxc"""
23
24 import string # pylint: disable=W0402
25 import unittest
26 import tempfile
27 import shutil
28 import random
29 import os
30
31 from ganeti import constants
32 from ganeti import objects
33 from ganeti import hypervisor
34 from ganeti import utils
35 from ganeti import errors
36 from ganeti import compat
37
38 from ganeti.hypervisor import hv_xen
39
40 import testutils
41
42
43 # Map from hypervisor class to hypervisor name
44 HVCLASS_TO_HVNAME = utils.InvertDict(hypervisor._HYPERVISOR_MAP)
45
46
47 class TestConsole(unittest.TestCase):
48 def test(self):
49 for cls in [hv_xen.XenPvmHypervisor, hv_xen.XenHvmHypervisor]:
50 instance = objects.Instance(name="xen.example.com",
51 primary_node="node24828")
52 cons = cls.GetInstanceConsole(instance, {}, {})
53 self.assertTrue(cons.Validate())
54 self.assertEqual(cons.kind, constants.CONS_SSH)
55 self.assertEqual(cons.host, instance.primary_node)
56 self.assertEqual(cons.command[-1], instance.name)
57
58
59 class TestCreateConfigCpus(unittest.TestCase):
60 def testEmpty(self):
61 for cpu_mask in [None, ""]:
62 self.assertEqual(hv_xen._CreateConfigCpus(cpu_mask),
63 "cpus = [ ]")
64
65 def testAll(self):
66 self.assertEqual(hv_xen._CreateConfigCpus(constants.CPU_PINNING_ALL),
67 None)
68
69 def testOne(self):
70 self.assertEqual(hv_xen._CreateConfigCpus("9"), "cpu = \"9\"")
71
72 def testMultiple(self):
73 self.assertEqual(hv_xen._CreateConfigCpus("0-2,4,5-5:3:all"),
74 ("cpus = [ \"0,1,2,4,5\", \"3\", \"%s\" ]" %
75 constants.CPU_PINNING_ALL_XEN))
76
77
78 class TestParseXmList(testutils.GanetiTestCase):
79 def test(self):
80 data = testutils.ReadTestData("xen-xm-list-4.0.1-dom0-only.txt")
81
82 # Exclude node
83 self.assertEqual(hv_xen._ParseXmList(data.splitlines(), False), [])
84
85 # Include node
86 result = hv_xen._ParseXmList(data.splitlines(), True)
87 self.assertEqual(len(result), 1)
88 self.assertEqual(len(result[0]), 6)
89
90 # Name
91 self.assertEqual(result[0][0], hv_xen._DOM0_NAME)
92
93 # ID
94 self.assertEqual(result[0][1], 0)
95
96 # Memory
97 self.assertEqual(result[0][2], 1023)
98
99 # VCPUs
100 self.assertEqual(result[0][3], 1)
101
102 # State
103 self.assertEqual(result[0][4], "r-----")
104
105 # Time
106 self.assertAlmostEqual(result[0][5], 121152.6)
107
108 def testWrongLineFormat(self):
109 tests = [
110 ["three fields only"],
111 ["name InvalidID 128 1 r----- 12345"],
112 ]
113
114 for lines in tests:
115 try:
116 hv_xen._ParseXmList(["Header would be here"] + lines, False)
117 except errors.HypervisorError, err:
118 self.assertTrue("Can't parse output of xm list" in str(err))
119 else:
120 self.fail("Exception was not raised")
121
122
123 class TestGetXmList(testutils.GanetiTestCase):
124 def _Fail(self):
125 return utils.RunResult(constants.EXIT_FAILURE, None,
126 "stdout", "stderr", None,
127 NotImplemented, NotImplemented)
128
129 def testTimeout(self):
130 fn = testutils.CallCounter(self._Fail)
131 try:
132 hv_xen._GetXmList(fn, False, _timeout=0.1)
133 except errors.HypervisorError, err:
134 self.assertTrue("timeout exceeded" in str(err))
135 else:
136 self.fail("Exception was not raised")
137
138 self.assertTrue(fn.Count() < 10,
139 msg="'xm list' was called too many times")
140
141 def _Success(self, stdout):
142 return utils.RunResult(constants.EXIT_SUCCESS, None, stdout, "", None,
143 NotImplemented, NotImplemented)
144
145 def testSuccess(self):
146 data = testutils.ReadTestData("xen-xm-list-4.0.1-four-instances.txt")
147
148 fn = testutils.CallCounter(compat.partial(self._Success, data))
149
150 result = hv_xen._GetXmList(fn, True, _timeout=0.1)
151
152 self.assertEqual(len(result), 4)
153
154 self.assertEqual(map(compat.fst, result), [
155 "Domain-0",
156 "server01.example.com",
157 "web3106215069.example.com",
158 "testinstance.example.com",
159 ])
160
161 self.assertEqual(fn.Count(), 1)
162
163
164 class TestParseNodeInfo(testutils.GanetiTestCase):
165 def testEmpty(self):
166 self.assertEqual(hv_xen._ParseNodeInfo(""), {})
167
168 def testUnknownInput(self):
169 data = "\n".join([
170 "foo bar",
171 "something else goes",
172 "here",
173 ])
174 self.assertEqual(hv_xen._ParseNodeInfo(data), {})
175
176 def testBasicInfo(self):
177 data = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
178 result = hv_xen._ParseNodeInfo(data)
179 self.assertEqual(result, {
180 "cpu_nodes": 1,
181 "cpu_sockets": 2,
182 "cpu_total": 4,
183 "hv_version": (4, 0),
184 "memory_free": 8004,
185 "memory_total": 16378,
186 })
187
188
189 class TestMergeInstanceInfo(testutils.GanetiTestCase):
190 def testEmpty(self):
191 self.assertEqual(hv_xen._MergeInstanceInfo({}, lambda _: []), {})
192
193 def _FakeXmList(self, include_node):
194 self.assertTrue(include_node)
195 return [
196 (hv_xen._DOM0_NAME, NotImplemented, 4096, 7, NotImplemented,
197 NotImplemented),
198 ("inst1.example.com", NotImplemented, 2048, 4, NotImplemented,
199 NotImplemented),
200 ]
201
202 def testMissingNodeInfo(self):
203 result = hv_xen._MergeInstanceInfo({}, self._FakeXmList)
204 self.assertEqual(result, {
205 "memory_dom0": 4096,
206 "dom0_cpus": 7,
207 })
208
209 def testWithNodeInfo(self):
210 info = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
211 result = hv_xen._GetNodeInfo(info, self._FakeXmList)
212 self.assertEqual(result, {
213 "cpu_nodes": 1,
214 "cpu_sockets": 2,
215 "cpu_total": 4,
216 "dom0_cpus": 7,
217 "hv_version": (4, 0),
218 "memory_dom0": 4096,
219 "memory_free": 8004,
220 "memory_hv": 2230,
221 "memory_total": 16378,
222 })
223
224
225 class TestGetConfigFileDiskData(unittest.TestCase):
226 def testLetterCount(self):
227 self.assertEqual(len(hv_xen._DISK_LETTERS), 26)
228
229 def testNoDisks(self):
230 self.assertEqual(hv_xen._GetConfigFileDiskData([], "hd"), [])
231
232 def testManyDisks(self):
233 for offset in [0, 1, 10]:
234 disks = [(objects.Disk(dev_type=constants.LD_LV), "/tmp/disk/%s" % idx)
235 for idx in range(len(hv_xen._DISK_LETTERS) + offset)]
236
237 if offset == 0:
238 result = hv_xen._GetConfigFileDiskData(disks, "hd")
239 self.assertEqual(result, [
240 "'phy:/tmp/disk/%s,hd%s,r'" % (idx, string.ascii_lowercase[idx])
241 for idx in range(len(hv_xen._DISK_LETTERS) + offset)
242 ])
243 else:
244 try:
245 hv_xen._GetConfigFileDiskData(disks, "hd")
246 except errors.HypervisorError, err:
247 self.assertEqual(str(err), "Too many disks")
248 else:
249 self.fail("Exception was not raised")
250
251 def testTwoLvDisksWithMode(self):
252 disks = [
253 (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDWR),
254 "/tmp/diskFirst"),
255 (objects.Disk(dev_type=constants.LD_LV, mode=constants.DISK_RDONLY),
256 "/tmp/diskLast"),
257 ]
258
259 result = hv_xen._GetConfigFileDiskData(disks, "hd")
260 self.assertEqual(result, [
261 "'phy:/tmp/diskFirst,hda,w'",
262 "'phy:/tmp/diskLast,hdb,r'",
263 ])
264
265 def testFileDisks(self):
266 disks = [
267 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
268 physical_id=[constants.FD_LOOP]),
269 "/tmp/diskFirst"),
270 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDONLY,
271 physical_id=[constants.FD_BLKTAP]),
272 "/tmp/diskTwo"),
273 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
274 physical_id=[constants.FD_LOOP]),
275 "/tmp/diskThree"),
276 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
277 physical_id=[constants.FD_BLKTAP]),
278 "/tmp/diskLast"),
279 ]
280
281 result = hv_xen._GetConfigFileDiskData(disks, "sd")
282 self.assertEqual(result, [
283 "'file:/tmp/diskFirst,sda,w'",
284 "'tap:aio:/tmp/diskTwo,sdb,r'",
285 "'file:/tmp/diskThree,sdc,w'",
286 "'tap:aio:/tmp/diskLast,sdd,w'",
287 ])
288
289 def testInvalidFileDisk(self):
290 disks = [
291 (objects.Disk(dev_type=constants.LD_FILE, mode=constants.DISK_RDWR,
292 physical_id=["#unknown#"]),
293 "/tmp/diskinvalid"),
294 ]
295
296 self.assertRaises(KeyError, hv_xen._GetConfigFileDiskData, disks, "sd")
297
298
299 class TestXenHypervisorUnknownCommand(unittest.TestCase):
300 def test(self):
301 cmd = "#unknown command#"
302 self.assertFalse(cmd in constants.KNOWN_XEN_COMMANDS)
303 hv = hv_xen.XenHypervisor(_cfgdir=NotImplemented,
304 _run_cmd_fn=NotImplemented,
305 _cmd=cmd)
306 self.assertRaises(errors.ProgrammerError, hv._RunXen, [])
307
308
309 class TestXenHypervisorWriteConfigFile(unittest.TestCase):
310 def setUp(self):
311 self.tmpdir = tempfile.mkdtemp()
312
313 def tearDown(self):
314 shutil.rmtree(self.tmpdir)
315
316 def testWriteError(self):
317 cfgdir = utils.PathJoin(self.tmpdir, "foobar")
318
319 hv = hv_xen.XenHypervisor(_cfgdir=cfgdir,
320 _run_cmd_fn=NotImplemented,
321 _cmd=NotImplemented)
322
323 self.assertFalse(os.path.exists(cfgdir))
324
325 try:
326 hv._WriteConfigFile("name", "data")
327 except errors.HypervisorError, err:
328 self.assertTrue(str(err).startswith("Cannot write Xen instance"))
329 else:
330 self.fail("Exception was not raised")
331
332
333 class _TestXenHypervisor(object):
334 TARGET = NotImplemented
335 CMD = NotImplemented
336 HVNAME = NotImplemented
337
338 def setUp(self):
339 super(_TestXenHypervisor, self).setUp()
340
341 self.tmpdir = tempfile.mkdtemp()
342
343 self.vncpw = "".join(random.sample(string.ascii_letters, 10))
344
345 self.vncpw_path = utils.PathJoin(self.tmpdir, "vncpw")
346 utils.WriteFile(self.vncpw_path, data=self.vncpw)
347
348 def tearDown(self):
349 super(_TestXenHypervisor, self).tearDown()
350
351 shutil.rmtree(self.tmpdir)
352
353 def _GetHv(self, run_cmd=NotImplemented):
354 return self.TARGET(_cfgdir=self.tmpdir, _run_cmd_fn=run_cmd, _cmd=self.CMD)
355
356 def _SuccessCommand(self, stdout, cmd):
357 self.assertEqual(cmd[0], self.CMD)
358
359 return utils.RunResult(constants.EXIT_SUCCESS, None, stdout, "", None,
360 NotImplemented, NotImplemented)
361
362 def _FailingCommand(self, cmd):
363 self.assertEqual(cmd[0], self.CMD)
364
365 return utils.RunResult(constants.EXIT_FAILURE, None,
366 "", "This command failed", None,
367 NotImplemented, NotImplemented)
368
369 def testRemovingAutoConfigFile(self):
370 name = "inst8206.example.com"
371 cfgfile = utils.PathJoin(self.tmpdir, name)
372 autodir = utils.PathJoin(self.tmpdir, "auto")
373 autocfgfile = utils.PathJoin(autodir, name)
374
375 os.mkdir(autodir)
376
377 utils.WriteFile(autocfgfile, data="")
378
379 hv = self._GetHv()
380
381 self.assertTrue(os.path.isfile(autocfgfile))
382 hv._WriteConfigFile(name, "content")
383 self.assertFalse(os.path.exists(autocfgfile))
384 self.assertEqual(utils.ReadFile(cfgfile), "content")
385
386 def testVerify(self):
387 output = testutils.ReadTestData("xen-xm-info-4.0.1.txt")
388 hv = self._GetHv(run_cmd=compat.partial(self._SuccessCommand,
389 output))
390 self.assertTrue(hv.Verify() is None)
391
392 def testVerifyFailing(self):
393 hv = self._GetHv(run_cmd=self._FailingCommand)
394 self.assertTrue("failed:" in hv.Verify())
395
396
397 def _MakeTestClass(cls, cmd):
398 """Makes a class for testing.
399
400 The returned class has structure as shown in the following pseudo code:
401
402 class Test{cls.__name__}{cmd}(_TestXenHypervisor, unittest.TestCase):
403 TARGET = {cls}
404 CMD = {cmd}
405 HVNAME = {Hypervisor name retrieved using class}
406
407 @type cls: class
408 @param cls: Hypervisor class to be tested
409 @type cmd: string
410 @param cmd: Hypervisor command
411 @rtype: tuple
412 @return: Class name and class object (not instance)
413
414 """
415 name = "Test%sCmd%s" % (cls.__name__, cmd.title())
416 bases = (_TestXenHypervisor, unittest.TestCase)
417 hvname = HVCLASS_TO_HVNAME[cls]
418
419 return (name, type(name, bases, dict(TARGET=cls, CMD=cmd, HVNAME=hvname)))
420
421
422 # Create test classes programmatically instead of manually to reduce the risk
423 # of forgetting some combinations
424 for cls in [hv_xen.XenPvmHypervisor, hv_xen.XenHvmHypervisor]:
425 for cmd in constants.KNOWN_XEN_COMMANDS:
426 (name, testcls) = _MakeTestClass(cls, cmd)
427
428 assert name not in locals()
429
430 locals()[name] = testcls
431
432
433 if __name__ == "__main__":
434 testutils.GanetiTestProgram()