From 0514047feed649aa049c04388f29496f94860709 Mon Sep 17 00:00:00 2001 From: Jan Pokorny Date: Wed, 22 Mar 2023 12:09:46 +0100 Subject: [PATCH] Incorporated review comments - changed API - other fixes --- .github/workflows/anaconda_tests.yml | 2 +- blivet/actionlist.py | 14 +- blivet/blivet.py | 5 +- blivet/fstab.py | 854 +++++++++++++++++++-------- misc/files/libmount_deb.py | 123 ++++ misc/install-test-dependencies.yml | 16 + tests/pylint/runpylint.py | 3 +- tests/storage_tests/fstab_test.py | 75 ++- tests/unit_tests/fstab_test.py | 129 ++-- 9 files changed, 900 insertions(+), 321 deletions(-) create mode 100644 misc/files/libmount_deb.py diff --git a/.github/workflows/anaconda_tests.yml b/.github/workflows/anaconda_tests.yml index b3e3188ec..2a94c3c60 100644 --- a/.github/workflows/anaconda_tests.yml +++ b/.github/workflows/anaconda_tests.yml @@ -39,7 +39,7 @@ jobs: podman run -i --rm -v ./blivet:/blivet:z -v ./anaconda:/anaconda:z quay.io/rhinstaller/anaconda-ci:$TARGET_BRANCH sh -c " \ set -xe; \ dnf remove -y python3-blivet; \ - dnf install -y python3-blockdev libblockdev-plugins-all python3-bytesize libbytesize python3-pyparted parted libselinux-python3; \ + dnf install -y python3-blockdev libblockdev-plugins-all python3-bytesize libbytesize python3-pyparted python3-libmount parted libselinux-python3; \ cd /blivet; \ python3 ./setup.py install; \ cd /anaconda; \ diff --git a/blivet/actionlist.py b/blivet/actionlist.py index d60fdb8ca..0e7b22d75 100644 --- a/blivet/actionlist.py +++ b/blivet/actionlist.py @@ -286,6 +286,12 @@ def process(self, callbacks=None, devices=None, fstab=None, dry_run=None): if dry_run: continue + # get (b)efore (a)ction.(e)xecute fstab entry + # (device may not exist afterwards) + if fstab is not None: + entry = fstab.entry_from_device(action.device) + bae_entry = fstab.find_entry(entry=entry) + with blivet_lock: try: @@ -314,11 +320,11 @@ def process(self, callbacks=None, devices=None, fstab=None, dry_run=None): device.update_name() device.format.device = device.path - if fstab is not None: - fstab.update(devices) - fstab.write() - self._completed_actions.append(self._actions.pop(0)) _callbacks.action_executed(action=action) + if fstab is not None: + fstab.update(action, bae_entry) + fstab.write() + self._post_process(devices=devices) 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..59832cf70 100644 --- a/blivet/fstab.py +++ b/blivet/fstab.py @@ -1,9 +1,26 @@ -import os - -import gi +# fstab.py +# Fstab management. +# +# Copyright (C) 2023 Red Hat, Inc. +# +# This copyrighted material is made available to anyone wishing to use, +# modify, copy, or redistribute it subject to the terms and conditions of +# the GNU Lesser General Public License v.2, or (at your option) any later +# version. This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY expressed or implied, including the implied +# warranties of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See +# the GNU Lesser General Public License for more details. You should have +# received a copy of the GNU Lesser General Public License along with this +# program; if not, write to the Free Software Foundation, Inc., 51 Franklin +# Street, Fifth Floor, Boston, MA 02110-1301, USA. Any Red Hat trademarks +# that are incorporated in the source code or documentation are not subject +# to the GNU Lesser General Public License and may only be used or +# replicated with the express permission of Red Hat, Inc. +# +# Red Hat Author(s): Jan Pokorny +# -gi.require_version("BlockDev", "2.0") -from gi.repository import BlockDev as blockdev +import os from libmount import Table, Fs, MNT_ITER_FORWARD from libmount import Error as LibmountException @@ -12,19 +29,250 @@ 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): + + # Note: "*" in arguments means that every following parameter can be used + # only with its key (e.g. "FSTabEntry(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 "" + + @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 -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. + :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 which is singleton, + # 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._table.enable_comments(True) self.src_file = src_file self.dest_file = dest_file @@ -37,81 +285,97 @@ def __deepcopy__(self, memo): clone._table = Table() clone._table.enable_comments(True) + # Two special variables for the first and last comment. When assigning None to them, libmount fails. + # Notice that we are copying the value from the instance of the same type. + if self._table.intro_comment is not None: + clone._table.intro_comment = self._table.intro_comment + if self._table.trailing_comment is not None: + clone._table.trailing_comment = self._table.trailing_comment + 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) + 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() + """ + # Nope, it has to be done like this. 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() + + if device.format.uuid: + entry.spec = "UUID=%s" % device.format.uuid + else: + entry.spec = getattr(device, "fstab_spec", None) - file = 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,245 +384,327 @@ def read(self, src_file=''): else: self.src_file = src_file - self._table.errcb = 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 + self._table.errcb = self._parser_errcb + try: + self._table.parse_fstab(self.src_file) + except Exception as e: # pylint: disable=broad-except + # libmount throws general Exception. Okay... + if str(e) == "No such file or directory": + log.warning("Fstab file '%s' does not exist", self.src_file) + else: + raise + + 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 + # Add '# ' at the start of any comment line and newline to the end of comment. + # Has to be done here since libmount won't do it. 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 + # since libmount happily inserts incomplete lines into the fstab. + # Invalid lines should be skipped, but the whole table is written at once. + # Also the incomplete lines need to be preserved in the object. + # Conclusion: Create the second table, prune invalid/incomplete lines 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 + # Two special variables for the first and last comment. When assigning None libmount fails. + # Notice that we are copying the value from the same type instance. + if self._table.intro_comment is not None: + clean_table.intro_comment = self._table.intro_comment + if self._table.trailing_comment is not None: + clean_table.trailing_comment = self._table.trailing_comment - if spec is not None and file is not 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 os.path.exists(dest_file): + clean_table.replace_file(dest_file) + else: + try: + 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: - entry = self._table.find_pair(spec, file, MNT_ITER_FORWARD) + 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 + def update(self, action, bae_entry): + """ Update fstab based on action type and device. Does not commit changes to a file. - # 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() + :param action: just executed blivet action + :type action: :class: `~.deviceaction.DeviceAction` + :param bae_entry: fstab entry based on action.device before action.execute was called + :type bae_entry: :class: `FSTabEntry` or None + """ - for device in devices: - args = self._from_device(device) - entry = self.find_entry_by_specs(*args) + if not action._applied: + return - # remove from the list if found - try: - fstab_entries.remove(entry) - except ValueError: - pass - - # All entries left in the fstab_entries do not have their counterpart - # in devices and are to be removed from self._table - for entry in fstab_entries: - self._table.remove_fs(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) - 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]) + if action.is_destroy and bae_entry is not None: + # remove destroyed device from the fstab + self.remove_entry(entry=bae_entry) + return + + if action.is_create and action.is_device and action.device.type == "luks/dm-crypt": + # when creating luks format, two actions are made. Device creation + # does not have UUID assigned yet, so we skip that one + return + + if action.is_create and action.device.format.mountable: + # add the device to the fstab + # make sure it is not already present there + entry = self.entry_from_device(action.device) + found = self.find_entry(entry=entry) + if found is None and action.device.format.mountpoint is not None: + # device is not present in fstab and has a defined mountpoint => add it + self.add_entry(entry=entry) + elif found and found.file != action.device.format.mountpoint and action.device.format.mountpoint is not None: + # device already exists in fstab but with a different mountpoint => add it + self.add_entry(file=action.device.format.mountpoint, entry=found) + return + + if action.is_configure and action.is_format and bae_entry is not None: + # Handle change of the mountpoint: + # Change its value if it is defined, remove the fstab entry if it is None + entry = self.entry_from_device(action.device) + if action.device.format.mountpoint is not None and bae_entry.file != action.device.format.mountpoint: + self.remove_entry(entry=bae_entry) + self.add_entry(file=action.device.format.mountpoint, entry=bae_entry) + elif action.device.format.mountpoint is None: + self.remove_entry(entry=bae_entry) diff --git a/misc/files/libmount_deb.py b/misc/files/libmount_deb.py new file mode 100644 index 000000000..43812f5db --- /dev/null +++ b/misc/files/libmount_deb.py @@ -0,0 +1,123 @@ +#!/usr/bin/python + +# Debian based distributions do not have python3-libmount (and never will). This does manual +# installation of it. +# Requires autopoint and bison packages to work. + +from __future__ import print_function + +import os +import re +import shutil +import subprocess +import sys +import tempfile + +LM_GIT = 'https://github.com/util-linux/util-linux' +LM_CONFIG_CMD = './autogen.sh && ./configure --prefix=/usr ' \ + '--with-python=3 ' \ + '--disable-all-programs --enable-pylibmount ' \ + '--enable-libmount --enable-libblkid' +LM_BUILD_PATH = 'util-linux' + + +def run_command(command, cwd=None): + res = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, cwd=cwd) + + out, err = res.communicate() + if res.returncode != 0: + output = out.decode().strip() + '\n' + err.decode().strip() + else: + output = out.decode().strip() + return (res.returncode, output) + +def get_arch(): + _ret, arch = run_command('uname -p') + return arch + +def get_distro(): + _ret, distro = run_command('cat /etc/os-release | grep ^ID= | cut -d= -f2 | tr -d \"') + return distro + +def main(): + + tempname = tempfile.mkdtemp() + + # clone the repo + print("Cloning '%s' repository into '%s'... " % (LM_GIT, tempname), end='', flush=True) + ret, out = run_command('git clone --depth 1 %s' % LM_GIT, tempname) + if ret != 0: + raise RuntimeError('Cloning libmount failed:\n%s' % out) + print('Done') + + print("Getting libmount version... ", end='', flush=True) + ret, out = run_command('mount --version') + if ret != 0: + raise RuntimeError('Getting mount version failed:\n%s' % out) + + match = re.match("^.*libmount (.*):.*$", out) + if not match: + raise RuntimeError('Getting mount version failed:\n%s' % out) + + libmount_version = 'v'+match.group(1) + + print('Done. (libmount version: %s)' % libmount_version) + + # Python-libmount wrapper is a part of util-linux repo, paths have to be set accordingly. + # Correct version of the repo has to be checked out as well. + + workpath = os.path.join(tempname, LM_BUILD_PATH, 'libmount') + + print("Fetching tags (takes a minute)... ", end='', flush=True) + ret, out = run_command('git fetch origin +refs/tags/%s:refs/tags/%s' % + (libmount_version, libmount_version), cwd=workpath) + if ret != 0: + raise RuntimeError('Fetching tags failed:\n%s' % out) + print('Done') + + print("Checking out '%s'... " % libmount_version, end='', flush=True) + + ret, out = run_command('git checkout tags/%s -b tag_temp' % libmount_version, cwd=workpath) + if ret != 0: + raise RuntimeError("Checking out tag '%s' failed:\n%s" % (libmount_version, out)) + print('Done') + + print("Running configure... ", end='', flush=True) + ret, out = run_command(LM_CONFIG_CMD, os.path.join(tempname, LM_BUILD_PATH)) + if ret != 0: + raise RuntimeError('Configure of libmount failed:\n%s' % out) + print('Done') + + print("Running make & make install... ", end='', flush=True) + ret, out = run_command('make -j6 && sudo make install', os.path.join(tempname, LM_BUILD_PATH)) + if ret != 0: + raise RuntimeError('Installing of libmount failed:\n%s' % out) + print('Done') + + print("Creating symlinks 'site-packages -> dist-packages'... ", end='', flush=True) + python_ver = '.'.join(sys.version.split('.')[0:2]) + install_dir = '/usr/lib/python%s/site-packages/libmount' % python_ver + target_dir = '/usr/lib/python%s/dist-packages/libmount' % python_ver + target_dir_local = '/usr/local/lib/python%s/dist-packages/libmount' % python_ver + + ret1, out1 = run_command('ln -fs %s %s' % (install_dir, target_dir)) + ret2, out2 = run_command('ln -fs %s %s' % (install_dir, target_dir_local)) + + # One can/will fail but not both + if ret1 + ret2 > 1: + raise RuntimeError('Symlink creation for libmount failed:\n%s\n%s' % (out1, out2)) + print('Done') + + shutil.rmtree(tempname) + + +if __name__ == '__main__': + + try: + main() + except RuntimeError as e: + print(e) + sys.exit(1) + + sys.exit(0) diff --git a/misc/install-test-dependencies.yml b/misc/install-test-dependencies.yml index 0902ad90d..3ba098d71 100644 --- a/misc/install-test-dependencies.yml +++ b/misc/install-test-dependencies.yml @@ -50,6 +50,7 @@ - libselinux-python3 - python3-blockdev - python3-bytesize + - python3-libmount - python3-libvirt - python3-paramiko - targetcli @@ -102,6 +103,7 @@ - libselinux-python3 - python3-blockdev - python3-bytesize + - python3-libmount - python3-libvirt - python3-pip - targetcli @@ -146,6 +148,8 @@ package: state: present name: + - autopoint + - bison - dosfstools - e2fsprogs - xfsprogs @@ -163,6 +167,18 @@ - open-iscsi when: ansible_distribution == 'Debian' or ansible_distribution == 'Ubuntu' + - name: Install libmount (Debian/Ubuntu) + block: + - name: Copy the libmount_deb script (Debian and CentOS non-x86_64) + copy: + src: files/libmount_deb.py + dest: "/tmp/libmount_deb.py" + mode: 0755 + + - name: Make and install libmount + shell: "python3 /tmp/libmount_deb.py" + when: ansible_distribution == 'Debian' or ansible_distribution == 'Ubuntu' + - name: Install pocketlint using pip (Debian/Ubuntu) pip: name=pocketlint executable=pip3 when: ansible_distribution == 'Debian' or ansible_distribution == 'Ubuntu' diff --git a/tests/pylint/runpylint.py b/tests/pylint/runpylint.py index 77f14a02c..196222495 100755 --- a/tests/pylint/runpylint.py +++ b/tests/pylint/runpylint.py @@ -28,7 +28,8 @@ def __init__(self): FalsePositive(r"Bad option value '(subprocess-popen-preexec-fn|try-except-raise|environment-modify|arguments-renamed|redundant-u-string-prefix)'"), 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"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..c8353eb0e 100644 --- a/tests/unit_tests/fstab_test.py +++ b/tests/unit_tests/fstab_test.py @@ -1,7 +1,13 @@ +try: + from unittest.mock import Mock +except ImportError: + from mock import Mock + 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 +32,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(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") - self.fstab.remove_entry_by_specs(None, "/mnt/mountpath", None, None) - entry = self.fstab.find_entry_by_specs(None, "/mnt/mountpath", None, None) + entry = self.fstab.find_entry(file="/mnt/mountpath") self.assertEqual(entry, None) # write new fstab @@ -51,48 +69,89 @@ 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): # Reset table self.fstab.read(None) - # Device already present in fstab._table that should be kept there after fstab.update() - dev1 = DiskDevice("already_present_keep", fmt=get_format("ext4", exists=True)) - dev1.format.mountpoint = "/media/fstab_test1" + # Device is not in the table and should be added + dev1 = DiskDevice("device1", fmt=get_format("ext4", exists=True)) + dev1.format.mountpoint = "/media/fstab_original_mountpoint" - # Device already present in fstab._table which should be removed by fstab.update() - dev2 = DiskDevice("already_present_remove", fmt=get_format("ext4", exists=True)) - dev2.format.mountpoint = "/media/fstab_test2" + action = Mock() + action.device = dev1 + action.is_create = True - # Device not at fstab._table which should not be added since it is not mountable - dev3 = DiskDevice("unmountable_skip") + self.fstab.update(action, None) - # Device not at fstab._table that should be added there after fstab.update() - dev4 = DiskDevice("new", fmt=get_format("ext4", exists=True)) - dev4.format.mountpoint = "/media/fstab_test3" + entry = self.fstab.entry_from_device(dev1) + self.assertIsNotNone(self.fstab.find_entry(entry=entry)) - self.fstab.add_entry_by_device(dev1) - self.fstab.add_entry_by_device(dev2) + # Device is in the table and its mountpoint has changed + action = Mock() + dev1.format.mountpoint = "/media/fstab_changed_mountpoint" + action.device = dev1 + action.is_destroy = False + action.is_create = False + action.is_configure = True + action.is_format = True + bae_entry = self.fstab.find_entry(entry=entry) - self.fstab.update([dev1, dev3, dev4]) + self.fstab.update(action, bae_entry) + self.assertIsNotNone(self.fstab.find_entry(spec="/dev/device1", file="/media/fstab_changed_mountpoint")) - # write new fstab - self.fstab.write("/tmp/test_blivet_fstab2") + # Device is already present in the table and should be removed + action = Mock() + action.device = dev1 + action.is_destroy = True + bae_entry = self.fstab.find_entry(entry=entry) - with open("/tmp/test_blivet_fstab2", "r") as f: - contents = f.read() + self.fstab.update(action, bae_entry) + self.assertIsNone(self.fstab.find_entry(entry=entry)) + + # Device not at fstab._table which should not be added since it is not mountable + dev2 = DiskDevice("unmountable_skip") + + # Add mountpoint just to mess with fstab manager + dev2.format.mountpoint = "/media/absent_in_fstab" + + # Make sure that the device is not mountable + self.assertFalse(dev2.format.mountable) + + action = Mock() + action.device = dev2 + action.is_create = True + + entry = self.fstab.entry_from_device(dev2) - self.assertTrue("already_present_keep" in contents) - self.assertFalse("already_present_remove" in contents) - self.assertFalse("unmountable_skip" in contents) - self.assertTrue("new" in contents) + self.fstab.update(action, None) + self.assertIsNone(self.fstab.find_entry(entry=entry)) def test_find_device(self): # Reset table @@ -104,7 +163,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 +178,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)