Merge branch '3.10-devel'

.packit.yaml

@@ -4,7 +4,7 @@ actions:
     - 'git config user.email "blivet-ci@example.com"'
     - 'git config user.name "Blivet CI"'
     # merge the release branch to get correct version in spec
-    - 'git merge --ff origin/3.9-release'
+    - 'git merge --ff origin/3.10-release'
     # bump release to 99 to always be ahead of Fedora builds
     - 'bash -c "sed -i -r \"s/Release:(\s*)\S+/Release: 99%{?dist}/\" python-blivet.spec"'
   get-current-version:
@@ -30,7 +30,7 @@ jobs:
   trigger: commit
   owner: "@storage"
   project: blivet-daily
-  branch: 3.9-devel
+  branch: 3.10-devel
   preserve_project: true
 
 srpm_build_deps:

blivet/actionlist.py

@@ -290,7 +290,7 @@ def process(self, callbacks=None, devices=None, fstab=None, dry_run=None):
             # (device may not exist afterwards)
             if not skip_fstab:
                 try:
-                    entry = fstab.entry_from_device(action.device)
+                    entry = fstab.entry_from_action(action)
                 except ValueError:
                     # this device should not be in fstab
                     bae_entry = None

blivet/blivet.py

@@ -55,7 +55,10 @@
 log = logging.getLogger("blivet")
 
 
-FSTAB_PATH = "/etc/fstab"
+# Default path to fstab file. Left empty to prevent blivet from using
+# fstab functionality by default.
+# TODO Change to "/etc/fstab" at next major version
+FSTAB_PATH = ""
 
 
 class Blivet(object, metaclass=SynchronizedMeta):

blivet/devices/btrfs.py

@@ -357,7 +357,7 @@ def _get_any_btrfs_mountpoint(self):
 
         # now try every possible mountpoint with any subvolspec in our cache
         parents = [p.name for p in self.parents]
-        mount_spec = next(((dev, subvol) for dev, subvol in mounts_cache.mountpoints if dev in parents), None)
+        mount_spec = next(((dev, subvol) for dev, subvol in mounts_cache.mountpoints if udev.resolve_devspec(dev) in parents), None)
         if mount_spec:
             try:
                 return mounts_cache.get_mountpoints(devspec=mount_spec[0],

blivet/fstab.py

@@ -391,11 +391,48 @@ def entry_from_device(self, device):
         entry.spec = self._get_spec(device)
         if entry.spec is None:
             entry.spec = getattr(device, "fstab_spec", None)
-
         entry.vfstype = device.format.type
 
         return entry
 
+    def entry_from_action(self, action):
+        """ Generate FSTabEntry object based on given blivet Action
+
+            :keyword action: action to process
+            :type action: :class: `blivet.deviceaction.DeviceAction`
+            :returns: fstab entry object based on device processed by action
+            :rtype: :class: `FSTabEntry`
+        """
+
+        if not action.is_format:
+            raise ValueError("""cannot generate fstab entry from action '%s' because
+                                its type is not 'format'""" % action)
+
+        fmt = action.format
+        if action.is_destroy:
+            fmt = action.orig_format
+        if action.is_create:
+            fmt = action.device.format
+
+        entry = FSTabEntry()
+
+        entry.file = None
+        if fmt.mountable:
+            entry.file = fmt.mountpoint
+        elif fmt.type == "swap":
+            entry.file = "swap"
+        else:
+            raise ValueError("""cannot generate fstab entry from action '%s' because
+                                it is neither mountable nor swap type""" % action)
+
+        entry.spec = self._get_spec(action.device)
+        if entry.spec is None:
+            entry.spec = getattr(action.device, "fstab_spec", None)
+
+        entry.vfstype = action.device.format.type
+
+        return entry
+
     def read(self):
         """ Read the fstab file from path stored in self.src_file. Resets currently loaded table contents.
         """
@@ -546,7 +583,7 @@ def add_entry(self, spec=None, file=None, vfstype=None, mntops=None,
 
         if mntops is not None:
             _entry.mntops = mntops
-        elif _entry.mntops is None:
+        if _entry.mntops is None:
             _entry.mntops = ['defaults']
 
         if freq is not None:
@@ -696,7 +733,11 @@ def _get_spec(self, device):
         # Use "globally" set (on FSTabManager level) spec type otherwise
 
         spec = None
-        spec_type = device.format.fstab.spec_type or self.spec_type
+
+        if hasattr(device.format, 'fstab') and device.format.fstab.spec_type:
+            spec_type = device.format.fstab.spec_type
+        else:
+            spec_type = self.spec_type
 
         if spec_type == "LABEL" and device.format.label:
             spec = "LABEL=%s" % device.format.label
@@ -737,14 +778,15 @@ def update(self, action, bae_entry):
             # does not have UUID assigned yet, so we skip that one
             return
 
-        if action.is_create and action.device.format.mountable:
+        if action.is_create and action.is_format and action.device.format.mountable:
             # add the device to the fstab
             # make sure it is not already present there
             try:
                 entry = self.entry_from_device(action.device)
             except ValueError:
                 # this device should not be at fstab
                 found = None
+                entry = None
             else:
                 found = self.find_entry(entry=entry)
 
@@ -754,6 +796,7 @@ def update(self, action, bae_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(spec=spec,
+                               file=action.device.format.mountpoint,
                                mntops=action.device.format.fstab.mntops,
                                freq=action.device.format.fstab.freq,
                                passno=action.device.format.fstab.passno,

blivet/tasks/availability.py

@@ -22,7 +22,6 @@
 import abc
 import shutil
 
-from .. import safe_dbus
 from ..devicelibs.stratis import STRATIS_SERVICE, STRATIS_PATH
 
 import gi
@@ -316,8 +315,7 @@ def availability_errors(self, resource):
         """
         try:
             avail = blockdev.utils.dbus_service_available(None, Gio.BusType.SYSTEM, self.dbus_name, self.dbus_path)
-            avail = safe_dbus.check_object_available(self.dbus_name, self.dbus_path)
-        except safe_dbus.DBusCallError:
+        except blockdev.UtilsError:
             return ["DBus service %s not available" % resource.name]
         else:
             if avail:

tests/run_tests.py

@@ -3,6 +3,7 @@
 from __future__ import print_function
 
 import argparse
+import csv
 import dbus
 import os
 import pdb
@@ -54,26 +55,72 @@ def get_version_from_pretty_name(pretty_name):
     return (distro, version)
 
 
-def get_version():
-    """ Try to get distro and version
+def get_version_from_cpe(cpe):
+    """ Try to get distro and version from 'OperatingSystemCPEName'
+        hostname property.
+
+        It should look like this:
+         - "cpe:/o:fedoraproject:fedora:39"
+         - "cpe:/o:redhat:enterprise_linux:7.3:GA:server
+    """
+    # 2nd to 4th fields from e.g. "cpe:/o:fedoraproject:fedora:25" or "cpe:/o:redhat:enterprise_linux:7.3:GA:server"
+    _project, distro, version = tuple(cpe.split(":")[2:5])
+    version = str(int(float(version)))
+    return (distro, version)
+
+
+def get_version_from_dbus():
+    """ Try to get distro and version from dbus
     """
 
     bus = dbus.SystemBus()
 
     # get information about the distribution from systemd (hostname1)
     sys_info = bus.get_object("org.freedesktop.hostname1", "/org/freedesktop/hostname1")
-    cpe = str(sys_info.Get("org.freedesktop.hostname1", "OperatingSystemCPEName", dbus_interface=dbus.PROPERTIES_IFACE))
 
+    cpe = str(sys_info.Get("org.freedesktop.hostname1", "OperatingSystemCPEName", dbus_interface=dbus.PROPERTIES_IFACE))
     if cpe:
-        # 2nd to 4th fields from e.g. "cpe:/o:fedoraproject:fedora:25" or "cpe:/o:redhat:enterprise_linux:7.3:GA:server"
-        _project, distro, version = tuple(cpe.split(":")[2:5])
-        # we want just the major version, so remove all decimal places (if any)
-        version = str(int(float(version)))
-    else:
-        pretty_name = str(sys_info.Get("org.freedesktop.hostname1", "OperatingSystemPrettyName", dbus_interface=dbus.PROPERTIES_IFACE))
-        distro, version = get_version_from_pretty_name(pretty_name)
+        return get_version_from_cpe(cpe)
 
-    return (distro, version)
+    pretty_name = str(sys_info.Get("org.freedesktop.hostname1", "OperatingSystemPrettyName", dbus_interface=dbus.PROPERTIES_IFACE))
+    if pretty_name:
+        return get_version_from_pretty_name(pretty_name)
+
+    raise RuntimeError("Failed to get distro and version from DBus")
+
+
+def get_version_from_os():
+    """ Try to get distro and version from /etc/os-release
+    """
+    if not os.path.isfile("/etc/os-release"):
+        raise RuntimeError("/etc/os-release does not exist")
+
+    with open('/etc/os-release') as csvfile:
+        reader = csv.reader(csvfile, delimiter='=')
+        release = dict(reader)
+
+    if 'CPE_NAME' in release.keys():
+        return get_version_from_cpe(release['CPE_NAME'])
+    elif 'PRETTY_NAME' in release.keys():
+        return get_version_from_cpe(release['PRETTY_NAME'])
+    elif 'ID' in release.keys() and 'VERSION_ID' in release.keys():
+        return (release['ID'], release['VERSION_ID'])
+
+    raise RuntimeError("Failed to get distro and version from /etc/os-release")
+
+
+def get_version():
+    try:
+        return get_version_from_dbus()
+    except Exception as err:  # pylint: disable=broad-except
+        print(err)
+
+    try:
+        return get_version_from_os()
+    except Exception as err:  # pylint: disable=broad-except
+        print(err)
+
+    raise RuntimeError("Failed to get distro and version")
 
 
 def _should_skip(distro=None, version=None, arch=None, reason=None):  # pylint: disable=unused-argument

tests/storage_tests/fstab_test.py

@@ -1,10 +1,9 @@
 import os
-
-from .storagetestcase import StorageTestCase
-
 import blivet
 import tempfile
 
+from .storagetestcase import StorageTestCase
+
 
 class FstabTestCase(StorageTestCase):
 
@@ -97,6 +96,50 @@ def test_fstab(self):
                 self.assertFalse("blivetTestLVMine" in contents)
                 self.assertFalse("/mnt/test2" in contents)
 
+    def test_luks_creation(self):
+        # test creation of a multiple layer device
+        disk = self.storage.devicetree.get_device_by_path(self.vdevs[0])
+        self.assertIsNotNone(disk)
+
+        fstab_path = '/tmp/myfstab'
+
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            fstab_path = os.path.join(tmpdirname, 'fstab')
+
+            # change write path of blivet.fstab
+            self.storage.fstab.dest_file = fstab_path
+
+            self.storage.initialize_disk(disk)
+
+            var = self.storage.new_partition(fmt_type="luks",
+                                             fmt_args={"passphrase": "opensaysme"},
+                                             size=blivet.size.Size("200MiB"),
+                                             parents=[disk],
+                                             mountpoint="/mnt/test_fstab_luks_wrong")
+
+            self.storage.create_device(var)
+
+            varenc = blivet.devices.LUKSDevice(name="blivetTest_fstab_luks",
+                                               size=var.size,
+                                               parents=var)
+            self.storage.create_device(varenc)
+
+            varfs = blivet.formats.get_format(fmt_type="ext4",
+                                              device=varenc.path,
+                                              mountpoint="/mnt/test_fstab_luks_correct")
+            self.storage.format_device(varenc, varfs)
+
+            blivet.partitioning.do_partitioning(self.storage)
+
+            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("/mnt/test_fstab_luks_correct" in contents)
+                self.assertFalse("/mnt/test_fstab_luks_wrong" in contents)
+
     def test_swap_creation(self):
         # test swap creation for presence of FSTabOptions object
         disk = self.storage.devicetree.get_device_by_path(self.vdevs[0])