fix: align changes with tests

craft_parts/overlays/chroot.py

@@ -153,63 +153,72 @@ def __init__(
         if fstype is not None:
             self.args.append(f"-t{fstype}")
 
-    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
-        mountpoint_str = str(mountpoint)
-        src_str = str(src)
-
-        # Only mount if mountpoint exists.
-        pid = os.getpid()
-        if mountpoint.exists():
-            logger.debug("[pid=%d] mount %r on chroot", pid, mountpoint_str)
-            os_utils.mount(src_str, mountpoint_str, *args)
-        else:
-            logger.error("[pid=%d] mountpoint %r does not exist", pid, mountpoint_str)
-
     @staticmethod
     def get_abs_path(path: Path, chroot_path: Path):
         return path / str(chroot_path).lstrip("/")
 
-    def mount_to(self, path: Path, *args) -> None:
-        abs_mountpoint = self.get_abs_path(path, self.mountpoint)
+    def mountpoint_exists(self, mountpoint: Path):
+        return mountpoint.exists()
+
+    def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
+        logger.debug("[pid=%d] mount %r on chroot", os.getpid(), src)
+        os_utils.mount(str(src), str(mountpoint), *args)
+
+    def _umount(self, mountpoint: Path, *args: str) -> None:
+        logger.debug("[pid=%d] umount: %r", os.getpid(), mountpoint)
+        os_utils.umount(str(mountpoint), *args)
+
+    def mount_to(self, chroot: Path, *args: str) -> None:
+        abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
+
+        if not self.mountpoint_exists(abs_mountpoint):
+            logger.warning("[pid=%d] mount: %r not found!", os.getpid(), abs_mountpoint)
+            return
 
         self._mount(self.src, abs_mountpoint, *self.args, *args)
 
-    def _umount(self, mountpoint: Path, *args) -> None:
-        mountpoint_str = str(mountpoint)
+    def unmount_from(self, chroot: Path, *args: str) -> None:
+        abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
 
-        pid = os.getpid()
-        if mountpoint.exists():
-            logger.debug("[pid=%d] umount: %r", pid, mountpoint_str)
-            os_utils.umount(mountpoint_str, *args)
-        else:
-            logger.warning("[pid=%d] umount: %r not found!", pid, mountpoint_str)
+        if not self.mountpoint_exists(abs_mountpoint):
+            logger.warning("[pid=%d] umount: %r not found!", os.getpid(), chroot)
+            return
 
-    def unmount_from(self, path: Path, *args) -> None:
-        abs_mountpoint = self.get_abs_path(path, self.mountpoint)
         self._umount(abs_mountpoint, *args)
 
 
 class _BindMount(_Mount):
     bind_type = "bind"
 
-    def __init__(self, src: str | Path, mountpoint: str | Path, *args) -> None:
+    def __init__(
+        self,
+        src: str | Path,
+        mountpoint: str | Path,
+        *args: str,
+    ) -> None:
         super().__init__(src, mountpoint, f"--{self.bind_type}", *args)
 
-    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
+    def mountpoint_exists(self, mountpoint: Path):
+        if self.src.exists() and self.src.is_file():
+            return mountpoint.parent.exists()
+
+        return mountpoint.exists()
+
+    def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
         if src.is_dir():
             # remove existing content of dir
             if mountpoint.exists():
                 rmtree(mountpoint)
 
             # prep mount point
-            mountpoint.mkdir(parents=True, exist_ok=True)
+            mountpoint.mkdir(exist_ok=True)
 
         elif src.is_file():
             # remove existing file
             if mountpoint.exists():
                 mountpoint.unlink()
             else:
-                mountpoint.parent.mkdir(parents=True, exist_ok=True)
+                mountpoint.parent.mkdir(exist_ok=True)
 
             # prep mount point
             mountpoint.touch()

tests/unit/overlays/test_chroot.py

@@ -62,7 +62,7 @@ def test_chroot(self, mocker, new_dir, mock_chroot):
         for subdir in ["etc", "proc", "sys", "dev", "dev/shm"]:
             Path(new_root, subdir).mkdir()
 
-        chroot.chroot(new_root, target_func, "content")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
@@ -95,7 +95,7 @@ def test_chroot_no_mountpoints(self, mocker, new_dir):
         mocker.patch("os.chroot")
 
         Path("dir1").mkdir()
-        chroot.chroot(new_root, target_func, "content")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
@@ -119,8 +119,8 @@ def test_chroot_symlinked_resolv_conf(self, mocker, new_dir):
 
         Path("dir1").mkdir()
         Path("dir1/etc").mkdir()
-        Path("dir1/etc/resolv.con").symlink_to("whatever")
-        chroot.chroot(new_root, target_func, "content")
+        Path("dir1/etc/resolv.conf").symlink_to("whatever")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
@@ -148,7 +148,7 @@ def test_chroot_no_resolv_conf(self, mocker, new_dir):
 
         Path("dir1").mkdir()
         Path("dir1/etc").mkdir()
-        chroot.chroot(new_root, target_func, "content")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [

tests/unit/overlays/test_overlay_manager.py

@@ -177,9 +177,13 @@ def test_refresh_packages_list(self, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", self.mock_refresh_packages_list
+            new_dir / "overlay/overlay",
+            self.mock_refresh_packages_list,
+            use_host_sources=False,
+        )
+        self.mock_refresh_packages_list.assert_called_once_with(
+            use_host_sources=False,
         )
-        self.mock_refresh_packages_list.assert_called_once_with()
 
     def test_download_packages(self, mocker, new_dir):
         mock_download_packages = mocker.patch(
@@ -196,9 +200,15 @@ def test_download_packages(self, mocker, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", mock_download_packages, ["pkg1", "pkg2"]
+            new_dir / "overlay/overlay",
+            mock_download_packages,
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
+        )
+        mock_download_packages.assert_called_once_with(
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
         )
-        mock_download_packages.assert_called_once_with(["pkg1", "pkg2"])
 
     def test_install_packages(self, mocker, new_dir):
         mock_install_packages = mocker.patch(
@@ -218,11 +228,14 @@ def test_install_packages(self, mocker, new_dir):
         self.mock_chroot.assert_called_once_with(
             new_dir / "overlay/overlay",
             mock_install_packages,
-            ["pkg1", "pkg2"],
-            refresh_package_cache=False,
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
         mock_install_packages.assert_called_once_with(
-            ["pkg1", "pkg2"], refresh_package_cache=False
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
 
     def test_package_cache_mount_refresh(self, new_dir):
@@ -242,9 +255,13 @@ def test_package_cache_mount_refresh(self, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", self.mock_refresh_packages_list
+            new_dir / "overlay/overlay",
+            self.mock_refresh_packages_list,
+            use_host_sources=False,
+        )
+        self.mock_refresh_packages_list.assert_called_once_with(
+            use_host_sources=False,
         )
-        self.mock_refresh_packages_list.assert_called_once_with()
         self.mock_umount.assert_called_once_with(new_dir / "overlay/overlay")
 
     def test_package_cache_mount_download(self, mocker, new_dir):
@@ -267,10 +284,18 @@ def test_package_cache_mount_download(self, mocker, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", mock_download_packages, ["pkg1", "pkg2"]
+            new_dir / "overlay/overlay",
+            mock_download_packages,
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
+        )
+        mock_download_packages.assert_called_once_with(
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
+        )
+        self.mock_umount.assert_called_once_with(
+            new_dir / "overlay/overlay",
         )
-        mock_download_packages.assert_called_once_with(["pkg1", "pkg2"])
-        self.mock_umount.assert_called_once_with(new_dir / "overlay/overlay")
 
     def test_layer_mount_install(self, mocker, new_dir):
         mocker.patch("craft_parts.packages.Repository.download_packages")
@@ -296,10 +321,13 @@ def test_layer_mount_install(self, mocker, new_dir):
         self.mock_chroot.assert_called_once_with(
             new_dir / "overlay/overlay",
             mock_install_packages,
-            ["pkg1", "pkg2"],
-            refresh_package_cache=False,
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
         mock_install_packages.assert_called_once_with(
-            ["pkg1", "pkg2"], refresh_package_cache=False
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
         self.mock_umount.assert_called_once_with(new_dir / "overlay/overlay")

tests/unit/test_lifecycle_manager.py

@@ -213,6 +213,7 @@ def test_executor_creation(self, new_dir, mocker):
                 track_stage_packages=False,
                 base_layer_dir=None,
                 base_layer_hash=None,
+                use_host_sources=False,
             )
         ]