diff --git a/blivet/blivet.py b/blivet/blivet.py index 61b0b309f..e1c0042d8 100644 --- a/blivet/blivet.py +++ b/blivet/blivet.py @@ -56,6 +56,9 @@ log = logging.getLogger("blivet") +FSTAB_PATH = "/etc/fstab" + + @six.add_metaclass(SynchronizedMeta) class Blivet(object): @@ -74,7 +77,7 @@ def __init__(self): # fstab write location purposedly set to None. It has to be overriden # manually when using blivet. - self.fstab = FSTabManager(src_file="/etc/fstab", dest_file=None) + self.fstab = FSTabManager(src_file=FSTAB_PATH, dest_file=None) self._short_product_name = 'blivet' diff --git a/blivet/fstab.py b/blivet/fstab.py index e5d9ce9c8..319b53121 100644 --- a/blivet/fstab.py +++ b/blivet/fstab.py @@ -1,10 +1,5 @@ import os -import gi - -gi.require_version("BlockDev", "2.0") -from gi.repository import BlockDev as blockdev - from libmount import Table, Fs, MNT_ITER_FORWARD from libmount import Error as LibmountException @@ -12,18 +7,247 @@ log = logging.getLogger("blivet") -def parser_errcb(tb, fname, line): # pylint: disable=unused-argument - print("{:s}:{:d}: parse error".format(fname, line)) - return 1 +class FSTabEntry(object): + """ One processed line of fstab + """ + + def __init__(self, spec=None, file=None, vfstype=None, mntops=None, + freq=None, passno=None, comment=None, *, entry=None): + + # "*" in arguments means that every following parameter can be used + # only with its name (e.g. "entry=Fs()") + + if entry is None: + self._entry = Fs() + else: + self._entry = entry + + if spec is not None: + self.spec = spec + if file is not None: + self.file = file + if vfstype is not None: + self.vfstype = vfstype + if mntops is not None: + self.mntops = mntops + if freq is not None: + self.freq = freq + if passno is not None: + self.passno = passno + if comment is not None: + self.comment = comment + + self._none_to_empty() + + def __repr__(self): + _comment = "" + if self._entry.comment not in ("", None): + _comment = "%s\n" % self._entry.comment + _line = "%s\t%s\t%s\t%s\t%s\t%s\t" % (self._entry.source, self._entry.target, self._entry.fstype, + self._entry.options, self._entry.freq, self._entry.passno) + return _comment + _line + + def __eq__(self, other): + if not isinstance(other, FSTabEntry): + return False + + if self._entry.source != other._entry.source: + return False + + if self._entry.target != other._entry.target: + return False + + if self._entry.fstype != other._entry.fstype: + return False + + if self._entry.options != other._entry.options: + return False + + if self._entry.freq != other._entry.freq: + return False + + if self._entry.passno != other._entry.passno: + return False + + return True + + def _none_to_empty(self): + """ Workaround function that internally replaces all None values with empty strings. + Reason: While libmount.Fs() initializes with parameters set to None, it does not + allow to store None as a valid value, blocking all value resets. + """ + + affected_params = [self._entry.source, + self._entry.target, + self._entry.fstype, + self._entry.options, + self._entry.comment] + + for param in affected_params: + if param is None: + param = "" + + @property + def entry(self): + return self._entry + + @entry.setter + def entry(self, value): + """ Setter for the whole internal entry value + + :param value: fstab entry + :type value: :class: `libmount.Fs` + """ + self._entry = value + + @property + def spec(self): + return self._entry.source if self._entry.source != "" else None + + @spec.setter + def spec(self, value): + self._entry.source = value if value is not None else "" + + @property + def file(self): + return self._entry.target if self._entry.target != "" else None + @file.setter + def file(self, value): + self._entry.target = value if value is not None else "" -class FSTabManager(): - # Read, write and modify fstab file - # This class is meant to work even without blivet. - # However some of its methods require blivet and will not function without - # it. These methods will import what they need when they are run. + @property + def vfstype(self): + return self._entry.fstype if self._entry.fstype != "" else None + + @vfstype.setter + def vfstype(self, value): + self._entry.fstype = value if value is not None else "" + + @property + def mntops(self): + """ Return mount options + + :returns: list of mount options or None when not set + :rtype: list of str + """ + + if self._entry.options == "": + return None + + return self._entry.options.split(',') + + def get_raw_mntops(self): + """ Return mount options + + :returns: comma separated string of mount options or None when not set + :rtype: str + """ + + return self._entry.options if self._entry.options != "" else None + + @mntops.setter + def mntops(self, values): + """ Set new mount options from the list of strings + + :param values: mount options (see man fstab(5) fs_mntops) + :type values: list of str + """ + + # libmount.Fs() internally stores options as a comma separated string + if values is None: + self._entry.options = "" + else: + self._entry.options = ','.join([x for x in values if x != ""]) + + def mntops_add(self, values): + """ Append new mount options to already existing ones + + :param values: mount options (see man fstab(5) fs_mntops) + :type values: list of str + """ + + self._entry.append_options(','.join([x for x in values if x != ""])) + + @property + def freq(self): + return self._entry.freq + + @freq.setter + def freq(self, value): + self._entry.freq = value + + @property + def passno(self): + return self._entry.passno + + @passno.setter + def passno(self, value): + self._entry.passno = value + + @property + def comment(self): + return self._entry.comment if self._entry.comment != "" else None + + @comment.setter + def comment(self, value): + if value is None: + self._entry.comment = "" + return + self._entry.comment = value + + def is_valid(self): + """ Verify that this instance has enough data for valid fstab entry + + :returns: False if any of the listed values is not set; otherwise True + :rtype: bool + """ + items = [self.spec, self.file, self.vfstype, self.mntops, self.freq, self.passno] + + # (Property getters replace empty strings with None) + return not any(x is None for x in items) + + +class FSTabManagerIterator(object): + """ Iterator class for FSTabManager + Iteration over libmount Table entries is weird - only one iterator can run at a time. + This class purpose is to mitigate that. + """ + + def __init__(self, fstab): + # To avoid messing up the libmount Table iterator, set up independent entry list + entry = fstab._table.next_fs() + self.entries = [] + while entry: + self.entries.append(entry) + entry = fstab._table.next_fs() + self.cur_index = 0 + + def __iter__(self): + return self + + def __next__(self): + if self.cur_index < len(self.entries): + self.cur_index += 1 + return FSTabEntry(entry=self.entries[self.cur_index - 1]) + raise StopIteration + + +class FSTabManager(object): + """ Read, write and modify fstab file. + This class is meant to work even without blivet. + However some of its methods require blivet and will not function without it. + """ def __init__(self, src_file=None, dest_file=None): + """ Initialize internal table; load the file if specified + + :keyword src_file: Path to fstab file which will be read + :type src_file: str + :keyword dest_file: Path to file which will be overwritten with results + :type src_file: str + """ + self._table = Table() # Space for parsed fstab contents self.src_file = src_file @@ -38,80 +262,84 @@ def __deepcopy__(self, memo): clone._table.enable_comments(True) entry = self._table.next_fs() - entries = [entry] + entries = [] while entry: entries.append(entry) entry = self._table.next_fs() - # Libmount does not allow to simply use clone._table.add_fs(entry), so... for entry in entries: - new_entry = Fs() - new_entry.source = entry.source - new_entry.target = entry.target - new_entry.fstype = entry.fstype - new_entry.append_options(entry.options) - new_entry.freq = entry.freq - new_entry.passno = entry.passno - if entry.comment is not None: - new_entry.comment = entry.comment + new_entry = self._copy_fs_entry(entry) clone._table.add_fs(new_entry) return clone - def _get_containing_device(self, path, devicetree): - # Return the device that a path resides on - if not os.path.exists(path): - return None - - st = os.stat(path) - major = os.major(st.st_dev) - minor = os.minor(st.st_dev) - link = "/sys/dev/block/%s:%s" % (major, minor) - if not os.path.exists(link): - return None - - try: - device_name = os.path.basename(os.readlink(link)) - except Exception: # pylint: disable=broad-except - log.error("failed to find device name for path %s", path) - return None - - if device_name.startswith("dm-"): - # have I told you lately that I love you, device-mapper? - # (code and comment copied from anaconda, kept for entertaining purposes) - device_name = blockdev.dm.name_from_node(device_name) - - return devicetree.get_device_by_name(device_name) - - def _from_device(self, device): - # Return the list of fstab options obtained from device - # *(result) of this method is meant to be used as a parameter in other methods - - spec = getattr(device, 'fstab_spec', None) - - file = None + def __repr__(self): + entry = self._table.next_fs() + entries_str = "" + while entry: + entries_str += repr(FSTabEntry(entry=entry)) + '\n' + entry = self._table.next_fs() + return entries_str + + def __iter__(self): + return FSTabManagerIterator(self) + + def _copy_fs_entry(self, entry): + """ Create copy of libmount.Fs() + """ + # oh well... + new_entry = Fs() + new_entry.source = entry.source + new_entry.target = entry.target + new_entry.fstype = entry.fstype + new_entry.append_options(entry.options) + new_entry.freq = entry.freq + new_entry.passno = entry.passno + if entry.comment is not None: + new_entry.comment = entry.comment + return new_entry + + def _parser_errcb(self, tb, fname, line): # pylint: disable=unused-argument + """ Libmount interface error reporting function + """ + log.error("Fstab parse error '%s:%s'", fname, line) + return 1 + + def entry_from_device(self, device): + """ Generate FSTabEntry object based on given blivet Device + + :keyword device: device to process + :type device: :class: `blivet.devices.StorageDevice` + :returns: fstab entry object based on device + :rtype: :class: `FSTabEntry` + """ + + entry = FSTabEntry() + entry.spec = getattr(device, "fstab_spec", None) + entry.file = None if device.format.mountable: - file = device.format.mountpoint - if device.format.type == 'swap': - file = 'swap' + entry.file = device.format.mountpoint + if device.format.type == "swap": + entry.file = "swap" - vfstype = device.format.type - mntops = None + entry.vfstype = device.format.type - return spec, file, vfstype, mntops + return entry - def _from_entry(self, entry): - return entry.source, entry.target, entry.fstype, ','.join(entry.options) + def read(self, src_file=""): + """ Read the fstab file from path stored in self.src_file. Setting src_file parameter overrides + it with new value. - def read(self, src_file=''): - # Read fstab file + :keyword src_file: When set, reads fstab from path specified in it + :type src_file: str + """ # Reset table self._table = Table() self._table.enable_comments(True) # resolve which file to read - if src_file == '': + if src_file == "": if self.src_file is None: # No parameter given, no internal value return @@ -120,226 +348,289 @@ def read(self, src_file=''): else: self.src_file = src_file - self._table.errcb = parser_errcb + self._table.errcb = self._parser_errcb self._table.parse_fstab(self.src_file) - def find_device_by_specs(self, blivet, spec, file_dummy=None, vfstype_dummy=None, mntops=""): # pylint: disable=unused-argument - # Parse an fstab entry for a device, return the corresponding device, - # return None if not found - # dummy arguments allow using result of _from_device/_from_entry as a parameter of this method - return blivet.devicetree.resolve_device(spec, options=mntops) - - def find_device_by_entry(self, blivet, entry): - args = self._from_entry(entry) - return self.find_device_by_specs(blivet, *args) - - def get_device_by_specs(self, blivet, spec, file, vfstype, mntops): - # Parse an fstab entry for a device, return the corresponding device, - # create new one if it does not exist + def find_device(self, blivet, spec=None, mntops=None, blkid_tab=None, crypt_tab=None, *, entry=None): + """ Find a blivet device, based on spec or entry. Mount options can be used to refine the search. + If both entry and spec/mntops are given, spec/mntops are prioritized over entry values. + + :param blivet: Blivet instance with populated devicetree + :type blivet: :class: `Blivet` + :keyword spec: searched device specs (see man fstab(5) fs_spec) + :type spec: str + :keyword mntops: list of mount option strings (see man fstab(5) fs_mntops) + :type mnops: list + :keyword blkid_tab: Blkidtab object + :type blkid_tab: :class: `BlkidTab` + :keyword crypt_tab: Crypttab object + :type crypt_tab: :class: `CryptTab` + :keyword entry: fstab entry with its spec (and mntops) filled as an alternative input type + :type: :class: `FSTabEntry` + :returns: found device or None + :rtype: :class: `~.devices.StorageDevice` or None + """ + + _spec = spec or (entry.spec if entry is not None else None) + _mntops = mntops or (entry.mntops if entry is not None else None) + _mntops_str = ",".join(_mntops) if mntops is not None else None + + return blivet.devicetree.resolve_device(_spec, options=_mntops_str, blkid_tab=blkid_tab, crypt_tab=crypt_tab) + + def get_device(self, blivet, spec=None, file=None, vfstype=None, + mntops=None, blkid_tab=None, crypt_tab=None, *, entry=None): + """ Parse an fstab entry for a device and return the corresponding device from the devicetree. + If not found, try to create a new device based on given values. + Raises UnrecognizedFSTabError in case of invalid or incomplete data. + + :param blivet: Blivet instance with populated devicetree + :type blivet: :class: `Blivet` + :keyword spec: searched device specs (see man fstab(5) fs_spec) + :type spec: str + :keyword mntops: list of mount option strings (see man fstab(5) fs_mntops) + :type mnops: list + :keyword blkid_tab: Blkidtab object + :type blkid_tab: :class: `BlkidTab` + :keyword crypt_tab: Crypttab object + :type crypt_tab: :class: `CryptTab` + :keyword entry: fstab entry with its values filled as an alternative input type + :type: :class: `FSTabEntry` + :returns: found device + :rtype: :class: `~.devices.StorageDevice` + """ from blivet.formats import get_format - from blivet.devices import DirectoryDevice, NFSDevice, FileDevice, NoDevice - from blivet.formats import get_device_format_class + from blivet.devices import DirectoryDevice, FileDevice + from blivet.errors import UnrecognizedFSTabEntryError - # no sense in doing any legwork for a noauto entry - if "noauto" in mntops.split(","): - raise ValueError("Unrecognized fstab entry value 'noauto'") + _spec = spec or (entry.spec if entry is not None else None) + _mntops = mntops or (entry.mntops if entry is not None else None) + _mntops_str = ",".join(_mntops) if mntops is not None else None # find device in the tree - device = blivet.devicetree.resolve_device(spec, options=mntops) - - if device: - # fall through to the bottom of this block - pass - elif ":" in spec and vfstype.startswith("nfs"): - # NFS -- preserve but otherwise ignore - device = NFSDevice(spec, - fmt=get_format(vfstype, - exists=True, - device=spec)) - elif spec.startswith("/") and vfstype == "swap": - # swap file - device = FileDevice(spec, - parents=self._get_containing_device(spec, blivet.devicetree), - fmt=get_format(vfstype, - device=spec, - exists=True), - exists=True) - elif vfstype == "bind" or "bind" in mntops: - # bind mount... set vfstype so later comparison won't - # turn up false positives - vfstype = "bind" - - # This is probably not going to do anything useful, so we'll - # make sure to try again from FSSet.mount_filesystems. The bind - # mount targets should be accessible by the time we try to do - # the bind mount from there. - parents = self._get_containing_device(spec, blivet.devicetree) - device = DirectoryDevice(spec, parents=parents, exists=True) - device.format = get_format("bind", - device=device.path, - exists=True) - elif file in ("/proc", "/sys", "/dev/shm", "/dev/pts", - "/sys/fs/selinux", "/proc/bus/usb", "/sys/firmware/efi/efivars"): - # drop these now -- we'll recreate later - return None - else: - # nodev filesystem -- preserve or drop completely? - fmt = get_format(vfstype) - fmt_class = get_device_format_class("nodev") - if spec == "none" or \ - (fmt_class and isinstance(fmt, fmt_class)): - device = NoDevice(fmt=fmt) + device = blivet.devicetree.resolve_device(_spec, options=_mntops_str, blkid_tab=blkid_tab, crypt_tab=crypt_tab) if device is None: - log.error("failed to resolve %s (%s) from fstab", spec, - vfstype) - raise ValueError() + if vfstype == "swap": + # swap file + device = FileDevice(_spec, + parents=blivet.devicetree.resolve_device(_spec), + fmt=get_format(vfstype, device=_spec, exists=True), + exists=True) + elif vfstype == "bind" or (_mntops is not None and "bind" in _mntops): + # bind mount... set vfstype so later comparison won't + # turn up false positives + vfstype = "bind" + + parents = blivet.devicetree.resolve_device(_spec) + device = DirectoryDevice(_spec, parents=parents, exists=True) + device.format = get_format("bind", device=device.path, exists=True) + + if device is None: + raise UnrecognizedFSTabEntryError("Could not resolve entry %s %s" % (_spec, vfstype)) - device.setup() fmt = get_format(vfstype, device=device.path, exists=True) if vfstype != "auto" and None in (device.format.type, fmt.type): - log.info("Unrecognized filesystem type for %s (%s)", - device.name, vfstype) - device.teardown() - raise ValueError() - - # make sure, if we're using a device from the tree, that - # the device's format we found matches what's in the fstab - ftype = getattr(fmt, "mount_type", fmt.type) - dtype = getattr(device.format, "mount_type", device.format.type) - if hasattr(fmt, "test_mount") and vfstype != "auto" and ftype != dtype: - log.info("fstab says %s at %s is %s", dtype, file, ftype) - if fmt.test_mount(): # pylint: disable=no-member - device.format = fmt - else: - device.teardown() - log.info("There is an entry in your fstab file that contains " - "unrecognized file system type. The file says that " - "%s at %s is %s.", dtype, file, ftype) - return None + raise UnrecognizedFSTabEntryError("Unrecognized filesystem type for %s: '%s'" % (_spec, vfstype)) if hasattr(device.format, "mountpoint"): device.format.mountpoint = file - device.format.options = mntops + device.format.options = _mntops return device - def get_device_by_entry(self, blivet, entry): - args = self._from_entry(entry) - return self.get_device_by_specs(blivet, *args) - - def add_entry_by_specs(self, spec, file, vfstype, mntops, freq=None, passno=None, comment=None): - # Add new entry into the table + def add_entry(self, spec=None, file=None, vfstype=None, mntops=None, + freq=None, passno=None, comment=None, *, entry=None): + """ Add a new entry into the table + If both entry and other values are given, these values are prioritized over entry values. + If mntops/freq/passno is not set uses their respective default values. + + :keyword spec: device specs (see man fstab(5) fs_spec) + :type spec: str + :keyword file: device mount path (see man fstab(5) fs_file) + :type file: str + :keyword vfstype: device file system type (see man fstab(5) fs_vfstype) + :type vfstype: str + :keyword mntops: list of mount option strings (see man fstab(5) fs_mntops) + :type mnops: list + :keyword freq: whether to dump the filesystem (see man fstab(5) fs_freq) + :type freq: int + :keyword passno: fsck order or disable fsck if 0 (see man fstab(5) fs_passno) + :type passno: int + :keyword comment: multiline comment added to fstab before entry; each line will be prefixed with "# " + :type comment: str + :keyword entry: fstab entry as an alternative input type + :type: :class: `FSTabEntry` + """ # Default mount options if mntops is None: - mntops = 'defaults' - - # Whether the fs should be dumped by dump(8), defaults to 0 - if freq is None: - freq = 0 - - # Order of fsck run at boot time. '/' should have 1, other 2, defaults to 0 - if passno is None: - if file is None: - file = '' - if file == '/': - passno = 1 - elif file.startswith('/boot'): - passno = 2 - else: - passno = 0 + mntops = ['defaults'] - entry = Fs() + # Use existing FSTabEntry or create a new one + _entry = entry or FSTabEntry() - entry.source = spec - entry.target = file - entry.fstype = vfstype - entry.append_options(mntops) - entry.freq = freq - entry.passno = passno + if spec is not None: + _entry.spec = spec + + if file is not None: + _entry.file = file + + if vfstype is not None: + _entry.vfstype = vfstype + + if mntops is not None: + _entry.mntops = mntops + elif _entry.mntops is None: + _entry.mntops = ['defaults'] + + if freq is not None: + # Whether the fs should be dumped by dump(8) (default: 0, i.e. no) + _entry.freq = freq + elif _entry.freq is None: + _entry.freq = 0 + + if passno is not None: + _entry.passno = freq + elif _entry.passno is None: + # 'passno' represents order of fsck run at the boot time (default: 0, i.e. disabled). + # '/' should have 1, other checked should have 2 + if _entry.file == '/': + _entry.passno = 1 + elif _entry.file.startswith('/boot'): + _entry.passno = 2 + else: + _entry.passno = 0 if comment is not None: # add '# ' at the start of any comment line and newline to the end of comment modif_comment = '# ' + comment.replace('\n', '\n# ') + '\n' - entry.comment = modif_comment + _entry.comment = modif_comment - self._table.add_fs(entry) + self._table.add_fs(_entry.entry) - def add_entry_by_device(self, device): - args = self._from_device(device) - return self.add_entry_by_specs(*args) + def remove_entry(self, spec=None, file=None, *, entry=None): + """ Find and remove entry from fstab based on spec/file. + If both entry and spec/file are given, spec/file are prioritized over entry values. - def remove_entry_by_specs(self, spec, file, vfstype=None, mntops=""): - fs = self.find_entry_by_specs(spec, file, vfstype, mntops) - if fs: - self._table.remove_fs(fs) + :keyword spec: device specs (see man fstab(5) fs_spec) + :type spec: str + :keyword file: device mount path (see man fstab(5) fs_file) + :type file: str + :keyword entry: fstab entry as an alternative input type + :type: :class: `FSTabEntry` + """ - def remove_entry_by_device(self, device): - args = self._from_device(device) - return self.remove_entry_by_specs(*args) + fs = self.find_entry(spec, file, entry=entry) + if fs: + self._table.remove_fs(fs.entry) def write(self, dest_file=None): - # Commit the self._table into the file. + """ Commit the self._table into the self._dest_file. Setting dest_file parameter overrides + writing path with its value. + + :keyword dest_file: When set, writes fstab to the path specified in it + :type dest_file: str + """ + if dest_file is None: dest_file = self.dest_file if dest_file is None: log.info("Fstab path for writing was not specified") return - if os.path.exists(dest_file): - self._table.replace_file(dest_file) - else: - # write will fail if underlying directories do not exist - self._table.write_file(dest_file) + # Output sanity check to prevent saving an incomplete file entries. + # Invalid lines should be skipped && The whole table is written at once && + # Incomplete lines need to be preserved in the object. => Create second table, modify and write it. - def find_entry_by_specs(self, spec, file, vfstype_dummy=None, mntops_dummy=None): # pylint: disable=unused-argument - # Return line of self._table with given spec or file - # dummy arguments allow using result of _from_device/_from_entry as a parameter of this method + clean_table = Table() + clean_table.enable_comments(True) - entry = None + entry = self._table.next_fs() + while entry: + if FSTabEntry(entry=entry).is_valid(): + new_entry = self._copy_fs_entry(entry) + clean_table.add_fs(new_entry) + else: + log.warning("Fstab entry: '%s' is not complete, it will not be written into the file", entry) + entry = self._table.next_fs() - if spec is not None and file is not None: + if os.path.exists(dest_file): + clean_table.replace_file(dest_file) + else: try: - entry = self._table.find_pair(spec, file, MNT_ITER_FORWARD) + clean_table.write_file(dest_file) + except Exception as e: # pylint: disable=broad-except + # libmount throws general Exception if underlying directories do not exist. Okay... + if str(e) == "No such file or directory": + log.info("Underlying directory of fstab '%s' does not exist. creating...", dest_file) + os.makedirs(os.path.split(dest_file)[0]) + else: + raise + + def find_entry(self, spec=None, file=None, *, entry=None): + """ Return the line of loaded fstab with given spec and/or file. + If both entry and spec/file are given, spec/file are prioritized over entry values. + + :keyword spec: searched device specs (see man fstab(5) fs_spec) + :type spec: str + :keyword file: device mount path (see man fstab(5) fs_file) + :type file: str + :keyword entry: fstab entry as an alternative input type + :type: :class: `FSTabEntry` + :returns: found fstab entry object + :rtype: :class: `FSTabEntry` or None + """ + + _spec = spec or (entry.spec if entry is not None else None) + _file = file or (entry.file if entry is not None else None) + + found_entry = None + + if _spec is not None and _file is not None: + try: + found_entry = self._table.find_pair(_spec, _file, MNT_ITER_FORWARD) except LibmountException: - entry = None + return None + return FSTabEntry(entry=found_entry) - if spec is not None: + if _spec is not None: try: - entry = self._table.find_source(spec, MNT_ITER_FORWARD) + found_entry = self._table.find_source(_spec, MNT_ITER_FORWARD) except LibmountException: - entry = None + return None + return FSTabEntry(entry=found_entry) if file is not None: try: - entry = self._table.find_target(file, MNT_ITER_FORWARD) + found_entry = self._table.find_target(file, MNT_ITER_FORWARD) except LibmountException: - entry = None - return entry + return None + return FSTabEntry(entry=found_entry) - def find_entry_by_device(self, device): - args = self._from_device(device) - return self.find_entry_by_specs(*args) + return None def update(self, devices): - # Update self._table entries based on 'devices' list - # - keep unaffected devices entries unchanged - # - remove entries no longer tied to any device - # - add new entries for devices not present in self._table + """ Update loaded fstab table entries based on 'devices' list + Keep unaffected devices entries unchanged. + Remove entries not tied to any device in the list. + Add new entries for the devices not present in loaded fstab table. + + :param devices: list of StorageDevice instances + :type devices: list + """ # remove entries not tied to any device fstab_entries = [] - entry = self._table.next_fs() - while entry: - fstab_entries.append(entry) - entry = self._table.next_fs() + lm_entry = self._table.next_fs() + while lm_entry: + fstab_entries.append(FSTabEntry(entry=lm_entry)) + lm_entry = self._table.next_fs() for device in devices: - args = self._from_device(device) - entry = self.find_entry_by_specs(*args) + partial_entry = self.entry_from_device(device) + entry = self.find_entry(entry=partial_entry) # remove from the list if found try: @@ -348,17 +639,17 @@ def update(self, devices): pass # All entries left in the fstab_entries do not have their counterpart - # in devices and are to be removed from self._table + # in devices and are to be removed from the final table for entry in fstab_entries: - self._table.remove_fs(entry) + self.remove_entry(entry=entry) # add new entries based on devices not in the table for device in devices: # if mountable the device should be in the fstab if device.format.mountable: - args = self._from_device(device) - found = self.find_entry_by_specs(*args) + entry = self.entry_from_device(device) + found = self.find_entry(entry=entry) if found is None: - self.add_entry_by_specs(*args) - elif found.target != device.format.mountpoint: - self.add_entry_by_specs(args[0], device.format.mountpoint, args[2], args[3]) + self.add_entry(entry=entry) + elif found.file != device.format.mountpoint: + self.add_entry(file=device.format.mountpoint, entry=entry) diff --git a/tests/pylint/runpylint.py b/tests/pylint/runpylint.py index 77f14a02c..930388fe4 100755 --- a/tests/pylint/runpylint.py +++ b/tests/pylint/runpylint.py @@ -29,6 +29,7 @@ def __init__(self): FalsePositive(r"Instance of '(Action.*Device|Action.*Format|Action.*Member|Device|DeviceAction|DeviceFormat|Event|ObjectID|PartitionDevice|StorageDevice|BTRFS.*Device|LoopDevice)' has no 'id' member$"), FalsePositive(r"Instance of 'GError' has no 'message' member"), # overriding currently broken local pylint disable FalsePositive(r"Module '(gi.repository.Gio|gi.repository.GLib)' has no .* member") # pylint/astroid has issues with GI modules https://github.com/PyCQA/pylint/issues/6352 + FalsePositive(r"No name '.*' in module 'libmount'") ] def _files(self): diff --git a/tests/storage_tests/fstab_test.py b/tests/storage_tests/fstab_test.py index 079a6ad00..05fb4e520 100644 --- a/tests/storage_tests/fstab_test.py +++ b/tests/storage_tests/fstab_test.py @@ -3,8 +3,7 @@ from .storagetestcase import StorageTestCase import blivet - -FSTAB_WRITE_FILE = "/tmp/test-blivet-fstab" +import tempfile class FstabTestCase(StorageTestCase): @@ -25,6 +24,8 @@ def setUp(self): def _clean_up(self): + self.storage.fstab.dest_file = None + self.storage.reset() for disk in self.storage.disks: if disk.path not in self.vdevs: @@ -36,40 +37,64 @@ def _clean_up(self): # restore original fstab target file self.storage.fstab.dest_file = "/etc/fstab" - os.remove(FSTAB_WRITE_FILE) - return super()._clean_up() def test_fstab(self): disk = self.storage.devicetree.get_device_by_path(self.vdevs[0]) self.assertIsNotNone(disk) - # change write path of blivet.fstab - self.storage.fstab.dest_file = FSTAB_WRITE_FILE + with tempfile.TemporaryDirectory() as tmpdirname: + fstab_path = os.path.join(tmpdirname, 'fstab') - self.storage.initialize_disk(disk) + # change write path of blivet.fstab + self.storage.fstab.dest_file = fstab_path - pv = self.storage.new_partition(size=blivet.size.Size("100 MiB"), fmt_type="lvmpv", - parents=[disk]) - self.storage.create_device(pv) + self.storage.initialize_disk(disk) - blivet.partitioning.do_partitioning(self.storage) + pv = self.storage.new_partition(size=blivet.size.Size("100 MiB"), fmt_type="lvmpv", + parents=[disk]) + self.storage.create_device(pv) - vg = self.storage.new_vg(name="blivetTestVG", parents=[pv]) - self.storage.create_device(vg) + blivet.partitioning.do_partitioning(self.storage) - lv = self.storage.new_lv(fmt_type="ext4", size=blivet.size.Size("50 MiB"), - parents=[vg], name="blivetTestLVMine") - self.storage.create_device(lv) + vg = self.storage.new_vg(name="blivetTestVG", parents=[pv]) + self.storage.create_device(vg) - # Change the mountpoint, make sure the change will make it into the fstab - ac = blivet.deviceaction.ActionConfigureFormat(device=lv, attr="mountpoint", new_value="/mnt/test2") - self.storage.devicetree.actions.add(ac) + lv = self.storage.new_lv(fmt_type="ext4", size=blivet.size.Size("50 MiB"), + parents=[vg], name="blivetTestLVMine") + self.storage.create_device(lv) - self.storage.do_it() - self.storage.reset() + # Change the mountpoint, make sure the change will make it into the fstab + ac = blivet.deviceaction.ActionConfigureFormat(device=lv, attr="mountpoint", new_value="/mnt/test2") + self.storage.devicetree.actions.add(ac) + + self.storage.do_it() + self.storage.reset() + + # Check fstab contents for added device + with open(fstab_path, "r") as f: + contents = f.read() + self.assertTrue("blivetTestLVMine" in contents) + self.assertTrue("/mnt/test2" in contents) + + dev = self.storage.devicetree.get_device_by_name("blivetTestVG-blivetTestLVMine") + self.storage.recursive_remove(dev) + + self.storage.do_it() + self.storage.reset() + + # Check that previously added device is no longer in fstab + with open(fstab_path, "r") as f: + contents = f.read() + self.assertFalse("blivetTestLVMine" in contents) + self.assertFalse("/mnt/test2" in contents) + + def test_get_device(self): + disk = self.storage.devicetree.get_device_by_path(self.vdevs[0]) + self.assertIsNotNone(disk) + + with tempfile.TemporaryDirectory() as tmpdirname: + fstab_path = os.path.join(tmpdirname, 'fstab') - with open(FSTAB_WRITE_FILE, "r") as f: - contents = f.read() - self.assertTrue("blivetTestLVMine" in contents) - self.assertTrue("/mnt/test2" in contents) + # change write path of blivet.fstab + self.storage.fstab.dest_file = fstab_path diff --git a/tests/unit_tests/fstab_test.py b/tests/unit_tests/fstab_test.py index 965355eb9..a5ed6c877 100644 --- a/tests/unit_tests/fstab_test.py +++ b/tests/unit_tests/fstab_test.py @@ -1,7 +1,8 @@ import os import unittest +import copy -from blivet.fstab import FSTabManager +from blivet.fstab import FSTabManager, FSTabEntry from blivet.devices import DiskDevice from blivet.formats import get_format from blivet import Blivet @@ -26,21 +27,33 @@ def test_fstab(self): self.fstab.src_file = None self.fstab.dest_file = FSTAB_WRITE_FILE + entry = FSTabEntry("/dev/sda_dummy", "/media/wrongpath", "xfs", ["defaults"]) + # create new entries - self.fstab.add_entry_by_specs("/dev/sda_dummy", "/mnt/mountpath", "xfs", "defaults") - self.fstab.add_entry_by_specs("/dev/sdb_dummy", "/media/newpath", "ext4", "defaults") + + # entry.file value should be overridden by add_entry.file parameter + self.fstab.add_entry(file="/mnt/mountpath", entry=entry) + self.fstab.add_entry("/dev/sdb_dummy", "/media/newpath", "ext4", ["defaults"]) + + # test the iterator and check that 2 entries are present + self.assertEqual((sum(1 for _ in self.fstab)), 2, "invalid number of entries in fstab table") # try to find nonexistent entry based on device - entry = self.fstab.find_entry_by_specs("/dev/nonexistent", None, None, None) + entry = self.fstab.find_entry("/dev/nonexistent") self.assertEqual(entry, None) # try to find existing entry based on device - entry = self.fstab.find_entry_by_specs("/dev/sda_dummy", None, None, None) - # check that found item is the correct one - self.assertEqual(entry.target, "/mnt/mountpath") + entry = self.fstab.find_entry("/dev/sda_dummy") + # check that item was found and it is the correct one + self.assertIsNotNone(entry, self.fstab) + self.assertEqual(entry.file, "/mnt/mountpath") - self.fstab.remove_entry_by_specs(None, "/mnt/mountpath", None, None) - entry = self.fstab.find_entry_by_specs(None, "/mnt/mountpath", None, None) + self.fstab.remove_entry(file="/mnt/mountpath") + + # check that number of entries is now 1 + self.assertEqual((sum(1 for _ in self.fstab)), 1, "invalid number of entries in fstab table") + + entry = self.fstab.find_entry(file="/mnt/mountpath") self.assertEqual(entry, None) # write new fstab @@ -51,12 +64,31 @@ def test_fstab(self): contents = f.read() self.assertTrue("/media/newpath" in contents) - def test_from_device(self): + def test_deepcopy(self): + fstab1 = FSTabManager(None, 'dest') + + fstab1.add_entry("/dev/dev1", "path1", "xfs", ["Ph'nglui", "mglw'nafh", "Cthulhu"]) + fstab1.add_entry("/dev/dev2", "path2", "ext4", ["R'lyeh", "wgah'nagl", "fhtagn"]) + + fstab2 = copy.deepcopy(fstab1) + + self.assertEqual(fstab1.src_file, fstab2.src_file) + self.assertEqual(fstab1.dest_file, fstab2.dest_file) + + # Both versions has the same length + self.assertEqual(sum(1 for _ in fstab1), sum(1 for _ in fstab1)) + + # All entries are the same in the same order + for entry1, entry2 in zip(fstab1, fstab2): + self.assertEqual(entry1, entry2) + + def test_entry_from_device(self): device = DiskDevice("test_device", fmt=get_format("ext4", exists=True)) device.format.mountpoint = "/media/fstab_test" - self.assertEqual(self.fstab._from_device(device), ('/dev/test_device', '/media/fstab_test', 'ext4', None)) + _entry = self.fstab.entry_from_device(device) + self.assertEqual(_entry, FSTabEntry('/dev/test_device', '/media/fstab_test', 'ext4', None, 0, 0)) def test_update(self): @@ -78,15 +110,15 @@ def test_update(self): dev4 = DiskDevice("new", fmt=get_format("ext4", exists=True)) dev4.format.mountpoint = "/media/fstab_test3" - self.fstab.add_entry_by_device(dev1) - self.fstab.add_entry_by_device(dev2) + self.fstab.add_entry(entry=self.fstab.entry_from_device(dev1)) + self.fstab.add_entry(entry=self.fstab.entry_from_device(dev2)) self.fstab.update([dev1, dev3, dev4]) # write new fstab - self.fstab.write("/tmp/test_blivet_fstab2") + self.fstab.write(FSTAB_WRITE_FILE) - with open("/tmp/test_blivet_fstab2", "r") as f: + with open(FSTAB_WRITE_FILE, "r") as f: contents = f.read() self.assertTrue("already_present_keep" in contents) @@ -104,7 +136,7 @@ def test_find_device(self): dev1.format.mountpoint = "/mnt/mountpath" b.devicetree._add_device(dev1) - dev2 = self.fstab.find_device_by_specs(b, "/dev/sda_dummy", "/mnt/mountpath", "xfs", "defaults") + dev2 = self.fstab.find_device(b, "/dev/sda_dummy") self.assertEqual(dev1, dev2) @@ -119,6 +151,6 @@ def test_get_device(self): dev1.format.mountpoint = "/mnt/mountpath" b.devicetree._add_device(dev1) - dev2 = self.fstab.get_device_by_specs(b, "/dev/sda_dummy", "/mnt/mountpath", "xfs", "defaults") + dev2 = self.fstab.get_device(b, "/dev/sda_dummy", "/mnt/mountpath", "xfs", ["defaults"]) self.assertEqual(dev1, dev2)