X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/54dc4fdb89d9c432c52cdc9904a25f98b36e3e9b..8f9a87c57f0c6de8a2176f8c988e26d6efb2b50d:/test/ganeti.rapi.rlib2_unittest.py diff --git a/test/ganeti.rapi.rlib2_unittest.py b/test/ganeti.rapi.rlib2_unittest.py index 4316a6c..57836e1 100755 --- a/test/ganeti.rapi.rlib2_unittest.py +++ b/test/ganeti.rapi.rlib2_unittest.py @@ -31,12 +31,21 @@ from ganeti import constants from ganeti import opcodes from ganeti import compat from ganeti import http +from ganeti import query from ganeti.rapi import rlib2 import testutils +class TestConstants(unittest.TestCase): + def testConsole(self): + # Exporting the console field without authentication might expose + # information + assert "console" in query.INSTANCE_FIELDS + self.assertTrue("console" not in rlib2.I_FIELDS) + + class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): def setUp(self): testutils.GanetiTestCase.setUp(self) @@ -53,9 +62,6 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): # Disk with mode [{"size": 123, "mode": constants.DISK_RDWR, }], - - # With unknown setting - [{"size": 123, "unknown": 999 }], ] nic_variants = [ @@ -70,11 +76,8 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): { "ip": "192.0.2.6", "mode": constants.NIC_MODE_ROUTED, "mac": "01:23:45:67:68:9A", }, - { "mode": constants.NIC_MODE_BRIDGED, "link": "n0", "bridge": "br1", }, + { "mode": constants.NIC_MODE_BRIDGED, "link": "br1" }, ], - - # Unknown settings - [{ "unknown": 999, }, { "foobar": "Hello World", }], ] beparam_variants = [ @@ -107,6 +110,7 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): "nics": nics, "mode": mode, "disk_template": disk_template, + "os": "debootstrap", } if beparams is not None: @@ -117,7 +121,7 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): for dry_run in [False, True]: op = self.Parse(data, dry_run) - self.assert_(isinstance(op, opcodes.OpCreateInstance)) + self.assert_(isinstance(op, opcodes.OpInstanceCreate)) self.assertEqual(op.mode, mode) self.assertEqual(op.disk_template, disk_template) self.assertEqual(op.dry_run, dry_run) @@ -136,15 +140,69 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): self.assertFalse("foobar" in opnic) if beparams is None: - self.assertEqualValues(op.beparams, {}) + self.assertFalse(hasattr(op, "beparams")) else: self.assertEqualValues(op.beparams, beparams) if hvparams is None: - self.assertEqualValues(op.hvparams, {}) + self.assertFalse(hasattr(op, "hvparams")) else: self.assertEqualValues(op.hvparams, hvparams) + def testLegacyName(self): + name = "inst29128.example.com" + data = { + "name": name, + "disks": [], + "nics": [], + "mode": constants.INSTANCE_CREATE, + "disk_template": constants.DT_PLAIN, + } + op = self.Parse(data, False) + self.assert_(isinstance(op, opcodes.OpInstanceCreate)) + self.assertEqual(op.instance_name, name) + self.assertFalse(hasattr(op, "name")) + + # Define both + data = { + "name": name, + "instance_name": "other.example.com", + "disks": [], + "nics": [], + "mode": constants.INSTANCE_CREATE, + "disk_template": constants.DT_PLAIN, + } + self.assertRaises(http.HttpBadRequest, self.Parse, data, False) + + def testLegacyOs(self): + name = "inst4673.example.com" + os = "linux29206" + data = { + "name": name, + "os_type": os, + "disks": [], + "nics": [], + "mode": constants.INSTANCE_CREATE, + "disk_template": constants.DT_PLAIN, + } + op = self.Parse(data, False) + self.assert_(isinstance(op, opcodes.OpInstanceCreate)) + self.assertEqual(op.instance_name, name) + self.assertEqual(op.os_type, os) + self.assertFalse(hasattr(op, "os")) + + # Define both + data = { + "instance_name": name, + "os": os, + "os_type": "linux9584", + "disks": [], + "nics": [], + "mode": constants.INSTANCE_CREATE, + "disk_template": constants.DT_PLAIN, + } + self.assertRaises(http.HttpBadRequest, self.Parse, data, False) + def testErrors(self): # Test all required fields reqfields = { @@ -152,7 +210,7 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): "disks": [], "nics": [], "mode": constants.INSTANCE_CREATE, - "disk_template": constants.DT_PLAIN + "disk_template": constants.DT_PLAIN, } for name in reqfields.keys(): @@ -162,14 +220,8 @@ class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase): # Invalid disks and nics for field in ["disks", "nics"]: - invalid_values = [None, 1, "", {}, [1, 2, 3], ["hda1", "hda2"]] - - if field == "disks": - invalid_values.append([ - # Disks without size - {}, - { "mode": constants.DISK_RDWR, }, - ]) + invalid_values = [None, 1, "", {}, [1, 2, 3], ["hda1", "hda2"], + [{"_unknown_": 999, }]] for invvalue in invalid_values: data = reqfields.copy() @@ -190,17 +242,17 @@ class TestParseExportInstanceRequest(testutils.GanetiTestCase): "destination": [(1, 2, 3), (99, 99, 99)], "shutdown": True, "remove_instance": True, - "x509_key_name": ("name", "hash"), - "destination_x509_ca": ("x", "y", "z"), + "x509_key_name": ["name", "hash"], + "destination_x509_ca": "---cert---" } op = self.Parse(name, data) - self.assert_(isinstance(op, opcodes.OpExportInstance)) + self.assert_(isinstance(op, opcodes.OpBackupExport)) self.assertEqual(op.instance_name, name) self.assertEqual(op.mode, constants.EXPORT_MODE_REMOTE) self.assertEqual(op.shutdown, True) self.assertEqual(op.remove_instance, True) self.assertEqualValues(op.x509_key_name, ("name", "hash")) - self.assertEqualValues(op.destination_x509_ca, ("x", "y", "z")) + self.assertEqual(op.destination_x509_ca, "---cert---") def testDefaults(self): name = "inst1" @@ -209,10 +261,12 @@ class TestParseExportInstanceRequest(testutils.GanetiTestCase): "shutdown": False, } op = self.Parse(name, data) - self.assert_(isinstance(op, opcodes.OpExportInstance)) + self.assert_(isinstance(op, opcodes.OpBackupExport)) self.assertEqual(op.instance_name, name) - self.assertEqual(op.mode, constants.EXPORT_MODE_LOCAL) - self.assertEqual(op.remove_instance, False) + self.assertEqual(op.target_node, "node2") + self.assertFalse(hasattr(op, "mode")) + self.assertFalse(hasattr(op, "remove_instance")) + self.assertFalse(hasattr(op, "destination")) def testErrors(self): self.assertRaises(http.HttpBadRequest, self.Parse, "err1", @@ -221,5 +275,339 @@ class TestParseExportInstanceRequest(testutils.GanetiTestCase): { "remove_instance": "False", }) +class TestParseMigrateInstanceRequest(testutils.GanetiTestCase): + def setUp(self): + testutils.GanetiTestCase.setUp(self) + + self.Parse = rlib2._ParseMigrateInstanceRequest + + def test(self): + name = "instYooho6ek" + + for cleanup in [False, True]: + for mode in constants.HT_MIGRATION_MODES: + data = { + "cleanup": cleanup, + "mode": mode, + } + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpInstanceMigrate)) + self.assertEqual(op.instance_name, name) + self.assertEqual(op.mode, mode) + self.assertEqual(op.cleanup, cleanup) + + def testDefaults(self): + name = "instnohZeex0" + + op = self.Parse(name, {}) + self.assert_(isinstance(op, opcodes.OpInstanceMigrate)) + self.assertEqual(op.instance_name, name) + self.assertFalse(hasattr(op, "mode")) + self.assertFalse(hasattr(op, "cleanup")) + + +class TestParseRenameInstanceRequest(testutils.GanetiTestCase): + def setUp(self): + testutils.GanetiTestCase.setUp(self) + + self.Parse = rlib2._ParseRenameInstanceRequest + + def test(self): + name = "instij0eeph7" + + for new_name in ["ua0aiyoo", "fai3ongi"]: + for ip_check in [False, True]: + for name_check in [False, True]: + data = { + "new_name": new_name, + "ip_check": ip_check, + "name_check": name_check, + } + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpInstanceRename)) + self.assertEqual(op.instance_name, name) + self.assertEqual(op.new_name, new_name) + self.assertEqual(op.ip_check, ip_check) + self.assertEqual(op.name_check, name_check) + + def testDefaults(self): + name = "instahchie3t" + + for new_name in ["thag9mek", "quees7oh"]: + data = { + "new_name": new_name, + } + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpInstanceRename)) + self.assertEqual(op.instance_name, name) + self.assertEqual(op.new_name, new_name) + self.assertFalse(hasattr(op, "ip_check")) + self.assertFalse(hasattr(op, "name_check")) + + +class TestParseModifyInstanceRequest(testutils.GanetiTestCase): + def setUp(self): + testutils.GanetiTestCase.setUp(self) + + self.Parse = rlib2._ParseModifyInstanceRequest + + def test(self): + name = "instush8gah" + + test_disks = [ + [], + [(1, { constants.IDISK_MODE: constants.DISK_RDWR, })], + ] + + for osparams in [{}, { "some": "value", "other": "Hello World", }]: + for hvparams in [{}, { constants.HV_KERNEL_PATH: "/some/kernel", }]: + for beparams in [{}, { constants.BE_MEMORY: 128, }]: + for force in [False, True]: + for nics in [[], [(0, { constants.INIC_IP: "192.0.2.1", })]]: + for disks in test_disks: + for disk_template in constants.DISK_TEMPLATES: + data = { + "osparams": osparams, + "hvparams": hvparams, + "beparams": beparams, + "nics": nics, + "disks": disks, + "force": force, + "disk_template": disk_template, + } + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpInstanceSetParams)) + self.assertEqual(op.instance_name, name) + self.assertEqual(op.hvparams, hvparams) + self.assertEqual(op.beparams, beparams) + self.assertEqual(op.osparams, osparams) + self.assertEqual(op.force, force) + self.assertEqual(op.nics, nics) + self.assertEqual(op.disks, disks) + self.assertEqual(op.disk_template, disk_template) + self.assertFalse(hasattr(op, "remote_node")) + self.assertFalse(hasattr(op, "os_name")) + self.assertFalse(hasattr(op, "force_variant")) + + def testDefaults(self): + name = "instir8aish31" + + op = self.Parse(name, {}) + self.assert_(isinstance(op, opcodes.OpInstanceSetParams)) + self.assertEqual(op.instance_name, name) + for i in ["hvparams", "beparams", "osparams", "force", "nics", "disks", + "disk_template", "remote_node", "os_name", "force_variant"]: + self.assertFalse(hasattr(op, i)) + + +class TestParseInstanceReinstallRequest(testutils.GanetiTestCase): + def setUp(self): + testutils.GanetiTestCase.setUp(self) + + self.Parse = rlib2._ParseInstanceReinstallRequest + + def _Check(self, ops, name): + expcls = [ + opcodes.OpInstanceShutdown, + opcodes.OpInstanceReinstall, + opcodes.OpInstanceStartup, + ] + + self.assert_(compat.all(isinstance(op, exp) + for op, exp in zip(ops, expcls))) + self.assert_(compat.all(op.instance_name == name for op in ops)) + + def test(self): + name = "shoo0tihohma" + + ops = self.Parse(name, {"os": "sys1", "start": True,}) + self.assertEqual(len(ops), 3) + self._Check(ops, name) + self.assertEqual(ops[1].os_type, "sys1") + self.assertFalse(ops[1].osparams) + + ops = self.Parse(name, {"os": "sys2", "start": False,}) + self.assertEqual(len(ops), 2) + self._Check(ops, name) + self.assertEqual(ops[1].os_type, "sys2") + + osparams = { + "reformat": "1", + } + ops = self.Parse(name, {"os": "sys4035", "start": True, + "osparams": osparams,}) + self.assertEqual(len(ops), 3) + self._Check(ops, name) + self.assertEqual(ops[1].os_type, "sys4035") + self.assertEqual(ops[1].osparams, osparams) + + def testDefaults(self): + name = "noolee0g" + + ops = self.Parse(name, {"os": "linux1"}) + self.assertEqual(len(ops), 3) + self._Check(ops, name) + self.assertEqual(ops[1].os_type, "linux1") + self.assertFalse(ops[1].osparams) + + +class TestParseRenameGroupRequest(testutils.GanetiTestCase): + def setUp(self): + testutils.GanetiTestCase.setUp(self) + + self.Parse = rlib2._ParseRenameGroupRequest + + def test(self): + name = "instij0eeph7" + data = { + "new_name": "ua0aiyoo", + } + + op = self.Parse(name, data, False) + + self.assert_(isinstance(op, opcodes.OpGroupRename)) + self.assertEqual(op.group_name, name) + self.assertEqual(op.new_name, "ua0aiyoo") + self.assertFalse(op.dry_run) + + def testDryRun(self): + name = "instij0eeph7" + data = { + "new_name": "ua0aiyoo", + } + + op = self.Parse(name, data, True) + + self.assert_(isinstance(op, opcodes.OpGroupRename)) + self.assertEqual(op.group_name, name) + self.assertEqual(op.new_name, "ua0aiyoo") + self.assert_(op.dry_run) + + +class TestParseInstanceReplaceDisksRequest(unittest.TestCase): + def setUp(self): + self.Parse = rlib2._ParseInstanceReplaceDisksRequest + + def test(self): + name = "inst22568" + + for disks in [range(1, 4), "1,2,3", "1, 2, 3"]: + data = { + "mode": constants.REPLACE_DISK_SEC, + "disks": disks, + "iallocator": "myalloc", + } + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpInstanceReplaceDisks)) + self.assertEqual(op.mode, constants.REPLACE_DISK_SEC) + self.assertEqual(op.disks, [1, 2, 3]) + self.assertEqual(op.iallocator, "myalloc") + + def testDefaults(self): + name = "inst11413" + data = { + "mode": constants.REPLACE_DISK_AUTO, + } + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpInstanceReplaceDisks)) + self.assertEqual(op.mode, constants.REPLACE_DISK_AUTO) + self.assertFalse(hasattr(op, "iallocator")) + self.assertFalse(hasattr(op, "disks")) + + def testWrong(self): + self.assertRaises(http.HttpBadRequest, self.Parse, "inst", + { "mode": constants.REPLACE_DISK_AUTO, + "disks": "hello world", + }) + + +class TestParseModifyGroupRequest(unittest.TestCase): + def setUp(self): + self.Parse = rlib2._ParseModifyGroupRequest + + def test(self): + name = "group6002" + + for policy in constants.VALID_ALLOC_POLICIES: + data = { + "alloc_policy": policy, + } + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpGroupSetParams)) + self.assertEqual(op.group_name, name) + self.assertEqual(op.alloc_policy, policy) + + def testUnknownPolicy(self): + data = { + "alloc_policy": "_unknown_policy_", + } + + self.assertRaises(http.HttpBadRequest, self.Parse, "name", data) + + def testDefaults(self): + name = "group6679" + data = {} + + op = self.Parse(name, data) + self.assert_(isinstance(op, opcodes.OpGroupSetParams)) + self.assertEqual(op.group_name, name) + self.assertFalse(hasattr(op, "alloc_policy")) + + +class TestParseCreateGroupRequest(unittest.TestCase): + def setUp(self): + self.Parse = rlib2._ParseCreateGroupRequest + + def test(self): + name = "group3618" + + for policy in constants.VALID_ALLOC_POLICIES: + data = { + "group_name": name, + "alloc_policy": policy, + } + + op = self.Parse(data, False) + self.assert_(isinstance(op, opcodes.OpGroupAdd)) + self.assertEqual(op.group_name, name) + self.assertEqual(op.alloc_policy, policy) + self.assertFalse(op.dry_run) + + def testUnknownPolicy(self): + data = { + "alloc_policy": "_unknown_policy_", + } + + self.assertRaises(http.HttpBadRequest, self.Parse, "name", data) + + def testDefaults(self): + name = "group15395" + data = { + "group_name": name, + } + + op = self.Parse(data, True) + self.assert_(isinstance(op, opcodes.OpGroupAdd)) + self.assertEqual(op.group_name, name) + self.assertFalse(hasattr(op, "alloc_policy")) + self.assertTrue(op.dry_run) + + def testLegacyName(self): + name = "group29852" + data = { + "name": name, + } + + op = self.Parse(data, True) + self.assert_(isinstance(op, opcodes.OpGroupAdd)) + self.assertEqual(op.group_name, name) + + if __name__ == '__main__': testutils.GanetiTestProgram()