def TestNodeStorage():
  master = qa_config.GetMasterNode()

  # FIXME: test all storage_types in constants.STORAGE_TYPES
  # as soon as they are implemented.
  enabled_storage_types = qa_config.GetEnabledStorageTypes()
  testable_storage_types = list(set(enabled_storage_types).intersection(
      set([constants.ST_FILE, constants.ST_LVM_VG, constants.ST_LVM_PV])))

  for storage_type in testable_storage_types:

    cmd = ["gnt-node", "list-storage", "--storage-type", storage_type]

    # Test simple list
    AssertCommand(cmd)

    # Test all storage fields
    cmd = ["gnt-node", "list-storage", "--storage-type", storage_type,
           "--output=%s" % ",".join(list(constants.VALID_STORAGE_FIELDS))]
    AssertCommand(cmd)

    # Get list of valid storage devices
    cmd = ["gnt-node", "list-storage", "--storage-type", storage_type,
           "--output=node,name,allocatable", "--separator=|",
           "--no-headers"]
    output = qa_utils.GetCommandOutput(master.primary,
                                       utils.ShellQuoteArgs(cmd))

    # Test with up to two devices
    testdevcount = 2

    for line in output.splitlines()[:testdevcount]:
      (node_name, st_name, st_allocatable) = line.split("|")

      # Dummy modification without any changes
      cmd = ["gnt-node", "modify-storage", node_name, storage_type, st_name]
      AssertCommand(cmd)

      # Make sure we end up with the same value as before
      if st_allocatable.lower() == "y":
        test_allocatable = ["no", "yes"]
      else:
        test_allocatable = ["yes", "no"]

      fail = (constants.SF_ALLOCATABLE not in
              constants.MODIFIABLE_STORAGE_FIELDS.get(storage_type, []))

      for i in test_allocatable:
        AssertCommand(["gnt-node", "modify-storage", "--allocatable", i,
                       node_name, storage_type, st_name], fail=fail)

        # Verify list output
        cmd = ["gnt-node", "list-storage", "--storage-type", storage_type,
               "--output=name,allocatable", "--separator=|",
               "--no-headers", node_name]
        listout = qa_utils.GetCommandOutput(master.primary,
                                            utils.ShellQuoteArgs(cmd))
        for line in listout.splitlines():
          (vfy_name, vfy_allocatable) = line.split("|")
          if vfy_name == st_name and not fail:
            AssertEqual(vfy_allocatable, i[0].upper())
          else:
            AssertEqual(vfy_allocatable, st_allocatable)

      # Test repair functionality
      fail = (constants.SO_FIX_CONSISTENCY not in
              constants.VALID_STORAGE_OPERATIONS.get(storage_type, []))
      AssertCommand(["gnt-node", "repair-storage", node_name,
                     storage_type, st_name], fail=fail)
update notification
dataset/ETHPy150Open ganeti/ganeti/qa/qa_node.py/TestNodeStorage--crossover with--dataset/ETHPy150Open nitely/Spirit/spirit/topic/notification/tests.py/TopicNotificationFormTest.test_notification