Refactor LiveOS installation to reuse existing mount point

Eliminate duplicate mounting by reusing /run/rootfsbase instead
of creating /run/install/sources/mount-0000-live_os_image.
This simplifies the code and removes redundant filesystem
operations during installation.

The original implementation created a separate mount point to follow the
same pattern as other sources (CDROM, NFS, HDD). However, LiveOS is a
special case where /run/rootfsbase is already mounted by the live system
and contains the same content. Creating a duplicate mount is redundant.

Relevant: rhbz#2413622

Author: KKoukiou

pyanaconda/modules/payloads/source/live_os/initialization.py

@@ -19,20 +19,18 @@
 import stat
 from collections import namedtuple
 
-import blivet.util
 from blivet.size import Size
 
 from pyanaconda.anaconda_loggers import get_module_logger
 from pyanaconda.core.util import execWithCapture
-from pyanaconda.modules.common.constants.objects import DEVICE_TREE
-from pyanaconda.modules.common.constants.services import STORAGE
 from pyanaconda.modules.common.errors.payload import SourceSetupError
-from pyanaconda.modules.common.structures.storage import DeviceData
 from pyanaconda.modules.common.task import Task
-from pyanaconda.modules.payloads.source.mount_tasks import SetUpMountTask
 
 log = get_module_logger(__name__)
 
+# Mount point for the Live OS rootfsbase (mounted by the live system)
+LIVE_OS_ROOTFSBASE_MOUNT = "/run/rootfsbase"
+
 SetupLiveOSResult = namedtuple("SetupLiveOSResult", ["required_space"])
 
 
@@ -53,7 +51,7 @@ def run(self):
         block_device = \
             self._check_block_device("/dev/mapper/live-base") or \
             self._check_block_device("/dev/mapper/live-osimg-min") or \
-            self._check_mount_point("/run/rootfsbase")
+            self._check_mount_point(LIVE_OS_ROOTFSBASE_MOUNT)
 
         if not block_device:
             raise SourceSetupError("No Live OS image found!")
@@ -89,22 +87,19 @@ def _check_mount_point(self, mount_point):
         return None
 
 
-class SetUpLiveOSSourceTask(SetUpMountTask):
+class SetUpLiveOSSourceTask(Task):
     """Task to set up a Live OS image."""
 
-    def __init__(self, image_path, target_mount):
+    def __init__(self, target_mount):
         """Create a new task.
 
-        :param image_path: a path to a Live OS image
         :param target_mount: a path to a mount point
         """
-        super().__init__(target_mount)
-        self._image_path = image_path
+        super().__init__()
+        self._target_mount = target_mount
 
     def run(self):
         """Run the task."""
-        super().run()
-
         required_space = self._calculate_required_space()
         return SetupLiveOSResult(required_space=required_space)
 
@@ -132,7 +127,7 @@ def _calculate_required_space(self):
             du_cmd_args.extend(["--exclude", f"{self._target_mount}{pattern}"])
 
         try:
-            # Execute the `du` command
+            # Execute the `du` command on the existing mount point
             result = execWithCapture("du", du_cmd_args)
             # Parse the output for the total size
             required_space = result.split()[0]  # First column is the total
@@ -144,50 +139,3 @@ def _calculate_required_space(self):
     @property
     def name(self):
         return "Set up a Live OS image"
-
-    def _do_mount(self):
-        """Run live installation source setup.
-
-        Mount the live device and copy from it instead of the overlay at /.
-        """
-        device_path = self._get_device_path()
-        self._mount_device(device_path)
-
-    def _get_device_path(self):
-        """Get a device path of the block device."""
-        log.debug("Resolving %s.", self._image_path)
-        device_tree = STORAGE.get_proxy(DEVICE_TREE)
-
-        # Get the device name.
-        device_id = device_tree.ResolveDevice(self._image_path)
-
-        if not device_id:
-            raise SourceSetupError("Failed to resolve the Live OS image.")
-
-        # Get the device path.
-        device_data = DeviceData.from_structure(
-            device_tree.GetDeviceData(device_id)
-        )
-        device_path = device_data.path
-
-        if not stat.S_ISBLK(os.stat(device_path)[stat.ST_MODE]):
-            raise SourceSetupError("{} is not a valid block device.".format(device_path))
-
-        return device_path
-
-    def _mount_device(self, device_path):
-        """Mount the specified device."""
-        log.debug("Mounting %s at %s.", device_path, self._target_mount)
-
-        try:
-            rc = blivet.util.mount(
-                device_path,
-                self._target_mount,
-                fstype="auto",
-                options="ro"
-            )
-        except OSError as e:
-            raise SourceSetupError(str(e)) from e
-
-        if rc != 0:
-            raise SourceSetupError("Failed to mount the Live OS image.")

pyanaconda/modules/payloads/source/live_os/live_os.py

@@ -23,22 +23,21 @@
 from pyanaconda.modules.payloads.constants import SourceState, SourceType
 from pyanaconda.modules.payloads.source.live_os.initialization import (
     DetectLiveOSImageTask,
+    LIVE_OS_ROOTFSBASE_MOUNT,
     SetupLiveOSResult,
     SetUpLiveOSSourceTask,
 )
 from pyanaconda.modules.payloads.source.live_os.live_os_interface import (
     LiveOSSourceInterface,
 )
-from pyanaconda.modules.payloads.source.mount_tasks import TearDownMountTask
 from pyanaconda.modules.payloads.source.source_base import (
-    MountingSourceMixin,
     PayloadSourceBase,
 )
 
 log = get_module_logger(__name__)
 
 
-class LiveOSSourceModule(PayloadSourceBase, MountingSourceMixin):
+class LiveOSSourceModule(PayloadSourceBase):
     """The Live OS source payload module."""
 
     def __init__(self):
@@ -78,6 +77,18 @@ def required_space(self):
         """
         return self._required_space
 
+    @property
+    def mount_point(self):
+        """Where the Live OS source is mounted.
+
+        The Live OS source reuses the existing /run/rootfsbase mount point
+        that is created by the live system.
+
+        :return: path to the mount point
+        :rtype: str
+        """
+        return LIVE_OS_ROOTFSBASE_MOUNT
+
     @property
     def image_path(self):
         """Path to the Live OS image.
@@ -113,7 +124,9 @@ def detect_image_with_task(self):
 
     def get_state(self):
         """Get state of this source."""
-        return SourceState.from_bool(self.get_mount_state())
+        # Since we're using an existing mount and don't have a traditional mount state,
+        # return NOT_APPLICABLE for consistency with other sources like live_image
+        return SourceState.NOT_APPLICABLE
 
     def set_up_with_tasks(self):
         """Set up the installation source for installation.
@@ -122,7 +135,6 @@ def set_up_with_tasks(self):
         :rtype: [Task]
         """
         task = SetUpLiveOSSourceTask(
-            image_path=self.image_path,
             target_mount=self.mount_point
         )
 
@@ -136,14 +148,13 @@ def _handle_live_os_task_result(self, result: SetupLiveOSResult):
     def tear_down_with_tasks(self):
         """Tear down the installation source.
 
-        :return: list of tasks required for the source clean-up
-        :rtype: [TearDownMountTask]
+        Note: We don't tear down /run/rootfsbase since it's mounted by the live system
+        and unmounting it would crash the system.
+
+        :return: empty list (no tear-down tasks needed)
+        :rtype: []
         """
-        return [
-            TearDownMountTask(
-                target_mount=self.mount_point
-            )
-        ]
+        return []  # No tear-down tasks needed
 
     def __repr__(self):
         """Return a string representation of the source."""

tests/unit_tests/pyanaconda_tests/modules/payloads/payload/test_module_payload_live_os.py

@@ -20,9 +20,10 @@
 import os
 import tempfile
 import unittest
+from unittest.mock import patch
 
 from pyanaconda.core.constants import PAYLOAD_TYPE_LIVE_OS, SOURCE_TYPE_LIVE_OS_IMAGE
-from pyanaconda.core.path import join_paths, make_directories, touch
+from pyanaconda.core.path import join_paths, make_directories
 from pyanaconda.modules.common.errors.payload import IncompatibleSourceError
 from pyanaconda.modules.payloads.constants import SourceState, SourceType
 from pyanaconda.modules.payloads.payload.live_image.installation import (
@@ -91,21 +92,19 @@ def _create_source(self, state=SourceState.READY):
         """Create a new source with a mocked state."""
         return PayloadSharedTest.prepare_source(SourceType.LIVE_OS_IMAGE, state)
 
-    def test_get_kernel_version_list(self):
+    @patch("pyanaconda.modules.payloads.payload.live_os.live_os.get_kernel_version_list")
+    def test_get_kernel_version_list(self, get_kernel_mock):
         """Test the get_kernel_version_list method."""
-        with tempfile.TemporaryDirectory() as tmp:
-            # Create the image source.
-            image_source = self._create_source()
-            image_source._mount_point = tmp
+        get_kernel_mock.return_value = ["1.2-3.x86_64"]
 
-            # Create a fake kernel file.
-            os.makedirs(join_paths(tmp, "boot"))
-            kernel_file = join_paths(tmp, "boot", "vmlinuz-1.2-3.x86_64")
-            touch(kernel_file)
+        # Create the image source.
+        image_source = self._create_source()
 
-            self.module._update_kernel_version_list(image_source)
+        self.module._update_kernel_version_list(image_source)
 
         assert self.module.get_kernel_version_list() == ["1.2-3.x86_64"]
+        # Verify it was called with the mount_point from the source
+        get_kernel_mock.assert_called_once_with(image_source.mount_point)
 
     def test_install_with_task(self):
         """Test the install_with_tasks method."""