From ef2a570e8100b2733536da800a9a6e55b0a10c26 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Mon, 21 Feb 2022 10:45:01 +0100 Subject: [PATCH 1/4] Add unit tests for apt.lowerdir_for Added some unit tests for translating different "types" of directories into a list of options suitable for mount -t overlay. The unit tests can also be used as examples to visualize how a given overlay would be configured depending on the parameters that are given for its creation. Signed-off-by: Olivier Gayot --- subiquity/server/tests/test_apt.py | 63 +++++++++++++++++++++++++++++- 1 file changed, 62 insertions(+), 1 deletion(-) 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) From 18c897f2d60c3e8fbbe2ff7520aed25687ccdbf8 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Mon, 21 Feb 2022 10:15:09 +0100 Subject: [PATCH 2/4] Add type hintings to overlay methods Signed-off-by: Olivier Gayot --- subiquity/server/apt.py | 63 ++++++++++++++++---------- subiquity/server/controllers/source.py | 3 +- 2 files changed, 41 insertions(+), 25 deletions(-) diff --git a/subiquity/server/apt.py b/subiquity/server/apt.py index 08cb588e..79526c6d 100644 --- a/subiquity/server/apt.py +++ b/subiquity/server/apt.py @@ -18,6 +18,7 @@ import logging import os import shutil import tempfile +from typing import List, Optional, Union from curtin.config import merge_config @@ -32,7 +33,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') @@ -44,15 +45,20 @@ class _MountBase: class Mountpoint(_MountBase): - def __init__(self, *, mountpoint): - self.mountpoint = mountpoint + def __init__(self, *, mountpoint: str): + self.mountpoint: str = mountpoint class OverlayMountpoint(_MountBase): - def __init__(self, *, lowers, upperdir, mountpoint): - self.lowers = lowers - self.upperdir = upperdir - self.mountpoint = mountpoint + def __init__(self, *, lowers, upperdir: Optional[str], mountpoint: str): + # The first element in lowers will be the bottom layer and the last + # element will be the top layer. + self.lowers: List[Lower] = lowers + self.upperdir: Optional[str] = upperdir + self.mountpoint: str = mountpoint + + +Lower = Union[Mountpoint, str, OverlayMountpoint] @functools.singledispatch @@ -119,10 +125,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 +151,11 @@ 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: Union[List[Lower], Lower]) \ + -> OverlayMountpoint: tdir = self.tdir() target = f'{tdir}/mount' lowerdir = lowerdir_for(lowers) @@ -221,23 +229,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 +254,22 @@ 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: Union[List[Lower], Lower]) \ + -> OverlayMountpoint: + # XXX This implementation expects that: + # - on first invocation, we pass a string as "lowers" + # - on second invocation, we pass the OverlayMountPoint returned by the + # first invocation. + if isinstance(lowers, OverlayMountpoint): + source = lowers.lowers[0] + else: + source = lowers target = self.tdir() os.mkdir(f'{target}/etc') await arun_command([ @@ -272,7 +287,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' From 7fb0c9b79d8852ec0a2663bb5d2caa3b1729e590 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Mon, 21 Feb 2022 14:47:49 +0100 Subject: [PATCH 3/4] Accept only sequence of lowers in setup_overlays AptConfigurer.setup_overlays() used to accept either a single Lower directory or a sequence of them. It oftentimes felt confusing so we now accept only a sequence of Lower's. If we want to pass a single Lower, we can just pass a list with a single element. Signed-off-by: Olivier Gayot --- subiquity/server/apt.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/subiquity/server/apt.py b/subiquity/server/apt.py index 79526c6d..1dd31b0d 100644 --- a/subiquity/server/apt.py +++ b/subiquity/server/apt.py @@ -154,8 +154,7 @@ class AptConfigurer: async def unmount(self, mountpoint: str): await self.app.command_runner.run(['umount', mountpoint]) - async def setup_overlay(self, lowers: Union[List[Lower], Lower]) \ - -> OverlayMountpoint: + async def setup_overlay(self, lowers: List[Lower]) -> OverlayMountpoint: tdir = self.tdir() target = f'{tdir}/mount' lowerdir = lowerdir_for(lowers) @@ -181,7 +180,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') @@ -195,7 +194,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( @@ -260,16 +259,17 @@ class AptConfigurer: class DryRunAptConfigurer(AptConfigurer): - async def setup_overlay(self, lowers: Union[List[Lower], Lower]) \ - -> OverlayMountpoint: + async def setup_overlay(self, lowers: List[Lower]) -> OverlayMountpoint: # XXX This implementation expects that: - # - on first invocation, we pass a string as "lowers" - # - on second invocation, we pass the OverlayMountPoint returned by the - # first invocation. - if isinstance(lowers, OverlayMountpoint): - source = lowers.lowers[0] + # - 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 = lowers + source = lowerdir target = self.tdir() os.mkdir(f'{target}/etc') await arun_command([ From 1c282546cc86df161cc92eb15f5df297f490d148 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Mon, 21 Feb 2022 15:01:10 +0100 Subject: [PATCH 4/4] Use attr classes to define Mountpoint and OverlayMountpoint Signed-off-by: Olivier Gayot --- subiquity/server/apt.py | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/subiquity/server/apt.py b/subiquity/server/apt.py index 1dd31b0d..9944e4bb 100644 --- a/subiquity/server/apt.py +++ b/subiquity/server/apt.py @@ -20,6 +20,8 @@ import shutil import tempfile from typing import List, Optional, Union +import attr + from curtin.config import merge_config from subiquitycore.file_util import write_file, generate_config_yaml @@ -44,18 +46,18 @@ class _MountBase: fp.write(content) +@attr.s(auto_attribs=True, kw_only=True) class Mountpoint(_MountBase): - def __init__(self, *, mountpoint: str): - self.mountpoint: str = mountpoint + mountpoint: str +@attr.s(auto_attribs=True, kw_only=True) class OverlayMountpoint(_MountBase): - def __init__(self, *, lowers, upperdir: Optional[str], mountpoint: str): - # The first element in lowers will be the bottom layer and the last - # element will be the top layer. - self.lowers: List[Lower] = lowers - self.upperdir: Optional[str] = upperdir - self.mountpoint: str = 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]