yield fields
+def _LvmPvGetAllocatable(attr):
+ """Determines whether LVM PV is allocatable.
+
+ @rtype: bool
+
+ """
+ if attr:
+ return (attr[0] == "a")
+ else:
+ logging.warning("Invalid PV attribute: %r", attr)
+ return False
+
+
class LvmPvStorage(_LvmBase): # pylint: disable-msg=W0223
"""LVM Physical Volume storage unit.
"""
- @staticmethod
- def _GetAllocatable(attr):
- if attr:
- return (attr[0] == "a")
- else:
- logging.warning("Invalid PV attribute: %r", attr)
- return False
-
LIST_COMMAND = "pvs"
# Make sure to update constants.VALID_STORAGE_FIELDS when changing field
(constants.SF_SIZE, ["pv_size"], _ParseSize),
(constants.SF_USED, ["pv_used"], _ParseSize),
(constants.SF_FREE, ["pv_free"], _ParseSize),
- (constants.SF_ALLOCATABLE, ["pv_attr"], _GetAllocatable),
+ (constants.SF_ALLOCATABLE, ["pv_attr"], _LvmPvGetAllocatable),
]
def _SetAllocatable(self, name, allocatable):
else:
test_allocatable = ["yes", "no"]
- if (constants.SF_ALLOCATABLE in
- constants.MODIFIABLE_STORAGE_FIELDS.get(storage_type, [])):
- assert_fn = AssertEqual
- else:
+ fail = (constants.SF_ALLOCATABLE not in
+ constants.MODIFIABLE_STORAGE_FIELDS.get(storage_type, []))
+
+ if fail:
assert_fn = AssertNotEqual
+ else:
+ assert_fn = AssertEqual
for i in test_allocatable:
cmd = ["gnt-node", "modify-storage", "--allocatable", i,
assert_fn(StartSSH(master["primary"],
utils.ShellQuoteArgs(cmd)).wait(), 0)
+ # 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
cmd = ["gnt-node", "repair-storage", node_name, storage_type, st_name]