Commit 70b634e6 authored by Thomas Thrainer's avatar Thomas Thrainer
Browse files

Add unit tests for LUBackupExport



This patch adds unit tests for LUBackupExport.
Signed-off-by: default avatarThomas Thrainer <thomasth@google.com>
Reviewed-by: default avatarHelga Velroyen <helgav@google.com>
parent 3388debb
......@@ -396,7 +396,7 @@ class LUBackupExport(LogicalUnit):
activate_disks = not self.instance.disks_active
if activate_disks:
# Activate the instance disks if we'exporting a stopped instance
# Activate the instance disks if we're exporting a stopped instance
feedback_fn("Activating disks for %s" % self.instance.name)
StartInstanceDisks(self, self.instance, None)
......
......@@ -21,9 +21,8 @@
"""Tests for LUBackup*"""
import mock
from ganeti import constants
from ganeti import objects
from ganeti import opcodes
from ganeti import query
......@@ -95,5 +94,116 @@ class TestLUBackupPrepare(CmdlibTestCase):
self.ExecOpCode(op)
class TestLUBackupExportBase(CmdlibTestCase):
def setUp(self):
super(TestLUBackupExportBase, self).setUp()
self.rpc.call_instance_start.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, True)
self.rpc.call_blockdev_assemble.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, None)
self.rpc.call_blockdev_shutdown.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, None)
self.rpc.call_blockdev_snapshot.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, ("mock_vg", "mock_id"))
self.rpc.call_blockdev_remove.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, None)
self.rpc.call_export_start.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, "export_daemon")
def ImpExpStatus(node_uuid, name):
return self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(node_uuid,
[objects.ImportExportStatus(
exit_status=0
)])
self.rpc.call_impexp_status.side_effect = ImpExpStatus
def ImpExpCleanup(node_uuid, name):
return self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(node_uuid)
self.rpc.call_impexp_cleanup.side_effect = ImpExpCleanup
self.rpc.call_finalize_export.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.master, None)
def testRemoveRunningInstanceWithoutShutdown(self):
inst = self.cfg.AddNewInstance(admin_state=constants.ADMINST_UP)
op = opcodes.OpBackupExport(instance_name=inst.name,
target_node=self.master.name,
shutdown=False,
remove_instance=True)
self.ExecOpCodeExpectOpPrereqError(
op, "Can not remove instance without shutting it down before")
def testUnsupportedDiskTemplate(self):
inst = self.cfg.AddNewInstance(disk_template=constants.DT_FILE)
op = opcodes.OpBackupExport(instance_name=inst.name,
target_node=self.master.name)
self.ExecOpCodeExpectOpPrereqError(
op, "Export not supported for instances with file-based disks")
class TestLUBackupExportLocalExport(TestLUBackupExportBase):
def setUp(self):
super(TestLUBackupExportLocalExport, self).setUp()
self.inst = self.cfg.AddNewInstance()
self.target_node = self.cfg.AddNewNode()
self.op = opcodes.OpBackupExport(mode=constants.EXPORT_MODE_LOCAL,
instance_name=self.inst.name,
target_node=self.target_node.name)
self.rpc.call_import_start.return_value = \
self.RpcResultsBuilder() \
.CreateSuccessfulNodeResult(self.target_node, "import_daemon")
def testExportWithShutdown(self):
inst = self.cfg.AddNewInstance(admin_state=constants.ADMINST_UP)
op = self.CopyOpCode(self.op, instance_name=inst.name, shutdown=True)
self.ExecOpCode(op)
def testExportDeactivatedDisks(self):
self.ExecOpCode(self.op)
def testExportRemoveInstance(self):
op = self.CopyOpCode(self.op, remove_instance=True)
self.ExecOpCode(op)
class TestLUBackupExportRemoteExport(TestLUBackupExportBase):
def setUp(self):
super(TestLUBackupExportRemoteExport, self).setUp()
self.inst = self.cfg.AddNewInstance()
self.op = opcodes.OpBackupExport(mode=constants.EXPORT_MODE_REMOTE,
instance_name=self.inst.name,
target_node=[],
x509_key_name=["mock_key_name"],
destination_x509_ca="mock_dest_ca")
def testRemoteExportWithoutX509KeyName(self):
op = self.CopyOpCode(self.op, x509_key_name=self.REMOVE)
self.ExecOpCodeExpectOpPrereqError(op,
"Missing X509 key name for encryption")
def testRemoteExportWithoutX509DestCa(self):
op = self.CopyOpCode(self.op, destination_x509_ca=self.REMOVE)
self.ExecOpCodeExpectOpPrereqError(op,
"Missing destination X509 CA")
if __name__ == "__main__":
testutils.GanetiTestProgram()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment