# check nics' parameter names
for nic in self.op.nics:
utils.ForceDictType(nic, constants.INIC_PARAMS_TYPES)
+ # check that NIC's parameters names are unique and valid
+ utils.ValidateDeviceNames("NIC", self.op.nics)
+
+ # check that disk's names are unique and valid
+ utils.ValidateDeviceNames("disk", self.op.disks)
cluster = self.cfg.GetClusterInfo()
if not self.op.disk_template in cluster.enabled_disk_templates:
" (%d), cannot add more" % constants.MAX_NICS,
errors.ECODE_STATE)
+ def _PrepareDiskMod(_, disk, params, __):
+ disk.name = params.get(constants.IDISK_NAME, None)
+
# Verify disk changes (operating on a copy)
- disks = instance.disks[:]
- ApplyContainerMods("disk", disks, None, self.diskmod, None, None, None)
+ disks = copy.deepcopy(instance.disks)
+ ApplyContainerMods("disk", disks, None, self.diskmod, None, _PrepareDiskMod,
+ None)
+ utils.ValidateDeviceNames("disk", disks)
if len(disks) > constants.MAX_DISKS:
raise errors.OpPrereqError("Instance has too many disks (%d), cannot add"
" more" % constants.MAX_DISKS,
nics = [nic.Copy() for nic in instance.nics]
ApplyContainerMods("NIC", nics, self._nic_chgdesc, self.nicmod,
self._CreateNewNic, self._ApplyNicMods, None)
+ # Verify that NIC names are unique and valid
+ utils.ValidateDeviceNames("NIC", nics)
self._new_nics = nics
ispec[constants.ISPEC_NIC_COUNT] = len(self._new_nics)
else:
"""
return [val for val in items if not self.Matches(val)]
+
+
+def ValidateDeviceNames(kind, container):
+ """Validate instance device names.
+
+ Check that a device container contains only unique and valid names.
+
+ @type kind: string
+ @param kind: One-word item description
+ @type container: list
+ @param container: Container containing the devices
+
+ """
+
+ valid = []
+ for device in container:
+ if isinstance(device, dict):
+ if kind == "NIC":
+ name = device.get(constants.INIC_NAME, None)
+ elif kind == "disk":
+ name = device.get(constants.IDISK_NAME, None)
+ else:
+ raise errors.OpPrereqError("Invalid container kind '%s'" % kind,
+ errors.ECODE_INVAL)
+ else:
+ name = device.name
+ # Check that a device name is not the UUID of another device
+ valid.append(device.uuid)
+
+ try:
+ int(name)
+ except (ValueError, TypeError):
+ pass
+ else:
+ raise errors.OpPrereqError("Invalid name '%s'. Purely numeric %s names"
+ " are not allowed" % (name, kind),
+ errors.ECODE_INVAL)
+
+ if name is not None and name.lower() != constants.VALUE_NONE:
+ if name in valid:
+ raise errors.OpPrereqError("%s name '%s' already used" % (kind, name),
+ errors.ECODE_NOTUNIQUE)
+ else:
+ valid.append(name)
some_keys, self.defaults)
+class TestValidateDeviceNames(unittest.TestCase):
+ def testEmpty(self):
+ utils.ValidateDeviceNames("NIC", [])
+ utils.ValidateDeviceNames("disk", [])
+
+ def testNoName(self):
+ nics = [{}, {}]
+ utils.ValidateDeviceNames("NIC", nics)
+
+ def testInvalidName(self):
+ self.assertRaises(errors.OpPrereqError, utils.ValidateDeviceNames,
+ "disk", [{constants.IDISK_NAME: "42"}])
+ self.assertRaises(errors.OpPrereqError, utils.ValidateDeviceNames,
+ "NIC", [{constants.INIC_NAME: "42"}])
+
+ def testUsedName(self):
+ disks = [{constants.IDISK_NAME: "name1"}, {constants.IDISK_NAME: "name1"}]
+ self.assertRaises(errors.OpPrereqError, utils.ValidateDeviceNames,
+ "disk", disks)
+
+
if __name__ == "__main__":
testutils.GanetiTestProgram()