Synnefo » snf-ganeti

#

#

# Copyright (C) 2007, 2010, 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.

"""Cluster related QA tests.

"""

import re

import tempfile

import os.path

from ganeti import _constants

from ganeti import constants

from ganeti import compat

from ganeti import utils

from ganeti import pathutils

import qa_config

import qa_daemon

import qa_utils

import qa_error

import qa_instance

from qa_utils import AssertEqual, AssertCommand, GetCommandOutput

# Prefix for LVM volumes created by QA code during tests

_QA_LV_PREFIX = "qa-"

#: cluster verify command

_CLUSTER_VERIFY = ["gnt-cluster", "verify"]

def _RemoveFileFromAllNodes(filename):

  """Removes a file from all nodes.

  """

  for node in qa_config.get("nodes"):

    AssertCommand(["rm", "-f", filename], node=node)

def _CheckFileOnAllNodes(filename, content):

  """Verifies the content of the given file on all nodes.

  """

  cmd = utils.ShellQuoteArgs(["cat", filename])

  for node in qa_config.get("nodes"):

    AssertEqual(qa_utils.GetCommandOutput(node.primary, cmd), content)

def _GetClusterField(field_path):

  """Get the value of a cluster field.

  @type field_path: list of strings

  @param field_path: Names of the groups/fields to navigate to get the desired

      value, e.g. C{["Default node parameters", "oob_program"]}

  @return: The effective value of the field (the actual type depends on the

      chosen field)

  """

  assert isinstance(field_path, list)

  assert field_path

  ret = qa_utils.GetObjectInfo(["gnt-cluster", "info"])

  for key in field_path:

    ret = ret[key]

  return ret

# Cluster-verify errors (date, "ERROR", then error code)

_CVERROR_RE = re.compile(r"^[\w\s:]+\s+- (ERROR|WARNING):([A-Z0-9_-]+):")

def _GetCVErrorCodes(cvout):

  errs = set()

  warns = set()

  for l in cvout.splitlines():

    m = _CVERROR_RE.match(l)

    if m:

      etype = m.group(1)

      ecode = m.group(2)

      if etype == "ERROR":

        errs.add(ecode)

      elif etype == "WARNING":

        warns.add(ecode)

  return (errs, warns)

def _CheckVerifyErrors(actual, expected, etype):

  exp_codes = compat.UniqueFrozenset(e for (_, e, _) in expected)

  if not actual.issuperset(exp_codes):

    missing = exp_codes.difference(actual)

    raise qa_error.Error("Cluster-verify didn't return these expected"

                         " %ss: %s" % (etype, utils.CommaJoin(missing)))

def _CheckVerifyNoWarnings(actual, expected):

  exp_codes = compat.UniqueFrozenset(e for (_, e, _) in expected)

  excess = actual.intersection(exp_codes)

  if excess:

    raise qa_error.Error("Cluster-verify returned these warnings:"

                         " %s" % (utils.CommaJoin(excess)))

def AssertClusterVerify(fail=False, errors=None,

                        warnings=None, no_warnings=None):

  """Run cluster-verify and check the result, ignoring warnings by default.

  @type fail: bool

  @param fail: if cluster-verify is expected to fail instead of succeeding.

  @type errors: list of tuples

  @param errors: List of CV_XXX errors that are expected; if specified, all the

      errors listed must appear in cluster-verify output. A non-empty value

      implies C{fail=True}.

  @type warnings: list of tuples

  @param warnings: List of CV_XXX warnings that are expected to be raised; if

      specified, all the errors listed must appear in cluster-verify output.

  @type no_warnings: list of tuples

  @param no_warnings: List of CV_XXX warnings that we expect NOT to be raised.

  """

  cvcmd = "gnt-cluster verify"

  mnode = qa_config.GetMasterNode()

  if errors or warnings or no_warnings:

    cvout = GetCommandOutput(mnode.primary, cvcmd + " --error-codes",

                             fail=(fail or errors))

    print cvout

    (act_errs, act_warns) = _GetCVErrorCodes(cvout)

    if errors:

      _CheckVerifyErrors(act_errs, errors, "error")

    if warnings:

      _CheckVerifyErrors(act_warns, warnings, "warning")

    if no_warnings:

      _CheckVerifyNoWarnings(act_warns, no_warnings)

  else:

    AssertCommand(cvcmd, fail=fail, node=mnode)

# data for testing failures due to bad keys/values for disk parameters

_FAIL_PARAMS = ["nonexistent:resync-rate=1",

                "drbd:nonexistent=1",

                "drbd:resync-rate=invalid",

                ]

def TestClusterInitDisk():

  """gnt-cluster init -D"""

  name = qa_config.get("name")

  for param in _FAIL_PARAMS:

    AssertCommand(["gnt-cluster", "init", "-D", param, name], fail=True)

def TestClusterInit(rapi_user, rapi_secret):

  """gnt-cluster init"""

  master = qa_config.GetMasterNode()

  rapi_users_path = qa_utils.MakeNodePath(master, pathutils.RAPI_USERS_FILE)

  rapi_dir = os.path.dirname(rapi_users_path)

  # First create the RAPI credentials

  fh = tempfile.NamedTemporaryFile()

  try:

    fh.write("%s %s write\n" % (rapi_user, rapi_secret))

    fh.flush()

    tmpru = qa_utils.UploadFile(master.primary, fh.name)

    try:

      AssertCommand(["mkdir", "-p", rapi_dir])

      AssertCommand(["mv", tmpru, rapi_users_path])

    finally:

      AssertCommand(["rm", "-f", tmpru])

  finally:

    fh.close()

  # Initialize cluster

  enabled_disk_templates = qa_config.GetEnabledDiskTemplates()

  cmd = [

    "gnt-cluster", "init",

    "--primary-ip-version=%d" % qa_config.get("primary_ip_version", 4),

    "--enabled-hypervisors=%s" % ",".join(qa_config.GetEnabledHypervisors()),

    "--enabled-disk-templates=%s" %

      ",".join(enabled_disk_templates),

    ]

  if constants.DT_FILE in enabled_disk_templates:

    cmd.append(

        "--file-storage-dir=%s" %

        qa_config.get("default-file-storage-dir",

                      pathutils.DEFAULT_FILE_STORAGE_DIR))

  for spec_type in ("mem-size", "disk-size", "disk-count", "cpu-count",

                    "nic-count"):

    for spec_val in ("min", "max", "std"):

      spec = qa_config.get("ispec_%s_%s" %

                           (spec_type.replace("-", "_"), spec_val), None)

      if spec is not None:

        cmd.append("--specs-%s=%s=%d" % (spec_type, spec_val, spec))

  if master.secondary:

    cmd.append("--secondary-ip=%s" % master.secondary)

  if utils.IsLvmEnabled(qa_config.GetEnabledDiskTemplates()):

    vgname = qa_config.get("vg-name", constants.DEFAULT_VG)

    if vgname:

      cmd.append("--vg-name=%s" % vgname)

    else:

      raise qa_error.Error("Please specify a volume group if you enable"

                           " lvm-based disk templates in the QA.")

  master_netdev = qa_config.get("master-netdev", None)

  if master_netdev:

    cmd.append("--master-netdev=%s" % master_netdev)

  nicparams = qa_config.get("default-nicparams", None)

  if nicparams:

    cmd.append("--nic-parameters=%s" %

               ",".join(utils.FormatKeyValue(nicparams)))

  # Cluster value of the exclusive-storage node parameter

  e_s = qa_config.get("exclusive-storage")

  if e_s is not None:

    cmd.extend(["--node-parameters", "exclusive_storage=%s" % e_s])

  else:

    e_s = False

  qa_config.SetExclusiveStorage(e_s)

  extra_args = qa_config.get("cluster-init-args")

  if extra_args:

    # This option was removed in 2.10, but in order to not break QA of older

    # branches we remove it from the extra_args if it is in there.

    opt_drbd_storage = "--no-drbd-storage"

    if opt_drbd_storage in extra_args:

      extra_args.remove(opt_drbd_storage)

    cmd.extend(extra_args)

  cmd.append(qa_config.get("name"))

  AssertCommand(cmd)

  cmd = ["gnt-cluster", "modify"]

  # hypervisor parameter modifications

  hvp = qa_config.get("hypervisor-parameters", {})

  for k, v in hvp.items():

    cmd.extend(["-H", "%s:%s" % (k, v)])

  # backend parameter modifications

  bep = qa_config.get("backend-parameters", "")

  if bep:

    cmd.extend(["-B", bep])

  if len(cmd) > 2:

    AssertCommand(cmd)

  # OS parameters

  osp = qa_config.get("os-parameters", {})

  for k, v in osp.items():

    AssertCommand(["gnt-os", "modify", "-O", v, k])

  # OS hypervisor parameters

  os_hvp = qa_config.get("os-hvp", {})

  for os_name in os_hvp:

    for hv, hvp in os_hvp[os_name].items():

      AssertCommand(["gnt-os", "modify", "-H", "%s:%s" % (hv, hvp), os_name])

def TestClusterRename():

  """gnt-cluster rename"""

  cmd = ["gnt-cluster", "rename", "-f"]

  original_name = qa_config.get("name")

  rename_target = qa_config.get("rename", None)

  if rename_target is None:

    print qa_utils.FormatError('"rename" entry is missing')

    return

  for data in [

    cmd + [rename_target],

    _CLUSTER_VERIFY,

    cmd + [original_name],

    _CLUSTER_VERIFY,

    ]:

    AssertCommand(data)

def TestClusterOob():

  """out-of-band framework"""

  oob_path_exists = "/tmp/ganeti-qa-oob-does-exist-%s" % utils.NewUUID()

  AssertCommand(_CLUSTER_VERIFY)

  AssertCommand(["gnt-cluster", "modify", "--node-parameters",

                 "oob_program=/tmp/ganeti-qa-oob-does-not-exist-%s" %

                 utils.NewUUID()])

  AssertCommand(_CLUSTER_VERIFY, fail=True)

  AssertCommand(["touch", oob_path_exists])

  AssertCommand(["chmod", "0400", oob_path_exists])

  AssertCommand(["gnt-cluster", "copyfile", oob_path_exists])

  try:

    AssertCommand(["gnt-cluster", "modify", "--node-parameters",

                   "oob_program=%s" % oob_path_exists])

    AssertCommand(_CLUSTER_VERIFY, fail=True)

    AssertCommand(["chmod", "0500", oob_path_exists])

    AssertCommand(["gnt-cluster", "copyfile", oob_path_exists])

    AssertCommand(_CLUSTER_VERIFY)

  finally:

    AssertCommand(["gnt-cluster", "command", "rm", oob_path_exists])

  AssertCommand(["gnt-cluster", "modify", "--node-parameters",

                 "oob_program="])

def TestClusterEpo():

  """gnt-cluster epo"""

  master = qa_config.GetMasterNode()

  # Assert that OOB is unavailable for all nodes

  result_output = GetCommandOutput(master.primary,

                                   "gnt-node list --verbose --no-headers -o"

                                   " powered")

  AssertEqual(compat.all(powered == "(unavail)"

                         for powered in result_output.splitlines()), True)

  # Conflicting

  AssertCommand(["gnt-cluster", "epo", "--groups", "--all"], fail=True)

  # --all doesn't expect arguments

  AssertCommand(["gnt-cluster", "epo", "--all", "some_arg"], fail=True)

  # Unless --all is given master is not allowed to be in the list

  AssertCommand(["gnt-cluster", "epo", "-f", master.primary], fail=True)

  # This shouldn't fail

  AssertCommand(["gnt-cluster", "epo", "-f", "--all"])

  # All instances should have been stopped now

  result_output = GetCommandOutput(master.primary,

                                   "gnt-instance list --no-headers -o status")

  # ERROR_down because the instance is stopped but not recorded as such

  AssertEqual(compat.all(status == "ERROR_down"

                         for status in result_output.splitlines()), True)

  # Now start everything again

  AssertCommand(["gnt-cluster", "epo", "--on", "-f", "--all"])

  # All instances should have been started now

  result_output = GetCommandOutput(master.primary,

                                   "gnt-instance list --no-headers -o status")

  AssertEqual(compat.all(status == "running"

                         for status in result_output.splitlines()), True)

def TestClusterVerify():

  """gnt-cluster verify"""

  AssertCommand(_CLUSTER_VERIFY)

  AssertCommand(["gnt-cluster", "verify-disks"])

def TestClusterVerifyDisksBrokenDRBD(instance, inst_nodes):

  """gnt-cluster verify-disks with broken DRBD"""

  qa_daemon.TestPauseWatcher()

  try:

    info = qa_instance.GetInstanceInfo(instance.name)

    snode = inst_nodes[1]

    for idx, minor in enumerate(info["drbd-minors"][snode.primary]):

      if idx % 2 == 0:

        break_drbd_cmd = \

          "(drbdsetup %d down >/dev/null 2>&1;" \

          " drbdsetup down resource%d >/dev/null 2>&1) || /bin/true" % \

          (minor, minor)

      else:

        break_drbd_cmd = \

          "(drbdsetup %d detach >/dev/null 2>&1;" \

          " drbdsetup detach %d >/dev/null 2>&1) || /bin/true" % \

          (minor, minor)

      AssertCommand(break_drbd_cmd, node=snode)

    verify_output = GetCommandOutput(qa_config.GetMasterNode().primary,

                                     "gnt-cluster verify-disks")

    activation_msg = "Activating disks for instance '%s'" % instance.name

    if activation_msg not in verify_output:

      raise qa_error.Error("gnt-cluster verify-disks did not activate broken"

                           " DRBD disks:\n%s" % verify_output)

    verify_output = GetCommandOutput(qa_config.GetMasterNode().primary,

                                     "gnt-cluster verify-disks")

    if activation_msg in verify_output:

      raise qa_error.Error("gnt-cluster verify-disks wants to activate broken"

                           " DRBD disks on second attempt:\n%s" % verify_output)

    AssertCommand(_CLUSTER_VERIFY)

  finally:

    qa_daemon.TestResumeWatcher()

def TestJobqueue():

  """gnt-debug test-jobqueue"""

  AssertCommand(["gnt-debug", "test-jobqueue"])

def TestDelay(node):

  """gnt-debug delay"""

  AssertCommand(["gnt-debug", "delay", "1"])

  AssertCommand(["gnt-debug", "delay", "--no-master", "1"])

  AssertCommand(["gnt-debug", "delay", "--no-master",

                 "-n", node.primary, "1"])

def TestClusterReservedLvs():

  """gnt-cluster reserved lvs"""

  # if no lvm-based templates are supported, skip the test

  if not qa_config.IsStorageTypeSupported(constants.ST_LVM_VG):

    return

  vgname = qa_config.get("vg-name", constants.DEFAULT_VG)

  lvname = _QA_LV_PREFIX + "test"

  lvfullname = "/".join([vgname, lvname])

  # Clean cluster

  AssertClusterVerify()

  AssertCommand(["gnt-cluster", "modify", "--reserved-lvs", ""])

  AssertCommand(["lvcreate", "-L1G", "-n", lvname, vgname])

  AssertClusterVerify(fail=False,

                      warnings=[constants.CV_ENODEORPHANLV])

  AssertCommand(["gnt-cluster", "modify", "--reserved-lvs",

                 "%s,.*/other-test" % lvfullname])

  AssertClusterVerify(no_warnings=[constants.CV_ENODEORPHANLV])

  AssertCommand(["gnt-cluster", "modify", "--reserved-lvs",

                ".*/%s.*" % _QA_LV_PREFIX])

  AssertClusterVerify(no_warnings=[constants.CV_ENODEORPHANLV])

  AssertCommand(["gnt-cluster", "modify", "--reserved-lvs", ""])

  AssertClusterVerify(fail=False,

                      warnings=[constants.CV_ENODEORPHANLV])

  AssertCommand(["lvremove", "-f", lvfullname])

  AssertClusterVerify()

def TestClusterModifyEmpty():

  """gnt-cluster modify"""

  AssertCommand(["gnt-cluster", "modify"], fail=True)

def TestClusterModifyDisk():

  """gnt-cluster modify -D"""

  for param in _FAIL_PARAMS:

    AssertCommand(["gnt-cluster", "modify", "-D", param], fail=True)

def _GetOtherEnabledDiskTemplate(undesired_disk_templates,

                                 enabled_disk_templates):

  """Returns one template that is not in the undesired set.

  @type undesired_disk_templates: list of string

  @param undesired_disk_templates: a list of disk templates that we want to

      exclude when drawing one disk template from the list of enabled

      disk templates

  @type enabled_disk_templates: list of string

  @param enabled_disk_templates: list of enabled disk templates (in QA)

  """

  desired_templates = list(set(enabled_disk_templates)

                                - set(undesired_disk_templates))

  if desired_templates:

    template = desired_templates[0]

  else:

    # If no desired disk template is available for QA, choose 'diskless' and

    # hope for the best.

    template = constants.ST_DISKLESS

  return template

def TestClusterModifyFileBasedStorageDir(

    file_disk_template, dir_config_key, default_dir, option_name):

  """Tests gnt-cluster modify wrt to file-based directory options.

  @type file_disk_template: string

  @param file_disk_template: file-based disk template

  @type dir_config_key: string

  @param dir_config_key: key for the QA config to retrieve the default

     directory value

  @type default_dir: string

  @param default_dir: default directory, if the QA config does not specify

     it

  @type option_name: string

  @param option_name: name of the option of 'gnt-cluster modify' to

     change the directory

  """

  enabled_disk_templates = qa_config.GetEnabledDiskTemplates()

  assert file_disk_template in constants.DTS_FILEBASED

  if not qa_config.IsTemplateSupported(file_disk_template):

    return

  # Get some non-file-based disk template to disable file storage

  other_disk_template = _GetOtherEnabledDiskTemplate(

    utils.storage.GetDiskTemplatesOfStorageTypes(constants.ST_FILE,

                                                 constants.ST_SHARED_FILE),

    enabled_disk_templates

  )

  file_storage_dir = qa_config.get(dir_config_key, default_dir)

  invalid_file_storage_dir = "/boot/"

  for fail, cmd in [

    (False, ["gnt-cluster", "modify",

             "--enabled-disk-templates=%s" % file_disk_template,

             "--ipolicy-disk-templates=%s" % file_disk_template]),

    (False, ["gnt-cluster", "modify",

             "--%s=%s" % (option_name, file_storage_dir)]),

    (False, ["gnt-cluster", "modify",

             "--%s=%s" % (option_name, invalid_file_storage_dir)]),

    # file storage dir is set to an inacceptable path, thus verify

    # should fail

    (True, ["gnt-cluster", "verify"]),

    # unsetting the storage dir while file storage is enabled

    # should fail

    (True, ["gnt-cluster", "modify",

            "--%s=" % option_name]),

    (False, ["gnt-cluster", "modify",

             "--%s=%s" % (option_name, file_storage_dir)]),

    (False, ["gnt-cluster", "modify",

             "--enabled-disk-templates=%s" % other_disk_template,

             "--ipolicy-disk-templates=%s" % other_disk_template]),

    (False, ["gnt-cluster", "modify",

             "--%s=%s" % (option_name, invalid_file_storage_dir)]),

    # file storage is set to an inacceptable path, but file storage

    # is disabled, thus verify should not fail

    (False, ["gnt-cluster", "verify"]),

    # unsetting the file storage dir while file storage is not enabled

    # should be fine

    (False, ["gnt-cluster", "modify",

             "--%s=" % option_name]),

    # resetting everything to sane values

    (False, ["gnt-cluster", "modify",

             "--%s=%s" % (option_name, file_storage_dir),

             "--enabled-disk-templates=%s" % ",".join(enabled_disk_templates),

             "--ipolicy-disk-templates=%s" % ",".join(enabled_disk_templates)])

    ]:

    AssertCommand(cmd, fail=fail)

def TestClusterModifyFileStorageDir():

  """gnt-cluster modify --file-storage-dir=..."""

  TestClusterModifyFileBasedStorageDir(

      constants.DT_FILE, "default-file-storage-dir",

      pathutils.DEFAULT_FILE_STORAGE_DIR,

      "file-storage-dir")

def TestClusterModifySharedFileStorageDir():

  """gnt-cluster modify --shared-file-storage-dir=..."""

  TestClusterModifyFileBasedStorageDir(

      constants.DT_SHARED_FILE, "default-shared-file-storage-dir",

      pathutils.DEFAULT_SHARED_FILE_STORAGE_DIR,

      "shared-file-storage-dir")

def TestClusterModifyDiskTemplates():

  """gnt-cluster modify --enabled-disk-templates=..."""

  enabled_disk_templates = qa_config.GetEnabledDiskTemplates()

  default_disk_template = qa_config.GetDefaultDiskTemplate()

  _TestClusterModifyDiskTemplatesArguments(default_disk_template)

  _TestClusterModifyDiskTemplatesDrbdHelper(enabled_disk_templates)

  _TestClusterModifyDiskTemplatesVgName(enabled_disk_templates)

  _RestoreEnabledDiskTemplates()

  nodes = qa_config.AcquireManyNodes(2)

  instance_template = enabled_disk_templates[0]

  instance = qa_instance.CreateInstanceByDiskTemplate(nodes, instance_template)

  _TestClusterModifyUsedDiskTemplate(instance_template,

                                     enabled_disk_templates)

  qa_instance.TestInstanceRemove(instance)

  _RestoreEnabledDiskTemplates()

def _RestoreEnabledDiskTemplates():

  """Sets the list of enabled disk templates back to the list of enabled disk

     templates from the QA configuration. This can be used to make sure that

     the tests that modify the list of disk templates do not interfere with

     other tests.

  """

  enabled_disk_templates = qa_config.GetEnabledDiskTemplates()

  cmd = ["gnt-cluster", "modify",

         "--enabled-disk-templates=%s" % ",".join(enabled_disk_templates),

         "--ipolicy-disk-templates=%s" % ",".join(enabled_disk_templates),

         ]

  if utils.IsLvmEnabled(qa_config.GetEnabledDiskTemplates()):

    vgname = qa_config.get("vg-name", constants.DEFAULT_VG)

    cmd.append("--vg-name=%s" % vgname)

  AssertCommand(cmd, fail=False)

def _TestClusterModifyDiskTemplatesDrbdHelper(enabled_disk_templates):

  """Tests argument handling of 'gnt-cluster modify' with respect to

     the parameter '--drbd-usermode-helper'. This test is independent

     of instances.

  """

  _RestoreEnabledDiskTemplates()

  if constants.DT_DRBD8 not in enabled_disk_templates:

    return

  if constants.DT_PLAIN not in enabled_disk_templates:

    return

  drbd_usermode_helper = qa_config.get("drbd-usermode-helper", "/bin/true")

  bogus_usermode_helper = "/tmp/pinkbunny"

  for command, fail in [

    (["gnt-cluster", "modify",

      "--enabled-disk-templates=%s" % constants.DT_DRBD8,

      "--ipolicy-disk-templates=%s" % constants.DT_DRBD8], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper=%s" % drbd_usermode_helper], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper=%s" % bogus_usermode_helper], True),

    # unsetting helper when DRBD is enabled should not work

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper="], True),

    (["gnt-cluster", "modify",

      "--enabled-disk-templates=%s" % constants.DT_PLAIN,

      "--ipolicy-disk-templates=%s" % constants.DT_PLAIN], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper="], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper=%s" % drbd_usermode_helper], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper=%s" % drbd_usermode_helper,

      "--enabled-disk-templates=%s" % constants.DT_DRBD8,

      "--ipolicy-disk-templates=%s" % constants.DT_DRBD8], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper=",

      "--enabled-disk-templates=%s" % constants.DT_PLAIN,

      "--ipolicy-disk-templates=%s" % constants.DT_PLAIN], False),

    (["gnt-cluster", "modify",

      "--drbd-usermode-helper=%s" % drbd_usermode_helper,

      "--enabled-disk-templates=%s" % constants.DT_DRBD8,

      "--ipolicy-disk-templates=%s" % constants.DT_DRBD8], False),

    ]:

    AssertCommand(command, fail=fail)

  _RestoreEnabledDiskTemplates()

def _TestClusterModifyDiskTemplatesArguments(default_disk_template):

  """Tests argument handling of 'gnt-cluster modify' with respect to

     the parameter '--enabled-disk-templates'. This test is independent

     of instances.

  """

  _RestoreEnabledDiskTemplates()

  # bogus templates

  AssertCommand(["gnt-cluster", "modify",

                 "--enabled-disk-templates=pinkbunny"],

                fail=True)

  # duplicate entries do no harm

  AssertCommand(

    ["gnt-cluster", "modify",

     "--enabled-disk-templates=%s,%s" %

      (default_disk_template, default_disk_template),

     "--ipolicy-disk-templates=%s" % default_disk_template],

    fail=False)

def _TestClusterModifyDiskTemplatesVgName(enabled_disk_templates):

  """Tests argument handling of 'gnt-cluster modify' with respect to

     the parameter '--enabled-disk-templates' and '--vg-name'. This test is

     independent of instances.

  """

  if not utils.IsLvmEnabled(enabled_disk_templates):

    # These tests only make sense if lvm is enabled for QA

    return

  # determine an LVM and a non-LVM disk template for the tests

  non_lvm_template = _GetOtherEnabledDiskTemplate(constants.DTS_LVM,

                                                  enabled_disk_templates)

  lvm_template = list(set(enabled_disk_templates) & constants.DTS_LVM)[0]

  vgname = qa_config.get("vg-name", constants.DEFAULT_VG)

  # Clean start: unset volume group name, disable lvm storage

  AssertCommand(

    ["gnt-cluster", "modify",

     "--enabled-disk-templates=%s" % non_lvm_template,

     "--ipolicy-disk-templates=%s" % non_lvm_template,

     "--vg-name="],