Synnefo » snf-ganeti

#!/usr/bin/python

#

# Copyright (C) 2008, 2011, 2012, 2013 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Tests for LUInstance*

"""

import copy

import itertools

import re

import unittest

import mock

import operator

from ganeti import compat

from ganeti import constants

from ganeti import errors

from ganeti import ht

from ganeti import opcodes

from ganeti import objects

from ganeti.rpc import node as rpc

from ganeti import utils

from ganeti.cmdlib import instance

from ganeti.cmdlib import instance_utils

from cmdlib.cmdlib_unittest import _StubComputeIPolicySpecViolation, _FakeLU

from testsupport import *

import testutils

class TestComputeIPolicyInstanceSpecViolation(unittest.TestCase):

  def test(self):

    ispec = {

      constants.ISPEC_MEM_SIZE: 2048,

      constants.ISPEC_CPU_COUNT: 2,

      constants.ISPEC_DISK_COUNT: 1,

      constants.ISPEC_DISK_SIZE: [512],

      constants.ISPEC_NIC_COUNT: 0,

      constants.ISPEC_SPINDLE_USE: 1,

      }

    stub = _StubComputeIPolicySpecViolation(2048, 2, 1, 0, [512], 1,

                                            constants.DT_PLAIN)

    ret = instance._ComputeIPolicyInstanceSpecViolation(NotImplemented, ispec,

                                                        constants.DT_PLAIN,

                                                        _compute_fn=stub)

    self.assertEqual(ret, [])

class TestLUInstanceCreate(CmdlibTestCase):

  def setUp(self):

    super(TestLUInstanceCreate, self).setUp()

    self.net = self.cfg.AddNewNetwork()

    self.cfg.ConnectNetworkToGroup(self.net, self.group)

    self.node1 = self.cfg.AddNewNode()

    self.node2 = self.cfg.AddNewNode()

    self.rpc.call_os_get.side_effect = \

      lambda node, _: self.RpcResultsBuilder() \

                        .CreateSuccessfulNodeResult(node, self.os)

    hv_info = ("bootid",

               [{

                 "type": constants.ST_LVM_VG,

                 "storage_free": 10000

               }],

               ({"memory_free": 10000}, ))

    self.rpc.call_node_info.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, hv_info) \

        .AddSuccessfulNode(self.node1, hv_info) \

        .AddSuccessfulNode(self.node2, hv_info) \

        .Build()

    self.rpc.call_blockdev_getmirrorstatus.side_effect = \

      lambda node, _: self.RpcResultsBuilder() \

                        .CreateSuccessfulNodeResult(node, [])

    self.iallocator_cls.return_value.result = [self.node1.name, self.node2.name]

    self.diskless_op = opcodes.OpInstanceCreate(

      instance_name="diskless.test.com",

      pnode=self.master.name,

      disk_template=constants.DT_DISKLESS,

      mode=constants.INSTANCE_CREATE,

      nics=[{}],

      disks=[],

      os_type=self.os_name_variant)

    self.plain_op = opcodes.OpInstanceCreate(

      instance_name="plain.test.com",

      pnode=self.master.name,

      disk_template=constants.DT_PLAIN,

      mode=constants.INSTANCE_CREATE,

      nics=[{}],

      disks=[{

        constants.IDISK_SIZE: 1024

      }],

      os_type=self.os_name_variant)

    self.block_op = opcodes.OpInstanceCreate(

      instance_name="block.test.com",

      pnode=self.master.name,

      disk_template=constants.DT_BLOCK,

      mode=constants.INSTANCE_CREATE,

      nics=[{}],

      disks=[{

        constants.IDISK_SIZE: 1024,

        constants.IDISK_ADOPT: "/dev/disk/block0"

      }],

      os_type=self.os_name_variant)

    self.drbd_op = opcodes.OpInstanceCreate(

      instance_name="drbd.test.com",

      pnode=self.node1.name,

      snode=self.node2.name,

      disk_template=constants.DT_DRBD8,

      mode=constants.INSTANCE_CREATE,

      nics=[{}],

      disks=[{

        constants.IDISK_SIZE: 1024

      }],

      os_type=self.os_name_variant)

    self.file_op = opcodes.OpInstanceCreate(

      instance_name="file.test.com",

      pnode=self.node1.name,

      disk_template=constants.DT_FILE,

      mode=constants.INSTANCE_CREATE,

      nics=[{}],

      disks=[{

        constants.IDISK_SIZE: 1024

      }],

      os_type=self.os_name_variant)

  def testSimpleCreate(self):

    op = self.CopyOpCode(self.diskless_op)

    self.ExecOpCode(op)

  def testStrangeHostnameResolve(self):

    op = self.CopyOpCode(self.diskless_op)

    self.netutils_mod.GetHostname.return_value = \

      HostnameMock("random.host.example.com", "203.0.113.1")

    self.ExecOpCodeExpectOpPrereqError(

      op, "Resolved hostname .* does not look the same as given hostname")

  def testOpportunisticLockingNoIAllocator(self):

    op = self.CopyOpCode(self.diskless_op,

                         opportunistic_locking=True,

                         iallocator=None)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Opportunistic locking is only available in combination with an"

          " instance allocator")

  def testNicWithNetAndMode(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_NETWORK: self.net.name,

                           constants.INIC_MODE: constants.NIC_MODE_BRIDGED

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "If network is given, no mode or link is allowed to be passed")

  def testAutoIpNoNameCheck(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: constants.VALUE_AUTO

                         }],

                         ip_check=False,

                         name_check=False)

    self.ExecOpCodeExpectOpPrereqError(

      op, "IP address set to auto but name checks have been skipped")

  def testAutoIp(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: constants.VALUE_AUTO

                         }])

    self.ExecOpCode(op)

  def testPoolIpNoNetwork(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: constants.NIC_IP_POOL

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "if ip=pool, parameter network must be passed too")

  def testValidIp(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: "203.0.113.1"

                         }])

    self.ExecOpCode(op)

  def testRoutedNoIp(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_MODE: constants.NIC_MODE_ROUTED

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "Routed nic mode requires an ip address")

  def testValicMac(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_MAC: "f0:df:f4:a3:d1:cf"

                         }])

    self.ExecOpCode(op)

  def testValidNicParams(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_MODE: constants.NIC_MODE_BRIDGED,

                           constants.INIC_LINK: "br_mock"

                         }])

    self.ExecOpCode(op)

  def testValidNicParamsOpenVSwitch(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_MODE: constants.NIC_MODE_OVS,

                           constants.INIC_VLAN: "1"

                         }])

    self.ExecOpCode(op)

  def testNicNoneName(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_NAME: constants.VALUE_NONE

                         }])

    self.ExecOpCode(op)

  def testConflictingIP(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: self.net.gateway[:-1] + "2"

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "The requested IP address .* belongs to network .*, but the target"

          " NIC does not.")

  def testVLanFormat(self):

    for vlan in [".pinky", ":bunny", ":1:pinky", "bunny"]:

      self.ResetMocks()

      op = self.CopyOpCode(self.diskless_op,

                           nics=[{

                             constants.INIC_VLAN: vlan

                           }])

      self.ExecOpCodeExpectOpPrereqError(

        op, "Specified VLAN parameter is invalid")

  def testPoolIp(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: constants.NIC_IP_POOL,

                           constants.INIC_NETWORK: self.net.name

                         }])

    self.ExecOpCode(op)

  def testPoolIpUnconnectedNetwork(self):

    net = self.cfg.AddNewNetwork()

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: constants.NIC_IP_POOL,

                           constants.INIC_NETWORK: net.name

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "No netparams found for network .*.")

  def testIpNotInNetwork(self):

    op = self.CopyOpCode(self.diskless_op,

                         nics=[{

                           constants.INIC_IP: "203.0.113.1",

                           constants.INIC_NETWORK: self.net.name

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "IP address .* already in use or does not belong to network .*")

  def testMixAdoptAndNotAdopt(self):

    op = self.CopyOpCode(self.diskless_op,

                         disk_template=constants.DT_PLAIN,

                         disks=[{

                           constants.IDISK_ADOPT: "lv1"

                         }, {}])

    self.ExecOpCodeExpectOpPrereqError(

      op, "Either all disks are adopted or none is")

  def testMustAdoptWithoutAdopt(self):

    op = self.CopyOpCode(self.diskless_op,

                         disk_template=constants.DT_BLOCK,

                         disks=[{}])

    self.ExecOpCodeExpectOpPrereqError(

      op, "Disk template blockdev requires disk adoption, but no 'adopt'"

          " parameter given")

  def testDontAdoptWithAdopt(self):

    op = self.CopyOpCode(self.diskless_op,

                         disk_template=constants.DT_DRBD8,

                         disks=[{

                           constants.IDISK_ADOPT: "lv1"

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "Disk adoption is not supported for the 'drbd' disk template")

  def testAdoptWithIAllocator(self):

    op = self.CopyOpCode(self.diskless_op,

                         disk_template=constants.DT_PLAIN,

                         disks=[{

                           constants.IDISK_ADOPT: "lv1"

                         }],

                         iallocator="mock")

    self.ExecOpCodeExpectOpPrereqError(

      op, "Disk adoption not allowed with an iallocator script")

  def testAdoptWithImport(self):

    op = self.CopyOpCode(self.diskless_op,

                         disk_template=constants.DT_PLAIN,

                         disks=[{

                           constants.IDISK_ADOPT: "lv1"

                         }],

                         mode=constants.INSTANCE_IMPORT)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Disk adoption not allowed for instance import")

  def testArgumentCombinations(self):

    op = self.CopyOpCode(self.diskless_op,

                         # start flag will be flipped

                         no_install=True,

                         start=True,

                         # no allowed combination

                         ip_check=True,

                         name_check=False)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Cannot do IP address check without a name check")

  def testInvalidFileDriver(self):

    op = self.CopyOpCode(self.diskless_op,

                         file_driver="invalid_file_driver")

    self.ExecOpCodeExpectOpPrereqError(

      op, "Parameter 'OP_INSTANCE_CREATE.file_driver' fails validation")

  def testMissingSecondaryNode(self):

    op = self.CopyOpCode(self.diskless_op,

                         pnode=self.master.name,

                         disk_template=constants.DT_DRBD8)

    self.ExecOpCodeExpectOpPrereqError(

      op, "The networked disk templates need a mirror node")

  def testIgnoredSecondaryNode(self):

    op = self.CopyOpCode(self.diskless_op,

                         pnode=self.master.name,

                         snode=self.node1.name,

                         disk_template=constants.DT_PLAIN)

    try:

      self.ExecOpCode(op)

    except Exception:

      pass

    self.mcpu.assertLogContainsRegex(

      "Secondary node will be ignored on non-mirrored disk template")

  def testMissingOsType(self):

    op = self.CopyOpCode(self.diskless_op,

                         os_type=self.REMOVE)

    self.ExecOpCodeExpectOpPrereqError(op, "No guest OS or OS image specified")

  def testBlacklistedOs(self):

    self.cluster.blacklisted_os = [self.os_name_variant]

    op = self.CopyOpCode(self.diskless_op)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Guest OS .* is not allowed for installation")

  def testMissingDiskTemplate(self):

    self.cluster.enabled_disk_templates = [constants.DT_DISKLESS]

    op = self.CopyOpCode(self.diskless_op,

                         disk_template=self.REMOVE)

    self.ExecOpCode(op)

  def testExistingInstance(self):

    inst = self.cfg.AddNewInstance()

    op = self.CopyOpCode(self.diskless_op,

                         instance_name=inst.name)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Instance .* is already in the cluster")

  def testPlainInstance(self):

    op = self.CopyOpCode(self.plain_op)

    self.ExecOpCode(op)

  def testPlainIAllocator(self):

    op = self.CopyOpCode(self.plain_op,

                         pnode=self.REMOVE,

                         iallocator="mock")

    self.ExecOpCode(op)

  def testIAllocatorOpportunisticLocking(self):

    op = self.CopyOpCode(self.plain_op,

                         pnode=self.REMOVE,

                         iallocator="mock",

                         opportunistic_locking=True)

    self.ExecOpCode(op)

  def testFailingIAllocator(self):

    self.iallocator_cls.return_value.success = False

    op = self.CopyOpCode(self.plain_op,

                         pnode=self.REMOVE,

                         iallocator="mock")

    self.ExecOpCodeExpectOpPrereqError(

      op, "Can't compute nodes using iallocator")

  def testDrbdInstance(self):

    op = self.CopyOpCode(self.drbd_op)

    self.ExecOpCode(op)

  def testDrbdIAllocator(self):

    op = self.CopyOpCode(self.drbd_op,

                         pnode=self.REMOVE,

                         snode=self.REMOVE,

                         iallocator="mock")

    self.ExecOpCode(op)

  def testFileInstance(self):

    op = self.CopyOpCode(self.file_op)

    self.ExecOpCode(op)

  def testFileInstanceNoClusterStorage(self):

    self.cluster.file_storage_dir = None

    op = self.CopyOpCode(self.file_op)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Cluster file storage dir for 'file' storage type not defined")

  def testFileInstanceAdditionalPath(self):

    op = self.CopyOpCode(self.file_op,

                         file_storage_dir="mock_dir")

    self.ExecOpCode(op)

  def testIdentifyDefaults(self):

    op = self.CopyOpCode(self.plain_op,

                         hvparams={

                           constants.HV_BOOT_ORDER: "cd"

                         },

                         beparams=constants.BEC_DEFAULTS.copy(),

                         nics=[{

                           constants.NIC_MODE: constants.NIC_MODE_BRIDGED

                         }],

                         osparams={

                           self.os_name_variant: {}

                         },

                         osparams_private={},

                         identify_defaults=True)

    self.ExecOpCode(op)

    inst = self.cfg.GetAllInstancesInfo().values()[0]

    self.assertEqual(0, len(inst.hvparams))

    self.assertEqual(0, len(inst.beparams))

    assert self.os_name_variant not in inst.osparams or \

            len(inst.osparams[self.os_name_variant]) == 0

  def testOfflineNode(self):

    self.node1.offline = True

    op = self.CopyOpCode(self.diskless_op,

                         pnode=self.node1.name)

    self.ExecOpCodeExpectOpPrereqError(op, "Cannot use offline primary node")

  def testDrainedNode(self):

    self.node1.drained = True

    op = self.CopyOpCode(self.diskless_op,

                         pnode=self.node1.name)

    self.ExecOpCodeExpectOpPrereqError(op, "Cannot use drained primary node")

  def testNonVmCapableNode(self):

    self.node1.vm_capable = False

    op = self.CopyOpCode(self.diskless_op,

                         pnode=self.node1.name)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Cannot use non-vm_capable primary node")

  def testNonEnabledHypervisor(self):

    self.cluster.enabled_hypervisors = [constants.HT_XEN_HVM]

    op = self.CopyOpCode(self.diskless_op,

                         hypervisor=constants.HT_FAKE)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Selected hypervisor .* not enabled in the cluster")

  def testAddTag(self):

    op = self.CopyOpCode(self.diskless_op,

                         tags=["tag"])

    self.ExecOpCode(op)

  def testInvalidTag(self):

    op = self.CopyOpCode(self.diskless_op,

                         tags=["too_long" * 20])

    self.ExecOpCodeExpectException(op, errors.TagError, "Tag too long")

  def testPingableInstanceName(self):

    self.netutils_mod.TcpPing.return_value = True

    op = self.CopyOpCode(self.diskless_op)

    self.ExecOpCodeExpectOpPrereqError(

      op, "IP .* of instance diskless.test.com already in use")

  def testPrimaryIsSecondaryNode(self):

    op = self.CopyOpCode(self.drbd_op,

                         snode=self.drbd_op.pnode)

    self.ExecOpCodeExpectOpPrereqError(

      op, "The secondary node cannot be the primary node")

  def testPrimarySecondaryDifferentNodeGroups(self):

    group = self.cfg.AddNewNodeGroup()

    self.node2.group = group.uuid

    op = self.CopyOpCode(self.drbd_op)

    self.ExecOpCode(op)

    self.mcpu.assertLogContainsRegex(

      "The primary and secondary nodes are in two different node groups")

  def testExclusiveStorageUnsupportedDiskTemplate(self):

    self.node1.ndparams[constants.ND_EXCLUSIVE_STORAGE] = True

    op = self.CopyOpCode(self.drbd_op)

    self.ExecOpCodeExpectOpPrereqError(

      op, "Disk template drbd not supported with exclusive storage")

  def testAdoptPlain(self):

    self.rpc.call_lv_list.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, {

          "xenvg/mock_disk_1": (10000, None, False)

        }) \

        .Build()

    op = self.CopyOpCode(self.plain_op)

    op.disks[0].update({constants.IDISK_ADOPT: "mock_disk_1"})

    self.ExecOpCode(op)

  def testAdoptPlainMissingLv(self):

    self.rpc.call_lv_list.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, {}) \

        .Build()

    op = self.CopyOpCode(self.plain_op)

    op.disks[0].update({constants.IDISK_ADOPT: "mock_disk_1"})

    self.ExecOpCodeExpectOpPrereqError(op, "Missing logical volume")

  def testAdoptPlainOnlineLv(self):

    self.rpc.call_lv_list.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, {

          "xenvg/mock_disk_1": (10000, None, True)

        }) \

        .Build()

    op = self.CopyOpCode(self.plain_op)

    op.disks[0].update({constants.IDISK_ADOPT: "mock_disk_1"})

    self.ExecOpCodeExpectOpPrereqError(

      op, "Online logical volumes found, cannot adopt")

  def testAdoptBlock(self):

    self.rpc.call_bdev_sizes.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, {

          "/dev/disk/block0": 10000

        }) \

        .Build()

    op = self.CopyOpCode(self.block_op)

    self.ExecOpCode(op)

  def testAdoptBlockDuplicateNames(self):

    op = self.CopyOpCode(self.block_op,

                         disks=[{

                           constants.IDISK_SIZE: 0,

                           constants.IDISK_ADOPT: "/dev/disk/block0"

                         }, {

                           constants.IDISK_SIZE: 0,

                           constants.IDISK_ADOPT: "/dev/disk/block0"

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "Duplicate disk names given for adoption")

  def testAdoptBlockInvalidNames(self):

    op = self.CopyOpCode(self.block_op,

                         disks=[{

                           constants.IDISK_SIZE: 0,

                           constants.IDISK_ADOPT: "/invalid/block0"

                         }])

    self.ExecOpCodeExpectOpPrereqError(

      op, "Device node.* lie outside .* and cannot be adopted")

  def testAdoptBlockMissingDisk(self):

    self.rpc.call_bdev_sizes.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, {}) \

        .Build()

    op = self.CopyOpCode(self.block_op)

    self.ExecOpCodeExpectOpPrereqError(op, "Missing block device")

  def testNoWaitForSyncDrbd(self):

    op = self.CopyOpCode(self.drbd_op,

                         wait_for_sync=False)

    self.ExecOpCode(op)

  def testNoWaitForSyncPlain(self):

    op = self.CopyOpCode(self.plain_op,

                         wait_for_sync=False)

    self.ExecOpCode(op)

  def testImportPlainFromGivenSrcNode(self):

    exp_info = """

[export]

version=0

os=mock_os

[instance]

name=old_name.example.com

"""

    self.rpc.call_export_info.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, exp_info)

    op = self.CopyOpCode(self.plain_op,

                         mode=constants.INSTANCE_IMPORT,

                         src_node=self.master.name)

    self.ExecOpCode(op)

  def testImportPlainWithoutSrcNodeNotFound(self):

    op = self.CopyOpCode(self.plain_op,

                         mode=constants.INSTANCE_IMPORT)

    self.ExecOpCodeExpectOpPrereqError(

      op, "No export found for relative path")

  def testImportPlainWithoutSrcNode(self):

    exp_info = """

[export]

version=0

os=mock_os

[instance]

name=old_name.example.com

"""

    self.rpc.call_export_list.return_value = \

      self.RpcResultsBuilder() \

        .AddSuccessfulNode(self.master, {"mock_path": {}}) \

        .Build()

    self.rpc.call_export_info.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, exp_info)

    op = self.CopyOpCode(self.plain_op,

                         mode=constants.INSTANCE_IMPORT,

                         src_path="mock_path")

    self.ExecOpCode(op)

  def testImportPlainCorruptExportInfo(self):

    exp_info = ""

    self.rpc.call_export_info.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, exp_info)

    op = self.CopyOpCode(self.plain_op,

                         mode=constants.INSTANCE_IMPORT,

                         src_node=self.master.name)

    self.ExecOpCodeExpectException(op, errors.ProgrammerError,

                                   "Corrupted export config")

  def testImportPlainWrongExportInfoVersion(self):

    exp_info = """

[export]

version=1

"""

    self.rpc.call_export_info.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, exp_info)

    op = self.CopyOpCode(self.plain_op,

                         mode=constants.INSTANCE_IMPORT,

                         src_node=self.master.name)

    self.ExecOpCodeExpectOpPrereqError(op, "Wrong export version")

  def testImportPlainWithParametersAndImport(self):

    exp_info = """

[export]

version=0

os=mock_os

[instance]

name=old_name.example.com

disk0_size=1024

disk1_size=1500

disk1_dump=mock_path

nic0_mode=bridged

nic0_link=br_mock

nic0_mac=f6:ab:f4:45:d1:af

nic0_ip=192.0.2.1

tags=tag1 tag2

hypervisor=xen-hvm

[hypervisor]

boot_order=cd

[backend]

memory=1024

vcpus=8

[os]

param1=val1

"""

    self.rpc.call_export_info.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, exp_info)

    self.rpc.call_import_start.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, "daemon_name")

    self.rpc.call_impexp_status.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master,

                                    [

                                      objects.ImportExportStatus(exit_status=0)

                                    ])

    self.rpc.call_impexp_cleanup.return_value = \

      self.RpcResultsBuilder() \

        .CreateSuccessfulNodeResult(self.master, True)

    op = self.CopyOpCode(self.plain_op,

                         disks=[],

                         nics=[],

                         tags=[],

                         hypervisor=None,

                         hvparams={},

                         mode=constants.INSTANCE_IMPORT,

                         src_node=self.master.name)

    self.ExecOpCode(op)

class TestCheckOSVariant(CmdlibTestCase):

  def testNoVariantsSupported(self):

    os = self.cfg.CreateOs(supported_variants=[])

    self.assertRaises(errors.OpPrereqError, instance_utils._CheckOSVariant,

                      os, "os+variant")

  def testNoVariantGiven(self):

    os = self.cfg.CreateOs(supported_variants=["default"])

    self.assertRaises(errors.OpPrereqError, instance_utils._CheckOSVariant,

                      os, "os")

  def testWrongVariantGiven(self):

    os = self.cfg.CreateOs(supported_variants=["default"])

    self.assertRaises(errors.OpPrereqError, instance_utils._CheckOSVariant,

                      os, "os+wrong_variant")

  def testOkWithVariant(self):

    os = self.cfg.CreateOs(supported_variants=["default"])

    instance_utils._CheckOSVariant(os, "os+default")

  def testOkWithoutVariant(self):

    os = self.cfg.CreateOs(supported_variants=[])

    instance_utils._CheckOSVariant(os, "os")

class TestCheckTargetNodeIPolicy(TestLUInstanceCreate):

  def setUp(self):

    super(TestCheckTargetNodeIPolicy, self).setUp()

    self.op = self.diskless_op

    self.instance = self.cfg.AddNewInstance()

    self.target_group = self.cfg.AddNewNodeGroup()

    self.target_node = self.cfg.AddNewNode(group=self.target_group)

  @withLockedLU

  def testNoViolation(self, lu):

    compute_recoder = mock.Mock(return_value=[])

    instance.CheckTargetNodeIPolicy(lu, NotImplemented, self.instance,

                                    self.target_node, NotImplemented,

                                    _compute_fn=compute_recoder)

    self.assertTrue(compute_recoder.called)

    self.mcpu.assertLogIsEmpty()

  @withLockedLU

  def testNoIgnore(self, lu):

    compute_recoder = mock.Mock(return_value=["mem_size not in range"])

    self.assertRaises(errors.OpPrereqError, instance.CheckTargetNodeIPolicy,

                      lu, NotImplemented, self.instance,

                      self.target_node, NotImplemented,

                      _compute_fn=compute_recoder)

    self.assertTrue(compute_recoder.called)

    self.mcpu.assertLogIsEmpty()