diff --git a/subiquity/server/apt.py b/subiquity/server/apt.py index 08cb588e..9944e4bb 100644 --- a/subiquity/server/apt.py +++ b/subiquity/server/apt.py @@ -18,6 +18,9 @@ import logging import os import shutil import tempfile +from typing import List, Optional, Union + +import attr from curtin.config import merge_config @@ -32,7 +35,7 @@ log = logging.getLogger('subiquity.server.apt') class _MountBase: - def p(self, *args): + def p(self, *args: str) -> str: for a in args: if a.startswith('/'): raise Exception('no absolute paths here please') @@ -43,16 +46,21 @@ class _MountBase: fp.write(content) +@attr.s(auto_attribs=True, kw_only=True) class Mountpoint(_MountBase): - def __init__(self, *, mountpoint): - self.mountpoint = mountpoint + mountpoint: str +@attr.s(auto_attribs=True, kw_only=True) class OverlayMountpoint(_MountBase): - def __init__(self, *, lowers, upperdir, mountpoint): - self.lowers = lowers - self.upperdir = upperdir - self.mountpoint = mountpoint + # The first element in lowers will be the bottom layer and the last element + # will be the top layer. + lowers: List["Lower"] + upperdir: Optional[str] + mountpoint: str + + +Lower = Union[Mountpoint, str, OverlayMountpoint] @functools.singledispatch @@ -119,10 +127,11 @@ class AptConfigurer: # system, or if it is not, just copy /var/lib/apt/lists from the # 'configured_tree' overlay. - def __init__(self, app, source): + def __init__(self, app, source: str): self.app = app - self.source = source - self.configured_tree = None + self.source: str = source + self.configured_tree: Optional[OverlayMountpoint] = None + self.install_tree: Optional[OverlayMountpoint] = None self.install_mount = None self._mounts = [] self._tdirs = [] @@ -144,10 +153,10 @@ class AptConfigurer: self._mounts.append(m) return m - async def unmount(self, mountpoint): + async def unmount(self, mountpoint: str): await self.app.command_runner.run(['umount', mountpoint]) - async def setup_overlay(self, lowers): + async def setup_overlay(self, lowers: List[Lower]) -> OverlayMountpoint: tdir = self.tdir() target = f'{tdir}/mount' lowerdir = lowerdir_for(lowers) @@ -173,7 +182,7 @@ class AptConfigurer: return {'apt': cfg} async def apply_apt_config(self, context): - self.configured_tree = await self.setup_overlay(self.source) + self.configured_tree = await self.setup_overlay([self.source]) config_location = os.path.join( self.app.root, 'var/log/installer/subiquity-curtin-apt.conf') @@ -187,7 +196,7 @@ class AptConfigurer: async def configure_for_install(self, context): assert self.configured_tree is not None - self.install_tree = await self.setup_overlay(self.configured_tree) + self.install_tree = await self.setup_overlay([self.configured_tree]) os.mkdir(self.install_tree.p('cdrom')) await self.mount( @@ -221,23 +230,23 @@ class AptConfigurer: for d in self._tdirs: shutil.rmtree(d) - async def deconfigure(self, context, target): - target = Mountpoint(mountpoint=target) + async def deconfigure(self, context, target: str) -> None: + target_mnt = Mountpoint(mountpoint=target) async def _restore_dir(dir): - shutil.rmtree(target.p(dir)) + shutil.rmtree(target_mnt.p(dir)) await self.app.command_runner.run([ - 'cp', '-aT', self.configured_tree.p(dir), target.p(dir), + 'cp', '-aT', self.configured_tree.p(dir), target_mnt.p(dir), ]) - await self.unmount(target.p('cdrom')) - os.rmdir(target.p('cdrom')) + await self.unmount(target_mnt.p('cdrom')) + os.rmdir(target_mnt.p('cdrom')) await _restore_dir('etc/apt') if self.app.base_model.network.has_network: await run_curtin_command( - self.app, context, "in-target", "-t", target.p(), + self.app, context, "in-target", "-t", target_mnt.p(), "--", "apt-get", "update") else: await _restore_dir('var/lib/apt/lists') @@ -246,15 +255,23 @@ class AptConfigurer: if self.app.base_model.network.has_network: await run_curtin_command( - self.app, context, "in-target", "-t", target.p(), + self.app, context, "in-target", "-t", target_mnt.p(), "--", "apt-get", "update") class DryRunAptConfigurer(AptConfigurer): - async def setup_overlay(self, source): - if isinstance(source, OverlayMountpoint): - source = source.lowers[0] + async def setup_overlay(self, lowers: List[Lower]) -> OverlayMountpoint: + # XXX This implementation expects that: + # - on first invocation, the lowers list contains a single string + # element. + # - on second invocation, the lowers list contains the + # OverlayMountPoint returned by the first invocation. + lowerdir = lowers[0] + if isinstance(lowerdir, OverlayMountpoint): + source = lowerdir.lowers[0] + else: + source = lowerdir target = self.tdir() os.mkdir(f'{target}/etc') await arun_command([ @@ -272,7 +289,7 @@ class DryRunAptConfigurer(AptConfigurer): return -def get_apt_configurer(app, source): +def get_apt_configurer(app, source: str): if app.opts.dry_run: return DryRunAptConfigurer(app, source) else: diff --git a/subiquity/server/controllers/source.py b/subiquity/server/controllers/source.py index c76f61ca..385ed3cf 100644 --- a/subiquity/server/controllers/source.py +++ b/subiquity/server/controllers/source.py @@ -13,6 +13,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from typing import Optional import os from curtin.commands.extract import get_handler_for_source @@ -54,7 +55,7 @@ class SourceController(SubiquityController): def __init__(self, app): super().__init__(app) self._handler = None - self.source_path = None + self.source_path: Optional[str] = None def start(self): path = '/cdrom/casper/install-sources.yaml' diff --git a/subiquity/server/tests/test_apt.py b/subiquity/server/tests/test_apt.py index 759a6e85..40111536 100644 --- a/subiquity/server/tests/test_apt.py +++ b/subiquity/server/tests/test_apt.py @@ -17,7 +17,12 @@ from unittest.mock import Mock from subiquitycore.tests import SubiTestCase from subiquitycore.tests.mocks import make_app -from subiquity.server.apt import AptConfigurer +from subiquity.server.apt import ( + AptConfigurer, + Mountpoint, + OverlayMountpoint, + lowerdir_for, +) from subiquity.models.mirror import MirrorModel, DEFAULT from subiquity.models.proxy import ProxyModel @@ -44,3 +49,59 @@ class TestAptConfigurer(SubiTestCase): expected['apt']['http_proxy'] = proxy expected['apt']['https_proxy'] = proxy self.assertEqual(expected, self.configurer.apt_config()) + + +class TestLowerDirFor(SubiTestCase): + def test_lowerdir_for_str(self): + self.assertEqual( + lowerdir_for("/tmp/lower1"), + "/tmp/lower1") + + def test_lowerdir_for_str_list(self): + self.assertEqual( + lowerdir_for(["/tmp/lower1", "/tmp/lower2"]), + "/tmp/lower2:/tmp/lower1") + + def test_lowerdir_for_mountpoint(self): + self.assertEqual( + lowerdir_for(Mountpoint(mountpoint="/mnt")), + "/mnt") + + def test_lowerdir_for_simple_overlay(self): + overlay = OverlayMountpoint( + lowers=["/tmp/lower1"], + upperdir="/tmp/upper1", + mountpoint="/mnt", + ) + self.assertEqual(lowerdir_for(overlay), "/tmp/upper1:/tmp/lower1") + + def test_lowerdir_for_overlay(self): + overlay = OverlayMountpoint( + lowers=["/tmp/lower1", "/tmp/lower2"], + upperdir="/tmp/upper1", + mountpoint="/mnt", + ) + self.assertEqual( + lowerdir_for(overlay), + "/tmp/upper1:/tmp/lower2:/tmp/lower1") + + def test_lowerdir_for_list(self): + overlay = OverlayMountpoint( + lowers=["/tmp/overlaylower1", "/tmp/overlaylower2"], + upperdir="/tmp/overlayupper1", + mountpoint="/mnt/overlay", + ) + mountpoint = Mountpoint(mountpoint="/mnt/mountpoint") + lowers = ["/tmp/lower1", "/tmp/lower2"] + self.assertEqual( + lowerdir_for([overlay, mountpoint, lowers]), + "/tmp/lower2:/tmp/lower1" + + ":/mnt/mountpoint" + + ":/tmp/overlayupper1:/tmp/overlaylower2:/tmp/overlaylower1") + + def test_lowerdir_for_other(self): + with self.assertRaises(NotImplementedError): + lowerdir_for(None) + + with self.assertRaises(NotImplementedError): + lowerdir_for(10)