Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, **kwargs):
"""
:keyword device: path to the underlying device (required)
:keyword uuid: this format's UUID
:keyword exists: whether this is an existing format
:type exists: bool
"""
log_method_call(self, **kwargs)
DeviceFormat.__init__(self, **kwargs)
# Initialize the attribute that will hold the block object.
self._member = None
def teardown(self, recursive=None):
""" Close, or tear down, a device. """
log_method_call(self, self.name, status=self.status,
controllable=self.controllable)
if not self._pre_teardown(recursive=recursive):
if recursive:
self.teardown_parents(recursive=recursive)
return
self._teardown(recursive=recursive)
self._post_teardown(recursive=recursive)
def _create(self):
""" Create the device. """
log_method_call(self, self.name, status=self.status)
pv_list = [pv.path for pv in self.parents]
blockdev.lvm.vgcreate(self.name, pv_list, self.peSize)
:type escrow_cert: str
:keyword add_backup_passphrase: generate a backup passphrase?
:type add_backup_passphrase: bool.
:keyword min_luks_entropy: minimum entropy in bits required for
format creation
:type min_luks_entropy: int
.. note::
The 'device' kwarg is required for existing formats. For non-
existent formats, it is only necessary that the :attr:`device`
attribute be set before the :meth:`create` method runs. Note
that you can specify the device at the last moment by specifying
it via the 'device' kwarg to the :meth:`create` method.
"""
log_method_call(self, **kwargs)
DeviceFormat.__init__(self, **kwargs)
self.cipher = kwargs.get("cipher")
self.key_size = kwargs.get("key_size")
self.mapName = kwargs.get("name")
if not self.exists and not self.cipher:
self.cipher = "aes-xts-plain64"
if not self.key_size:
# default to the max (512 bits) for aes-xts
self.key_size = 512
# FIXME: these should both be lists, but managing them will be a pain
self.__passphrase = kwargs.get("passphrase")
self._key_file = kwargs.get("key_file")
self.escrow_cert = kwargs.get("escrow_cert")
self.add_backup_passphrase = kwargs.get("add_backup_passphrase", False)
def add_child(self, child):
""" Increment the child counter for this device. """
log_method_call(self, name=self.name, child=child._name, kids=len(self.children))
if child in self._children:
raise ValueError("child is already accounted for")
self._children.append(child)
def handleUdevMDMemberFormat(self, info, device):
# pylint: disable=unused-argument
log_method_call(self, name=device.name, type=device.format.type)
md_info = blockdev.md.examine(device.path)
# Use mdadm info if udev info is missing
md_uuid = md_info.uuid
device.format.mdUuid = device.format.mdUuid or md_uuid
md_array = self.getDeviceByUuid(device.format.mdUuid, incomplete=True)
if md_array:
md_array.parents.append(device)
else:
# create the array with just this one member
# level is reported as, eg: "raid1"
md_level = md_info.level
md_devices = md_info.num_devices
if md_level is None:
def addUdevDevice(self, info, updateOrigFmt=False):
"""
:param :class:`pyudev.Device` info: udev info for the device
:keyword bool updateOrigFmt: update original format unconditionally
If a device is added to the tree based on info its original format
will be saved after the format has been detected. If the device
that corresponds to info is already in the tree, its original format
will not be updated unless updateOrigFmt is True.
"""
name = udev.device_get_name(info)
log_method_call(self, name=name, info=pprint.pformat(dict(info)))
uuid = udev.device_get_uuid(info)
sysfs_path = udev.device_get_sysfs_path(info)
# make sure this device was not scheduled for removal and also has not
# been hidden
removed = [a.device for a in self.devicetree.actions.find(
action_type="destroy",
object_type="device")]
for ignored in removed + self.devicetree._hidden:
if (sysfs_path and ignored.sysfsPath == sysfs_path) or \
(uuid and uuid in (ignored.uuid, ignored.format.uuid)):
if ignored in removed:
reason = "removed"
else:
reason = "hidden"
def create(self, *args, **kwargs):
log_method_call(self, device=self.device,
type=self.type, status=self.status)
raise DMRaidMemberError("creation of dmraid members is non-sense")
def updateSysfsPath(self):
""" Update this device's sysfs path. """
log_method_call(self, self.name, status=self.status)
if not self.exists:
raise errors.DeviceError("device has not been created", self.name)
self.sysfsPath = ''
def __init__(self, **kwargs):
"""
:keyword device: path to block device node
:keyword uuid: this member device's uuid
:keyword exists: whether this is an existing format
:type exists: bool
:keyword md_uuid: the uuid of the array this device belongs to
.. note::
The 'device' kwarg is required for existing formats.
"""
log_method_call(self, **kwargs)
DeviceFormat.__init__(self, **kwargs)
self.md_uuid = kwargs.get("md_uuid")
self.biosraid = kwargs.get("biosraid")