diff --git a/subiquity/server/apt.py b/subiquity/server/apt.py index f11a71fe..6d47ad14 100644 --- a/subiquity/server/apt.py +++ b/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}") diff --git a/subiquity/server/mounter.py b/subiquity/server/mounter.py index 53d45d2c..5aa87696 100644 --- a/subiquity/server/mounter.py +++ b/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: diff --git a/subiquity/server/tests/test_mounter.py b/subiquity/server/tests/test_mounter.py index 093e85bf..2c0e4cf2 100644 --- a/subiquity/server/tests/test_mounter.py +++ b/subiquity/server/tests/test_mounter.py @@ -14,19 +14,48 @@ # along with this program. If not, see . 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)