curtin-dev team mailing list archive
-
curtin-dev team
-
Mailing list archive
-
Message #02981
[Merge] ~mwhudson/curtin:more-grub-config-object into curtin:master
Michael Hudson-Doyle has proposed merging ~mwhudson/curtin:more-grub-config-object into curtin:master.
Commit message:
extend grub config object to handle all config under the 'grub' key
Pinched a hacked down version of my deserialization code from
subiquity...
I'm not sure this 100% preserves the behaviour of install_devices being
an empty list, None or absent entirely, the semantics around all that
seem to be pretty confused.
Requested reviews:
curtin developers (curtin-dev)
For more details, see:
https://code.launchpad.net/~mwhudson/curtin/+git/curtin/+merge/444608
--
Your team curtin developers is requested to review the proposed merge of ~mwhudson/curtin:more-grub-config-object into curtin:master.
diff --git a/curtin/commands/curthooks.py b/curtin/commands/curthooks.py
index 868fbad..759ed5e 100644
--- a/curtin/commands/curthooks.py
+++ b/curtin/commands/curthooks.py
@@ -423,9 +423,8 @@ def install_kernel(cfg, target):
" System may not boot.", package)
-def uefi_remove_old_loaders(grubcfg: dict, target):
+def uefi_remove_old_loaders(grubcfg: config.GrubConfig, target):
"""Removes the old UEFI loaders from efibootmgr."""
- grubcfg = config.fromdict(config.GrubConfig, grubcfg)
efi_state = util.get_efibootmgr(target)
LOG.debug('UEFI remove old olders efi state:\n%s', efi_state)
@@ -517,7 +516,7 @@ def _reorder_new_entry(
def uefi_reorder_loaders(
- grubcfg: dict,
+ grubcfg: config.GrubConfig,
target: str,
efi_orig_state: util.EFIBootState,
variant: str,
@@ -533,8 +532,6 @@ def uefi_reorder_loaders(
is installed after the the previous first entry (before we installed grub).
"""
- grubcfg = config.fromdict(config.GrubConfig, grubcfg)
-
if not grubcfg.reorder_uefi:
LOG.debug("Skipped reordering of UEFI boot methods.")
LOG.debug("Currently booted UEFI loader might no longer boot.")
@@ -578,9 +575,10 @@ def uefi_reorder_loaders(
in_chroot.subp(['efibootmgr', '-o', new_boot_order])
-def uefi_remove_duplicate_entries(grubcfg: dict, target: str) -> None:
- grubcfg = config.fromdict(config.GrubConfig, grubcfg)
-
+def uefi_remove_duplicate_entries(
+ grubcfg: config.GrubConfig,
+ target: str,
+ ) -> None:
if not grubcfg.remove_duplicate_entries:
LOG.debug("Skipped removing duplicate UEFI boot entries per config.")
return
@@ -725,7 +723,12 @@ def uefi_find_grub_device_ids(sconfig):
return grub_device_ids
-def setup_grub(cfg, target, osfamily, variant):
+def setup_grub(
+ cfg: dict,
+ target: str,
+ osfamily: str,
+ variant: str,
+ ) -> None:
# target is the path to the mounted filesystem
# FIXME: these methods need moving to curtin.block
@@ -733,11 +736,13 @@ def setup_grub(cfg, target, osfamily, variant):
from curtin.commands.block_meta import (extract_storage_ordered_dict,
get_path_to_storage_volume)
- grubcfg = cfg.get('grub', {})
+ grubcfg_d = cfg.get('grub', {})
# copy legacy top level name
- if 'grub_install_devices' in cfg and 'install_devices' not in grubcfg:
- grubcfg['install_devices'] = cfg['grub_install_devices']
+ if 'grub_install_devices' in cfg and 'install_devices' not in grubcfg_d:
+ grubcfg_d['install_devices'] = cfg['grub_install_devices']
+
+ grubcfg = config.fromdict(config.GrubConfig, grubcfg_d)
LOG.debug("setup grub on target %s", target)
# if there is storage config, look for devices tagged with 'grub_device'
@@ -763,19 +768,19 @@ def setup_grub(cfg, target, osfamily, variant):
get_path_to_storage_volume(item_id, storage_cfg_odict))
if len(storage_grub_devices) > 0:
- if len(grubcfg.get('install_devices', [])):
+ if grubcfg.install_devices is not None and \
+ len(grubcfg.install_devices) > 0:
LOG.warn("Storage Config grub device config takes precedence "
"over grub 'install_devices' value, ignoring: %s",
grubcfg['install_devices'])
- grubcfg['install_devices'] = storage_grub_devices
-
- LOG.debug("install_devices: %s", grubcfg.get('install_devices'))
- if 'install_devices' in grubcfg:
- instdevs = grubcfg.get('install_devices')
- if isinstance(instdevs, str):
- instdevs = [instdevs]
- if instdevs is None:
- LOG.debug("grub installation disabled by config")
+ grubcfg.install_devices = storage_grub_devices
+
+ LOG.debug("install_devices: %s", grubcfg.install_devices)
+ if grubcfg.install_devices is None:
+ LOG.debug("grub installation disabled by config")
+ instdevs = grubcfg.install_devices
+ elif len(grubcfg.install_devices) > 0:
+ instdevs = grubcfg.install_devices
else:
# If there were no install_devices found then we try to do the right
# thing. That right thing is basically installing on all block
@@ -823,7 +828,7 @@ def setup_grub(cfg, target, osfamily, variant):
else:
instdevs = ["none"]
- update_nvram = grubcfg.get('update_nvram', True)
+ update_nvram = grubcfg.update_nvram
if uefi_bootable and update_nvram:
efi_orig_state = util.get_efibootmgr(target)
uefi_remove_old_loaders(grubcfg, target)
diff --git a/curtin/commands/install_grub.py b/curtin/commands/install_grub.py
index 03b4670..2faa2c6 100644
--- a/curtin/commands/install_grub.py
+++ b/curtin/commands/install_grub.py
@@ -3,6 +3,7 @@ import re
import platform
import shutil
import sys
+from typing import List, Optional
from curtin import block
from curtin import config
@@ -137,7 +138,7 @@ def prepare_grub_dir(target, grub_cfg):
shutil.move(ci_cfg, ci_cfg + '.disabled')
-def get_carryover_params(distroinfo):
+def get_carryover_params(distroinfo) -> str:
# return a string to append to installed systems boot parameters
# it may include a '--' after a '---'
# see LP: 1402042 for some history here.
@@ -206,14 +207,17 @@ def replace_grub_cmdline_linux_default(target, new_args):
LOG.debug('updated %s to set: %s', target_grubconf, newcontent)
-def write_grub_config(target, grubcfg, grub_conf, new_params):
- replace_default = config.value_as_boolean(
- grubcfg.get('replace_linux_default', True))
+def write_grub_config(
+ target: str,
+ grubcfg: config.GrubConfig,
+ grub_conf: str,
+ new_params: str,
+ ) -> None:
+ replace_default = grubcfg.replace_linux_default
if replace_default:
replace_grub_cmdline_linux_default(target, new_params)
- probe_os = config.value_as_boolean(
- grubcfg.get('probe_additional_os', False))
+ probe_os = grubcfg.probe_additional_os
if not probe_os:
probe_content = [
('# Curtin disable grub os prober that might find other '
@@ -224,10 +228,7 @@ def write_grub_config(target, grubcfg, grub_conf, new_params):
"\n".join(probe_content), omode='a+')
# if terminal is present in config, but unset, then don't
- grub_terminal = grubcfg.get('terminal', 'console')
- if not isinstance(grub_terminal, str):
- raise ValueError("Unexpected value %s for 'terminal'. "
- "Value must be a string" % grub_terminal)
+ grub_terminal = grubcfg.terminal
if not grub_terminal.lower() == "unmodified":
terminal_content = [
'# Curtin configured GRUB_TERMINAL value',
@@ -394,7 +395,13 @@ def check_target_arch_machine(target, arch=None, machine=None, uefi=None):
raise RuntimeError(errmsg)
-def install_grub(devices, target, uefi=None, grubcfg=None):
+def install_grub(
+ devices: List[str],
+ target: str,
+ *,
+ grubcfg: config.GrubConfig,
+ uefi: Optional[bool] = None,
+ ):
"""Install grub to devices inside target chroot.
:param: devices: List of block device paths to install grub upon.
@@ -411,8 +418,8 @@ def install_grub(devices, target, uefi=None, grubcfg=None):
raise ValueError("Invalid parameter 'target': %s" % target)
LOG.debug("installing grub to target=%s devices=%s [replace_defaults=%s]",
- target, devices, grubcfg.get('replace_default'))
- update_nvram = config.value_as_boolean(grubcfg.get('update_nvram', True))
+ target, devices, grubcfg.replace_linux_default)
+ update_nvram = grubcfg.update_nvram
distroinfo = distro.get_distroinfo(target=target)
target_arch = distro.get_architecture(target=target)
rhel_ver = (distro.rpm_get_dist_id(target)
@@ -460,7 +467,7 @@ def install_grub_main(args):
cfg = config.load_command_config(args, state)
stack_prefix = state.get('report_stack_prefix', '')
uefi = util.is_uefi_bootable()
- grubcfg = cfg.get('grub')
+ grubcfg = config.fromdict(config.GrubConfig, cfg.get('grub'))
with events.ReportEventStack(
name=stack_prefix, reporting_enabled=True, level="INFO",
description="Installing grub to target devices"):
diff --git a/curtin/config.py b/curtin/config.py
index b65410b..39898a4 100644
--- a/curtin/config.py
+++ b/curtin/config.py
@@ -1,6 +1,7 @@
# This file is part of curtin. See LICENSE file for copyright and license info.
import json
+import typing
import attr
import yaml
@@ -129,23 +130,142 @@ def value_as_boolean(value):
return value not in false_values
+def _convert_install_devices(value):
+ if isinstance(value, str):
+ return [value]
+ return value
+
+
@attr.s(auto_attribs=True)
class GrubConfig:
- # This is not yet every option that appears under the "grub" config key,
- # but it is a work in progress.
+ install_devices: typing.Optional[typing.List[str]] = attr.ib(
+ converter=_convert_install_devices, default=attr.Factory(list))
+ probe_additional_os: bool = attr.ib(
+ default=False, converter=value_as_boolean)
+ remove_duplicate_entries: bool = True
remove_old_uefi_loaders: bool = True
reorder_uefi: bool = True
reorder_uefi_force_fallback: bool = attr.ib(
default=False, converter=value_as_boolean)
- remove_duplicate_entries: bool = True
+ replace_linux_default: bool = attr.ib(
+ default=True, converter=value_as_boolean)
+ terminal: str = "console"
+ update_nvram: bool = attr.ib(default=True, converter=value_as_boolean)
-def fromdict(cls, d):
- kw = {}
- for field in attr.fields(cls):
- if field.name in d:
- kw[field.name] = d[field.name]
- return cls(**kw)
+class SerializationError(Exception):
+ def __init__(self, obj, path, message):
+ self.obj = obj
+ self.path = path
+ self.message = message
+
+ def __str__(self):
+ p = self.path
+ if not p:
+ p = 'top-level'
+ return f"processing {self.obj}: at {p}, {self.message}"
+
+
+@attr.s(auto_attribs=True)
+class SerializationContext:
+ obj: typing.Any
+ cur: typing.Any
+ path: str
+ metadata: typing.Optional[typing.Dict]
+
+ @classmethod
+ def new(cls, obj):
+ return SerializationContext(obj, obj, '', {})
+
+ def child(self, path, cur, metadata=None):
+ if metadata is None:
+ metadata = self.metadata
+ return attr.evolve(
+ self, path=self.path + path, cur=cur, metadata=metadata)
+
+ def error(self, message):
+ raise SerializationError(self.obj, self.path, message)
+
+ def assert_type(self, typ):
+ if type(self.cur) is not typ:
+ self.error("{!r} is not a {}".format(self.cur, typ))
+
+
+class Deserializer:
+
+ def __init__(self):
+ self.typing_walkers = {
+ list: self._walk_List,
+ typing.List: self._walk_List,
+ typing.Union: self._walk_Union,
+ }
+ self.type_deserializers = {}
+ for typ in int, str, bool, list, dict, type(None):
+ self.type_deserializers[typ] = self._scalar
+
+ def _scalar(self, annotation, context):
+ context.assert_type(annotation)
+ return context.cur
+
+ def _walk_List(self, meth, args, context):
+ return [
+ meth(args[0], context.child(f'[{i}]', v))
+ for i, v in enumerate(context.cur)
+ ]
+
+ def _walk_Union(self, meth, args, context):
+ NoneType = type(None)
+ if NoneType in args:
+ args = [a for a in args if a is not NoneType]
+ if len(args) == 1:
+ # I.e. Optional[thing]
+ if context.cur is None:
+ return context.cur
+ return meth(args[0], context)
+ context.error(f"cannot serialize Union[{args}]")
+
+ def _deserialize_attr(self, annotation, context):
+ context.assert_type(dict)
+ args = {}
+ fields = {
+ field.name: field for field in attr.fields(annotation)
+ }
+ for key, value in context.cur.items():
+ if key not in fields:
+ continue
+ field = fields[key]
+ if field.converter:
+ value = field.converter(value)
+ args[field.name] = self._deserialize(
+ field.type,
+ context.child(f'[{key!r}]', value, field.metadata))
+ return annotation(**args)
+
+ def _deserialize(self, annotation, context):
+ if annotation is None:
+ context.assert_type(type(None))
+ return None
+ if annotation is typing.Any:
+ return context.cur
+ if attr.has(annotation):
+ return self._deserialize_attr(annotation, context)
+ origin = getattr(annotation, '__origin__', None)
+ if origin is not None:
+ return self.typing_walkers[origin](
+ self._deserialize, annotation.__args__, context)
+ return self.type_deserializers[annotation](annotation, context)
+
+ def deserialize(self, annotation, value):
+ context = SerializationContext.new(value)
+ return self._deserialize(annotation, context)
+
+
+T = typing.TypeVar("T")
+
+
+def fromdict(cls: typing.Type[T], d) -> T:
+ deserializer = Deserializer()
+ return deserializer.deserialize(cls, d)
# vi: ts=4 expandtab syntax=python
diff --git a/tests/unittests/test_commands_install_grub.py b/tests/unittests/test_commands_install_grub.py
index 00004d7..fc19da3 100644
--- a/tests/unittests/test_commands_install_grub.py
+++ b/tests/unittests/test_commands_install_grub.py
@@ -1,5 +1,6 @@
# This file is part of curtin. See LICENSE file for copyright and license info.
+from curtin import config
from curtin import distro
from curtin import util
from curtin import paths
@@ -456,7 +457,7 @@ class TestWriteGrubConfig(CiTestCase):
self.assertEqual(expected, found)
def test_write_grub_config_defaults(self):
- grubcfg = {}
+ grubcfg = config.GrubConfig()
new_params = ['foo=bar', 'wark=1']
expected_default = "\n".join([
'GRUB_CMDLINE_LINUX_DEFAULT="foo=bar wark=1"', ''])
@@ -473,7 +474,7 @@ class TestWriteGrubConfig(CiTestCase):
self._verify_expected(expected_default, expected_curtin)
def test_write_grub_config_no_replace(self):
- grubcfg = {'replace_linux_default': False}
+ grubcfg = config.GrubConfig(replace_linux_default=False)
new_params = ['foo=bar', 'wark=1']
expected_default = "\n".join([])
expected_curtin = "\n".join([
@@ -489,7 +490,7 @@ class TestWriteGrubConfig(CiTestCase):
self._verify_expected(expected_default, expected_curtin)
def test_write_grub_config_disable_probe(self):
- grubcfg = {'probe_additional_os': False} # DISABLE_OS_PROBER=1
+ grubcfg = config.GrubConfig(probe_additional_os=False)
new_params = ['foo=bar', 'wark=1']
expected_default = "\n".join([
'GRUB_CMDLINE_LINUX_DEFAULT="foo=bar wark=1"', ''])
@@ -506,7 +507,7 @@ class TestWriteGrubConfig(CiTestCase):
self._verify_expected(expected_default, expected_curtin)
def test_write_grub_config_enable_probe(self):
- grubcfg = {'probe_additional_os': True} # DISABLE_OS_PROBER=0, default
+ grubcfg = config.GrubConfig(probe_additional_os=True)
new_params = ['foo=bar', 'wark=1']
expected_default = "\n".join([
'GRUB_CMDLINE_LINUX_DEFAULT="foo=bar wark=1"', ''])
@@ -520,10 +521,9 @@ class TestWriteGrubConfig(CiTestCase):
self._verify_expected(expected_default, expected_curtin)
def test_write_grub_config_no_grub_settings_file(self):
- grubcfg = {
- 'probe_additional_os': True,
- 'terminal': 'unmodified',
- }
+ grubcfg = config.GrubConfig(
+ probe_additional_os=True,
+ terminal='unmodified')
new_params = []
install_grub.write_grub_config(
self.target, grubcfg, self.grubconf, new_params)
@@ -531,7 +531,8 @@ class TestWriteGrubConfig(CiTestCase):
self.assertFalse(os.path.exists(self.target_grubconf))
def test_write_grub_config_specify_terminal(self):
- grubcfg = {'terminal': 'serial'}
+ grubcfg = config.GrubConfig(
+ terminal='serial')
new_params = ['foo=bar', 'wark=1']
expected_default = "\n".join([
'GRUB_CMDLINE_LINUX_DEFAULT="foo=bar wark=1"', ''])
@@ -548,7 +549,7 @@ class TestWriteGrubConfig(CiTestCase):
self._verify_expected(expected_default, expected_curtin)
def test_write_grub_config_terminal_unmodified(self):
- grubcfg = {'terminal': 'unmodified'}
+ grubcfg = config.GrubConfig(terminal='unmodified')
new_params = ['foo=bar', 'wark=1']
expected_default = "\n".join([
'GRUB_CMDLINE_LINUX_DEFAULT="foo=bar wark=1"', ''])
@@ -562,13 +563,6 @@ class TestWriteGrubConfig(CiTestCase):
self._verify_expected(expected_default, expected_curtin)
- def test_write_grub_config_invalid_terminal(self):
- grubcfg = {'terminal': ['color-tv']}
- new_params = ['foo=bar', 'wark=1']
- with self.assertRaises(ValueError):
- install_grub.write_grub_config(
- self.target, grubcfg, self.grubconf, new_params)
-
class TestFindEfiLoader(CiTestCase):
@@ -1119,39 +1113,43 @@ class TestInstallGrub(CiTestCase):
def test_grub_install_raise_exception_on_no_devices(self):
devices = []
with self.assertRaises(ValueError):
- install_grub.install_grub(devices, self.target, False, {})
+ install_grub.install_grub(
+ devices, self.target, uefi=False, grubcfg=config.GrubConfig())
def test_grub_install_raise_exception_on_no_target(self):
devices = ['foobar']
with self.assertRaises(ValueError):
- install_grub.install_grub(devices, None, False, {})
+ install_grub.install_grub(
+ devices, None, uefi=False, grubcfg=config.GrubConfig())
def test_grub_install_raise_exception_on_s390x(self):
self.m_distro_get_architecture.return_value = 's390x'
self.m_platform_machine.return_value = 's390x'
devices = ['foobar']
with self.assertRaises(RuntimeError):
- install_grub.install_grub(devices, self.target, False, {})
+ install_grub.install_grub(
+ devices, self.target, uefi=False, grubcfg=config.GrubConfig())
def test_grub_install_raise_exception_on_armv7(self):
self.m_distro_get_architecture.return_value = 'armhf'
self.m_platform_machine.return_value = 'armv7l'
devices = ['foobar']
with self.assertRaises(RuntimeError):
- install_grub.install_grub(devices, self.target, False, {})
+ install_grub.install_grub(
+ devices, self.target, uefi=False, grubcfg=config.GrubConfig())
def test_grub_install_raise_exception_on_arm64_no_uefi(self):
self.m_distro_get_architecture.return_value = 'arm64'
self.m_platform_machine.return_value = 'aarch64'
- uefi = False
devices = ['foobar']
with self.assertRaises(RuntimeError):
- install_grub.install_grub(devices, self.target, uefi, {})
+ install_grub.install_grub(
+ devices, self.target, uefi=False, grubcfg=config.GrubConfig())
def test_grub_install_ubuntu(self):
devices = ['/dev/disk-a-part1']
uefi = False
- grubcfg = {}
+ grubcfg = config.GrubConfig()
grub_conf = self.tmp_path('grubconf')
new_params = []
self.m_get_grub_package_name.return_value = ('grub-pc', 'i386-pc')
@@ -1161,7 +1159,8 @@ class TestInstallGrub(CiTestCase):
self.m_gen_install_commands.return_value = (
[['/bin/true']], [['/bin/false']])
- install_grub.install_grub(devices, self.target, uefi, grubcfg)
+ install_grub.install_grub(
+ devices, self.target, uefi=uefi, grubcfg=grubcfg)
self.m_distro_get_distroinfo.assert_called_with(target=self.target)
self.m_distro_get_architecture.assert_called_with(target=self.target)
@@ -1189,8 +1188,7 @@ class TestInstallGrub(CiTestCase):
def test_uefi_grub_install_ubuntu(self):
devices = ['/dev/disk-a-part1']
uefi = True
- update_nvram = True
- grubcfg = {'update_nvram': update_nvram}
+ grubcfg = config.GrubConfig(update_nvram=True)
grub_conf = self.tmp_path('grubconf')
new_params = []
grub_name = 'grub-efi-amd64'
@@ -1203,7 +1201,8 @@ class TestInstallGrub(CiTestCase):
self.m_gen_uefi_install_commands.return_value = (
[['/bin/true']], [['/bin/false']])
- install_grub.install_grub(devices, self.target, uefi, grubcfg)
+ install_grub.install_grub(
+ devices, self.target, uefi=uefi, grubcfg=grubcfg)
self.m_distro_get_distroinfo.assert_called_with(target=self.target)
self.m_distro_get_architecture.assert_called_with(target=self.target)
@@ -1219,8 +1218,8 @@ class TestInstallGrub(CiTestCase):
self.m_get_grub_install_command.assert_called_with(
uefi, self.distroinfo, self.target)
self.m_gen_uefi_install_commands.assert_called_with(
- grub_name, grub_target, grub_cmd, update_nvram, self.distroinfo,
- devices, self.target)
+ grub_name, grub_target, grub_cmd, grubcfg.update_nvram,
+ self.distroinfo, devices, self.target)
self.m_subp.assert_has_calls([
mock.call(['/bin/true'], env=self.env, capture=True,
@@ -1232,8 +1231,7 @@ class TestInstallGrub(CiTestCase):
def test_uefi_grub_install_ubuntu_multiple_esp(self):
devices = ['/dev/disk-a-part1']
uefi = True
- update_nvram = True
- grubcfg = {'update_nvram': update_nvram}
+ grubcfg = config.GrubConfig(update_nvram=True)
grub_conf = self.tmp_path('grubconf')
new_params = []
grub_name = 'grub-efi-amd64'
@@ -1246,7 +1244,8 @@ class TestInstallGrub(CiTestCase):
self.m_gen_uefi_install_commands.return_value = (
[['/bin/true']], [['/bin/false']])
- install_grub.install_grub(devices, self.target, uefi, grubcfg)
+ install_grub.install_grub(
+ devices, self.target, uefi=uefi, grubcfg=grubcfg)
self.m_distro_get_distroinfo.assert_called_with(target=self.target)
self.m_distro_get_architecture.assert_called_with(target=self.target)
@@ -1262,8 +1261,8 @@ class TestInstallGrub(CiTestCase):
self.m_get_grub_install_command.assert_called_with(
uefi, self.distroinfo, self.target)
self.m_gen_uefi_install_commands.assert_called_with(
- grub_name, grub_target, grub_cmd, update_nvram, self.distroinfo,
- devices, self.target)
+ grub_name, grub_target, grub_cmd, grubcfg.update_nvram,
+ self.distroinfo, devices, self.target)
self.m_subp.assert_has_calls([
mock.call(['/bin/true'], env=self.env, capture=True,
diff --git a/tests/unittests/test_config.py b/tests/unittests/test_config.py
index af7f251..ae51744 100644
--- a/tests/unittests/test_config.py
+++ b/tests/unittests/test_config.py
@@ -3,6 +3,9 @@
import copy
import json
import textwrap
+import typing
+
+import attr
from curtin import config
from .helpers import CiTestCase
@@ -139,4 +142,58 @@ def _replace_consts(cfgstr):
cfgstr = cfgstr.replace(k, v)
return cfgstr
+
+class TestDeserializer(CiTestCase):
+
+ def test_scalar(self):
+ deserializer = config.Deserializer()
+ self.assertEqual(1, deserializer.deserialize(int, 1))
+ self.assertEqual("a", deserializer.deserialize(str, "a"))
+
+ def test_attr(self):
+ deserializer = config.Deserializer()
+
+ @attr.s(auto_attribs=True)
+ class Point:
+ x: int
+ y: int
+
+ self.assertEqual(
+ Point(x=1, y=2),
+ deserializer.deserialize(Point, {'x': 1, 'y': 2}))
+
+ def test_list(self):
+ deserializer = config.Deserializer()
+ self.assertEqual(
+ [1, 2, 3],
+ deserializer.deserialize(typing.List[int], [1, 2, 3]))
+
+ def test_optional(self):
+ deserializer = config.Deserializer()
+ self.assertEqual(
+ 1,
+ deserializer.deserialize(typing.Optional[int], 1))
+ self.assertEqual(
+ None,
+ deserializer.deserialize(typing.Optional[int], None))
+
+ def test_converter(self):
+ deserializer = config.Deserializer()
+
+ @attr.s(auto_attribs=True)
+ class WithoutConverter:
+ val: bool
+
+ with self.assertRaises(config.SerializationError):
+ deserializer.deserialize(WithoutConverter, {"val": "on"})
+
+ @attr.s(auto_attribs=True)
+ class WithConverter:
+ val: bool = attr.ib(converter=config.value_as_boolean)
+
+ self.assertEqual(
+ WithConverter(val=True),
+ deserializer.deserialize(WithConverter, {"val": "on"}))
+
+
# vi: ts=4 expandtab syntax=python
diff --git a/tests/unittests/test_curthooks.py b/tests/unittests/test_curthooks.py
index 6067ea4..d6b5445 100644
--- a/tests/unittests/test_curthooks.py
+++ b/tests/unittests/test_curthooks.py
@@ -636,13 +636,11 @@ class TestSetupGrub(CiTestCase):
cfg = {
'grub_install_devices': ['/dev/vdb']
}
- updated_cfg = {
- 'install_devices': ['/dev/vdb']
- }
curthooks.setup_grub(cfg, self.target, osfamily=self.distro_family,
variant=self.variant)
self.m_install_grub.assert_called_with(
- ['/dev/vdb'], self.target, uefi=False, grubcfg=updated_cfg)
+ ['/dev/vdb'], self.target, uefi=False,
+ grubcfg=config.GrubConfig(install_devices=['/dev/vdb']))
def test_uses_install_devices_in_grubcfg(self):
cfg = {
@@ -654,7 +652,8 @@ class TestSetupGrub(CiTestCase):
cfg, self.target,
osfamily=self.distro_family, variant=self.variant)
self.m_install_grub.assert_called_with(
- ['/dev/vdb'], self.target, uefi=False, grubcfg=cfg.get('grub'))
+ ['/dev/vdb'], self.target, uefi=False,
+ grubcfg=config.fromdict(config.GrubConfig, cfg.get('grub')))
@patch('curtin.commands.block_meta.multipath')
@patch('curtin.commands.curthooks.os.path.exists')
@@ -678,7 +677,7 @@ class TestSetupGrub(CiTestCase):
variant=self.variant)
self.m_install_grub.assert_called_with(
['/dev/vdb'], self.target, uefi=False,
- grubcfg={'install_devices': ['/dev/vdb']})
+ grubcfg=config.GrubConfig(install_devices=['/dev/vdb']))
@patch('curtin.commands.block_meta.multipath')
@patch('curtin.block.is_valid_device')
@@ -729,8 +728,9 @@ class TestSetupGrub(CiTestCase):
variant='centos')
self.m_install_grub.assert_called_with(
['/dev/vdb1'], self.target, uefi=True,
- grubcfg={'update_nvram': False, 'install_devices': ['/dev/vdb1']}
- )
+ grubcfg=config.GrubConfig(
+ update_nvram=False,
+ install_devices=['/dev/vdb1']))
def test_grub_install_installs_to_none_if_install_devices_None(self):
cfg = {
@@ -742,7 +742,7 @@ class TestSetupGrub(CiTestCase):
variant=self.variant)
self.m_install_grub.assert_called_with(
['none'], self.target, uefi=False,
- grubcfg={'install_devices': None}
+ grubcfg=config.GrubConfig(install_devices=None),
)
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
@@ -772,7 +772,8 @@ class TestSetupGrub(CiTestCase):
curthooks.setup_grub(cfg, self.target, osfamily=self.distro_family,
variant=self.variant)
self.m_install_grub.assert_called_with(
- ['/dev/vdb'], self.target, uefi=True, grubcfg=cfg.get('grub')
+ ['/dev/vdb'], self.target, uefi=True,
+ grubcfg=config.fromdict(config.GrubConfig, cfg.get('grub'))
)
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
@@ -1101,7 +1102,7 @@ class TestUefiRemoveDuplicateEntries(CiTestCase):
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
def test_uefi_remove_duplicate_entries(self):
- grubcfg = {}
+ grubcfg = config.GrubConfig()
curthooks.uefi_remove_duplicate_entries(grubcfg, self.target)
self.assertEqual([
call(['efibootmgr', '--bootnum=0001', '--delete-bootnum'],
@@ -1112,7 +1113,7 @@ class TestUefiRemoveDuplicateEntries(CiTestCase):
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
def test_uefi_remove_duplicate_entries_no_bootcurrent(self):
- grubcfg = {}
+ grubcfg = config.GrubConfig()
efiout = copy_efi_state(self.efibootmgr_output)
efiout.current = ''
self.m_efibootmgr.return_value = efiout
@@ -1126,15 +1127,15 @@ class TestUefiRemoveDuplicateEntries(CiTestCase):
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
def test_uefi_remove_duplicate_entries_disabled(self):
- grubcfg = {
- 'remove_duplicate_entries': False,
- }
+ grubcfg = config.GrubConfig(
+ remove_duplicate_entries=False,
+ )
curthooks.uefi_remove_duplicate_entries(grubcfg, self.target)
self.assertEquals([], self.m_subp.call_args_list)
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
def test_uefi_remove_duplicate_entries_skip_bootcurrent(self):
- grubcfg = {}
+ grubcfg = config.GrubConfig()
efiout = copy_efi_state(self.efibootmgr_output)
efiout.current = '0003'
self.m_efibootmgr.return_value = efiout
@@ -1148,7 +1149,7 @@ class TestUefiRemoveDuplicateEntries(CiTestCase):
@patch.object(util.ChrootableTarget, "__enter__", new=lambda a: a)
def test_uefi_remove_duplicate_entries_no_change(self):
- grubcfg = {}
+ grubcfg = config.GrubConfig()
self.m_efibootmgr.return_value = util.EFIBootState(
order=[],
timeout='',
Follow ups
