StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1188 from ogayot/modernize-overlay 

Add type hints to overlayfs helpers and add some unit tests.
This commit is contained in:
Olivier Gayot 2022-02-27 17:12:09 +01:00 committed by GitHub
parent 0f987ee6ba 1c282546cc
commit 43090a46ba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 107 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
subiquity/server/apt.py
View File

@ -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:

3
subiquity/server/controllers/source.py
View File

@ -13,6 +13,7 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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'

63
subiquity/server/tests/test_apt.py
View File

@ -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)