Merge pull request #1891 from ogayot/apt-mounter-pathlib

Mounter / APT use pathlib instead of os

subiquity/server/apt.py

@@ -18,7 +18,6 @@ import contextlib
 import enum
 import io
 import logging
-import os
 import pathlib
 import random
 import re
@@ -71,7 +70,7 @@ def get_index_targets() -> List[str]:
 def apt_sourceparts_files(mp: Mountpoint) -> List[str]:
     # Return the relative path of the files from
     # mp("etc/apt/sources.list.d") that apt will read.
-    root = pathlib.Path(mp.p())
+    root = mp.pp()
     sources_list_d = root / "etc/apt/sources.list.d"
     paths = list(sources_list_d.glob("*.list")) + list(sources_list_d.glob("*.sources"))
     return [str(p.relative_to(root)) for p in paths]
@@ -147,11 +146,12 @@ class AptConfigurer:
     async def apply_apt_config(self, context, final: bool):
         self.configured_tree = await self.mounter.setup_overlay([self.source_path])
 
-        config_location = os.path.join(
-            self.app.root, "var/log/installer/curtin-install/subiquity-curtin-apt.conf"
+        config_location = pathlib.Path(self.app.root).joinpath(
+            "var/log/installer/curtin-install/subiquity-curtin-apt.conf"
         )
-        generate_config_yaml(config_location, self.apt_config(final))
-        self.app.note_data_for_apport("CurtinAptConfig", config_location)
+
+        generate_config_yaml(str(config_location), self.apt_config(final))
+        self.app.note_data_for_apport("CurtinAptConfig", str(config_location))
 
         await run_curtin_command(
             self.app,
@@ -159,7 +159,7 @@ class AptConfigurer:
             "apt-config",
             "-t",
             self.configured_tree.p(),
-            config=config_location,
+            config=str(config_location),
             private_mounts=True,
         )
 
@@ -172,7 +172,7 @@ class AptConfigurer:
         with non-zero."""
         assert self.configured_tree is not None
 
-        pfx = pathlib.Path(self.configured_tree.p())
+        pfx = self.configured_tree.pp()
 
         apt_config = apt_pkg.Configuration()
 
@@ -239,21 +239,21 @@ class AptConfigurer:
 
         self.install_tree = await self.mounter.setup_overlay([self.configured_tree])
 
-        os.mkdir(self.install_tree.p("cdrom"))
+        self.install_tree.pp("cdrom").mkdir()
         await self.mounter.mount("/cdrom", self.install_tree.p("cdrom"), options="bind")
 
         if self.app.base_model.network.has_network:
             with contextlib.suppress(FileNotFoundError):
-                os.rename(
-                    self.install_tree.p("etc/apt/sources.list"),
-                    self.install_tree.p("etc/apt/sources.list.d/original.list"),
+                self.install_tree.pp("etc/apt/sources.list").rename(
+                    self.install_tree.pp("etc/apt/sources.list.d/original.list"),
                 )
         else:
-            proxy_path = self.install_tree.p("etc/apt/apt.conf.d/90curtin-aptproxy")
-            if os.path.exists(proxy_path):
-                os.unlink(proxy_path)
+            self.install_tree.pp("etc/apt/sources.list").unlink(missing_ok=True)
+            self.install_tree.pp("etc/apt/apt.conf.d/90curtin-aptproxy").unlink(
+                missing_ok=True
+            )
             for relpath in apt_sourceparts_files(self.configured_tree):
-                os.unlink(self.install_tree.p(relpath))
+                self.install_tree.pp(relpath).unlink()
 
         codename = lsb_release(dry_run=self.app.opts.dry_run)["codename"]
 
@@ -325,12 +325,11 @@ class AptConfigurer:
             shutil.copyfile(self.configured_tree.p(path), target_mnt.p(path))
 
         # The file only exists if we are online
-        with contextlib.suppress(FileNotFoundError):
-            os.unlink(target_mnt.p("etc/apt/sources.list.d/original.list"))
+        target_mnt.pp("etc/apt/sources.list.d/original.list").unlink(missing_ok=True)
         try:
             _restore_file("etc/apt/sources.list")
         except FileNotFoundError:
-            os.unlink(target_mnt.p("etc/apt/sources.list"))
+            target_mnt.pp("etc/apt/sources.list").unlink()
 
         with contextlib.suppress(FileNotFoundError):
             _restore_file("etc/apt/apt.conf.d/90curtin-aptproxy")
@@ -354,8 +353,8 @@ class AptConfigurer:
 
         await self.cleanup()
         try:
-            d = target_mnt.p("cdrom")
-            os.rmdir(d)
+            d = target_mnt.pp("cdrom")
+            d.rmdir()
         except OSError as ose:
             log.warning(f"failed to rmdir {d}: {ose}")
 

subiquity/server/mounter.py

@@ -51,12 +51,20 @@ class OverlayCleanupError(Exception):
     """Exception to raise when an overlay could not be cleaned up."""
 
 
+class AbsolutePathError(Exception):
+    pass
+
+
 class _MountBase:
-    def p(self, *args: str) -> str:
+    def pp(self, *args: Union[str, Path]) -> Path:
+        """Same as p() but returns a pathlib.Path."""
         for a in args:
-            if a.startswith("/"):
-                raise Exception("no absolute paths here please")
-        return os.path.join(self.mountpoint, *args)
+            if Path(a).is_absolute():
+                raise AbsolutePathError("no absolute paths here please")
+        return Path(self.mountpoint).joinpath(*args)
+
+    def p(self, *args: Union[str, Path]) -> str:
+        return str(self.pp(*args))
 
     def write(self, path, content):
         with open(self.p(path), "w") as fp:

subiquity/server/tests/test_mounter.py

@@ -14,19 +14,48 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
-import pathlib
+from pathlib import Path
 from unittest.mock import AsyncMock, Mock, call, patch
 
 from subiquity.server.mounter import (
+    AbsolutePathError,
     Mounter,
     Mountpoint,
     OverlayMountpoint,
+    _MountBase,
     lowerdir_for,
 )
 from subiquitycore.tests import SubiTestCase
 from subiquitycore.tests.mocks import make_app
 
 
+class Test_MountBase(SubiTestCase):
+    def setUp(self):
+        self.mountbase = _MountBase()
+        self.mountbase.mountpoint = "/target"
+
+    def test_pp(self):
+        mnt = self.mountbase
+
+        self.assertEqual(Path("/target/d1"), mnt.pp("d1"))
+        self.assertEqual(Path("/target/d1/d2/d3/d4"), mnt.pp("d1", "d2/d3", "d4"))
+
+        # Mix of strings and paths should produce the same result.
+        self.assertEqual(mnt.pp("d1", "d2/d3"), mnt.pp("d1", Path("d2/d3")))
+        self.assertEqual(mnt.pp("d1", "d2/d3"), mnt.pp(Path("d1"), "d2/d3"))
+
+    def test_pp__absolute(self):
+        with self.assertRaises(AbsolutePathError):
+            self.mountbase.pp("a", "/b")
+        with self.assertRaises(AbsolutePathError):
+            self.mountbase.pp("a", Path("/b"))
+
+    def test_p(self):
+        self.assertEqual(self.mountbase.p("foo", "bar"), "/target/foo/bar")
+        with self.assertRaises(AbsolutePathError):
+            self.mountbase.p("foo", "/bar")
+
+
 class TestMounter(SubiTestCase):
     def setUp(self):
         self.model = Mock()
@@ -116,7 +145,7 @@ class TestMounter(SubiTestCase):
         # When we are bind mounting a directory, the destination should be
         # created as a directory.
         src = self.tmp_dir()
-        dst = pathlib.Path(self.tmp_dir()) / "dst"
+        dst = Path(self.tmp_dir()) / "dst"
 
         self.app.command_runner = AsyncMock()
         await mounter.bind_mount_tree(src, dst)
@@ -126,9 +155,9 @@ class TestMounter(SubiTestCase):
         mounter = Mounter(self.app)
         # When we are bind mounting a file, the destination should be created
         # as a file.
-        src = pathlib.Path(self.tmp_dir()) / "src"
+        src = Path(self.tmp_dir()) / "src"
         src.touch()
-        dst = pathlib.Path(self.tmp_dir()) / "dst"
+        dst = Path(self.tmp_dir()) / "dst"
 
         self.app.command_runner = AsyncMock()
         await mounter.bind_mount_tree(src, dst)
@@ -138,9 +167,9 @@ class TestMounter(SubiTestCase):
         mounter = Mounter(self.app)
         # When we are mounting a device, the destination should be created
         # as a directory.
-        src = pathlib.Path(self.tmp_dir()) / "src"
+        src = Path(self.tmp_dir()) / "src"
         src.touch()
-        dst = pathlib.Path(self.tmp_dir()) / "dst"
+        dst = Path(self.tmp_dir()) / "dst"
 
         self.app.command_runner = AsyncMock()
         await mounter.mount(src, dst)