From b8efcff89f4bcd22d670e80a35585b11a40ca959 Mon Sep 17 00:00:00 2001 From: Robert Gildein Date: Mon, 24 Jul 2023 19:25:29 +0200 Subject: [PATCH] Add base logic for grub lib (#98) --- lib/charms/operator_libs_linux/v0/grub.py | 380 +++++++++++++++- tests/unit/test_grub.py | 501 ++++++++++++++++++++++ 2 files changed, 866 insertions(+), 15 deletions(-) create mode 100644 tests/unit/test_grub.py diff --git a/lib/charms/operator_libs_linux/v0/grub.py b/lib/charms/operator_libs_linux/v0/grub.py index d20ae6c..eff7eba 100644 --- a/lib/charms/operator_libs_linux/v0/grub.py +++ b/lib/charms/operator_libs_linux/v0/grub.py @@ -1,24 +1,67 @@ -"""TODO: Add a proper docstring here. +# Copyright 2023 Canonical Ltd. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. -This is a placeholder docstring for this charm library. Docstrings are -presented on Charmhub and updated whenever you push a new version of the -library. +"""Simple library for managing Linux kernel configuration via GRUB. -Complete documentation about creating and documenting libraries can be found -in the SDK docs at https://juju.is/docs/sdk/libraries. +This library is only used for setting additional parameters that will be stored in the +"/etc/default/grub.d/95-juju-charm.cfg" config file and not for editing other +configuration files. It's intended to be used in charms to help configure the machine. -See `charmcraft publish-lib` and `charmcraft fetch-lib` for details of how to -share and consume charm libraries. They serve to enhance collaboration -between charmers. Use a charmer's libraries for classes that handle -integration with their charm. +Configurations for individual charms will be stored in "/etc/default/grub.d/90-juju-", +but these configurations will only have informational value as all configurations will be merged +to "/etc/default/grub.d/95-juju-charm.cfg". -Bear in mind that new revisions of the different major API versions (v0, v1, -v2 etc) are maintained independently. You can continue to update v0 and v1 -after you have pushed v3. +Example of use: -Markdown is supported, following the CommonMark specification. +```python +class UbuntuCharm(CharmBase): + def __init__(self, *args): + ... + self.framework.observe(self.on.install, self._on_install) + self.framework.observe(self.on.update_status, self._on_update_status) + self.framework.observe(self.on.remove, self._on_remove) + self.grub = grub.GrubConfig(self.meta.name) + log.debug("found keys %s in GRUB config file", self.grub.keys()) + + def _on_install(self, _): + try: + self.grub.update( + {"GRUB_CMDLINE_LINUX_DEFAULT": "$GRUB_CMDLINE_LINUX_DEFAULT hugepagesz=1G"} + ) + except grub.ValidationError as error: + self.unit.status = BlockedStatus(f"[{error.key}] {error.message}") + + def _on_update_status(self, _): + if self.grub["GRUB_CMDLINE_LINUX_DEFAULT"] != "$GRUB_CMDLINE_LINUX_DEFAULT hugepagesz=1G": + self.unit.status = BlockedStatus("wrong GRUB configuration") + + def _on_remove(self, _): + self.grub.remove() +``` """ +import filecmp +import io +import logging +import os +import shlex +import subprocess +from pathlib import Path +from typing import Dict, Mapping, Optional, Set, Tuple + +logger = logging.getLogger(__name__) + # The unique Charmhub library identifier, never change it LIBID = "1f73a0e0c78349bc88850022e02b33c7" @@ -29,4 +72,311 @@ # to 0 if you are raising the major API version LIBPATCH = 1 -# TODO: add your code here! Happy coding! +GRUB_DIRECTORY = Path("/etc/default/grub.d/") +CHARM_CONFIG_PREFIX = "90-juju" +GRUB_CONFIG = GRUB_DIRECTORY / "95-juju-charm.cfg" +CONFIG_HEADER = f"""# This config file was produced by GRUB lib v{LIBAPI}.{LIBPATCH}. +# https://charmhub.io/operator-libs-linux/libraries/grub +""" +FILE_LINE_IN_DESCRIPTION = "# {path}" +CONFIG_DESCRIPTION = """ +# This file represents the output of the GRUB lib, which can combine multiple +# configurations into a single file like this. +# +# Original files: +{configs} +# +# If you change this file, run 'update-grub' afterwards to update +# /boot/grub/grub.cfg. +# For full documentation of the options in this file, see: +# info -f grub -n 'Simple configuration' +""" + + +class ValidationError(ValueError): + """Exception representing value validation error.""" + + def __init__(self, key: str, message: str) -> None: + super().__init__(message) + self.key = key + self.message = message + + def __str__(self) -> str: + """Return string representation of error.""" + return self.message + + +class IsContainerError(Exception): + """Exception if local machine is container.""" + + +class ApplyError(Exception): + """Exception if applying new config failed.""" + + +def _split_config_line(line: str) -> Tuple[str, str]: + """Split GRUB config line to obtain key and value.""" + key, raw_value = line.split("=", 1) + value, *not_expected_values = shlex.split(raw_value) + if not_expected_values: + logger.error("unexpected value %s for %s key", raw_value, key) + raise ValueError(f"unexpected value {raw_value} for {key} key") + + return key, value + + +def _parse_config(stream: io.TextIOWrapper) -> Dict[str, str]: + """Parse config file lines.""" + config = {} + for line in stream: + line = line.strip() + if not line or line.startswith("#"): + logger.debug("skipping line `%s`", line) + continue + + key, value = _split_config_line(line) + if key in config: + logger.warning("key %s is duplicated in config", key) + + config[key] = value + + return config + + +def _load_config(path: Path) -> Dict[str, str]: + """Load config file from /etc/default/grub.d/ directory.""" + if not path.exists(): + raise FileNotFoundError("GRUB config file %s was not found", path) + + with open(path, "r", encoding="UTF-8") as file: + config = _parse_config(file) + + logger.info("GRUB config file %s was loaded", path) + logger.debug("config file %s", config) + return config + + +def _save_config(path: Path, config: Dict[str, str], header: str = CONFIG_HEADER) -> None: + """Save GRUB config file.""" + if path.exists(): + logger.debug("GRUB config %s already exist and it will overwritten", path) + + context = [f"{key}={shlex.quote(value)}" for key, value in config.items()] + with open(path, "w", encoding="UTF-8") as file: + file.writelines([header, *context]) + + logger.info("GRUB config file %s was saved", path) + + +def check_update_grub() -> bool: + """Report whether an update to /boot/grub/grub.cfg is available.""" + main_grub_cfg = Path("/boot/grub/grub.cfg") + tmp_path = Path("/tmp/tmp_grub.cfg") + try: + subprocess.check_call( + ["/usr/sbin/grub-mkconfig", "-o", f"{tmp_path}"], stderr=subprocess.STDOUT + ) + except subprocess.CalledProcessError as error: + logger.exception(error) + raise + + return not filecmp.cmp(main_grub_cfg, tmp_path) + + +def is_container() -> bool: + """Report whether the local machine is a container.""" + try: + output = subprocess.check_output( + ["/usr/bin/systemd-detect-virt", "--container"], stderr=subprocess.STDOUT + ).decode() + logger.debug("detect virt type %s", output) + return True + except subprocess.CalledProcessError: + return False + + +class Config(Mapping[str, str]): + """Manages GRUB configuration. + + This object will load current configuration option for GRUB and provide option + to update it with simple validation, remove charm option and apply those changes. + """ + + _lazy_data: Optional[Dict[str, str]] = None + + def __init__(self, charm_name: str) -> None: + """Initialize the GRUB config.""" + self._charm_name = charm_name + + def __contains__(self, key: str) -> bool: + """Check if key is in config.""" + return key in self._data + + def __len__(self): + """Get size of config.""" + return len(self._data) + + def __iter__(self): + """Iterate over config.""" + return iter(self._data) + + def __getitem__(self, key: str) -> str: + """Get value for key form config.""" + return self._data[key] + + @property + def _data(self) -> Dict[str, str]: + """Data property.""" + if self._lazy_data is None: + try: + self._lazy_data = _load_config(GRUB_CONFIG) + except FileNotFoundError: + logger.debug("there is no GRUB config file %s yet", GRUB_CONFIG) + self._lazy_data = {} + + return self._lazy_data + + def _save_grub_configuration(self) -> None: + """Save current GRUB configuration.""" + logger.info("saving new GRUB config to %s", GRUB_CONFIG) + applied_configs = {self.path, *self.applied_configs} # using set to drop duplicity + registered_configs = os.linesep.join( + FILE_LINE_IN_DESCRIPTION.format(path=path) for path in applied_configs + ) + header = CONFIG_HEADER + CONFIG_DESCRIPTION.format(configs=registered_configs) + _save_config(GRUB_CONFIG, self._data, header) + + def _set_value(self, key: str, value: str, blocked_keys: Set[str]) -> bool: + """Set new value for key.""" + logger.debug("[%s] setting new value %s for key %s", self.charm_name, value, key) + current_value = self._data.get(key) + if current_value == value: + return False + + # validation + if key in self and current_value != value and key in blocked_keys: + logger.error( + "[%s] tries to overwrite key %s, which has value %s, with value %s", + self.charm_name, + key, + current_value, + value, + ) + raise ValidationError( + key, f"key {key} already exists and its value is {current_value}" + ) + + self._data[key] = value + return True + + def _update(self, config: Dict[str, str]) -> Set[str]: + """Update data in object.""" + logger.debug("[%s] updating current config", self.charm_name) + changed_keys = set() + blocked_keys = self.blocked_keys + for key, value in config.items(): + changed = self._set_value(key, value, blocked_keys) + if changed: + changed_keys.add(key) + + return changed_keys + + @property + def applied_configs(self) -> Dict[Path, Dict[str, str]]: + """Return list of charms configs which registered config in LIB_CONFIG_DIRECTORY.""" + configs = {} + for path in sorted(GRUB_DIRECTORY.glob(f"{CHARM_CONFIG_PREFIX}-*")): + configs[path] = _load_config(path) + logger.debug("load config file %s", path) + + return configs + + @property + def blocked_keys(self) -> Set[str]: + """Get set of configured keys by other charms.""" + return { + key + for path, config in self.applied_configs.items() + if path != self.path + for key in config + } + + @property + def charm_name(self) -> str: + """Get charm name or use value obtained from JUJU_UNIT_NAME env.""" + return self._charm_name + + @property + def path(self) -> Path: + """Return path for charm config.""" + return GRUB_DIRECTORY / f"{CHARM_CONFIG_PREFIX}-{self.charm_name}" + + def apply(self): + """Check if an update to /boot/grub/grub.cfg is available.""" + if not check_update_grub(): + logger.info("[%s] no available GRUB updates found", self.charm_name) + return + + try: + subprocess.check_call(["/usr/sbin/update-grub"], stderr=subprocess.STDOUT) + except subprocess.CalledProcessError as error: + logger.error( + "[%s] applying GRUB config failed with errors: %s", self.charm_name, error.stdout + ) + raise ApplyError("New config check failed.") from error + + def remove(self, apply: bool = True) -> Set[str]: + """Remove config for charm. + + This function will remove config file for charm and re-create the `95-juju-charm.cfg` + GRUB config file without changes made by this charm. + """ + if not self.path.exists(): + logger.debug("[%s] there is no charm config file %s", self.charm_name, self.path) + return set() + + self.path.unlink() + logger.info("[%s] charm config file %s was removed", self.charm_name, self.path) + config = {} + for _config in self.applied_configs.values(): + config.update(_config) + + changed_keys = set(self._data) - set(config.keys()) + self._lazy_data = config + self._save_grub_configuration() + if apply: + self.apply() + + return changed_keys + + def update(self, config: Dict[str, str], apply: bool = True) -> Set[str]: + """Update the Grub configuration.""" + if is_container(): + raise IsContainerError("Could not configure GRUB config on container.") + + snapshot = self._data.copy() + try: + changed_keys = self._update(config) + if changed_keys: + self._save_grub_configuration() + if apply: + self.apply() + except ValidationError as error: + logger.error("[%s] validation failed with message: %s", self.charm_name, error.message) + self._lazy_data = snapshot + logger.info("[%s] restored snapshot for Config object", self.charm_name) + raise + except ApplyError as error: + logger.error( + "[%s] applying new GRUB config failed with error: %s", self.charm_name, error + ) + self._lazy_data = snapshot + self._save_grub_configuration() # save snapshot copy of grub config + logger.info( + "[%s] restored snapshot for Config object and GRUB configuration", self.charm_name + ) + raise + + logger.debug("[%s] saving copy of charm config to %s", self.charm_name, GRUB_DIRECTORY) + _save_config(self.path, config) + return changed_keys diff --git a/tests/unit/test_grub.py b/tests/unit/test_grub.py new file mode 100644 index 0000000..d134cdb --- /dev/null +++ b/tests/unit/test_grub.py @@ -0,0 +1,501 @@ +# Copyright 2023 Canonical Ltd. +# See LICENSE file for licensing details. +import io +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import charms.operator_libs_linux.v0.grub as grub +import pytest + +GRUB_CONFIG_EXAMPLE_BODY = """ +GRUB_RECORDFAIL_TIMEOUT=0 +GRUB_TIMEOUT=0 +GRUB_TERMINAL=console +GRUB_CMDLINE_LINUX_DEFAULT="$GRUB_CMDLINE_LINUX_DEFAULT hugepagesz=1G" +""" +GRUB_CONFIG_EXAMPLE = f""" +{grub.CONFIG_HEADER} +{grub.CONFIG_DESCRIPTION.format(configs="# /tmp/test-path")} +# test commented line +{GRUB_CONFIG_EXAMPLE_BODY} +""" +EXP_GRUB_CONFIG = { + "GRUB_CMDLINE_LINUX_DEFAULT": "$GRUB_CMDLINE_LINUX_DEFAULT hugepagesz=1G", + "GRUB_RECORDFAIL_TIMEOUT": "0", + "GRUB_TERMINAL": "console", + "GRUB_TIMEOUT": "0", +} + + +def test_validation_error(): + """Test validation error and it's properties.""" + exp_key, exp_message = "test", "test message" + + error = grub.ValidationError(exp_key, exp_message) + with pytest.raises(ValueError): + raise error + + assert error.key == exp_key + assert error.message == exp_message + assert str(error) == exp_message + + +@pytest.mark.parametrize( + "output, exp_result", + [ + (mock.MagicMock(return_value=b"lxd"), True), + (mock.MagicMock(side_effect=subprocess.CalledProcessError(1, [])), False), + ], +) +def test_is_container(output, exp_result): + """Test helper function to validate if machine is container.""" + with mock.patch("subprocess.check_output", new=output) as mock_check_output: + assert grub.is_container() == exp_result + mock_check_output.assert_called_once_with( + ["/usr/bin/systemd-detect-virt", "--container"], stderr=subprocess.STDOUT + ) + + +class BaseTestGrubLib(unittest.TestCase): + def setUp(self) -> None: + tmp_dir = tempfile.TemporaryDirectory() + self.tmp_dir = Path(tmp_dir.name) + self.addCleanup(tmp_dir.cleanup) + + # configured paths + grub.GRUB_DIRECTORY = self.tmp_dir + + # change logger + mocked_logger = mock.patch.object(grub, "logger") + self.logger = mocked_logger.start() + self.addCleanup(mocked_logger.stop) + + +class TestGrubUtils(BaseTestGrubLib): + def test_split_config_line(self): + """Test splitting single line.""" + key, value = grub._split_config_line('test="1234"') + assert key == "test" + assert value == "1234" + + def test_split_config_line_failed(self): + """Test splitting single line.""" + with self.assertRaises(ValueError): + grub._split_config_line('test="1234" "5678"') + + def test_parse_config(self): + """Test parsing example GRUB config with skipping duplicated key.""" + stream = io.StringIO(GRUB_CONFIG_EXAMPLE) + result = grub._parse_config(stream) + + self.assertEqual(result, EXP_GRUB_CONFIG) + + def test_parse_config_with_duplicity(self): + """Test parsing example GRUB config with skipping duplicated key.""" + raw_config = ( + GRUB_CONFIG_EXAMPLE + 'GRUB_CMDLINE_LINUX_DEFAULT="$GRUB_CMDLINE_LINUX_DEFAULT pti=on"' + ) + stream = io.StringIO(raw_config) + result = grub._parse_config(stream) + + self.logger.warning.assert_called_once_with( + "key %s is duplicated in config", "GRUB_CMDLINE_LINUX_DEFAULT" + ) + self.assertEqual( + result["GRUB_CMDLINE_LINUX_DEFAULT"], + "$GRUB_CMDLINE_LINUX_DEFAULT pti=on", + ) + + def test_load_config_not_exists(self): + """Test load config from file which does not exist.""" + path = self.tmp_dir / "test_load_config" + + with pytest.raises(FileNotFoundError): + grub._load_config(path) + + @mock.patch.object(grub, "_parse_config") + def test_load_config(self, mock_parse_config): + """Test load config from file.""" + exp_config = {"test": "valid"} + mock_parse_config.return_value = exp_config + path = self.tmp_dir / "test_load_config" + path.touch() # create file + + with mock.patch.object(grub, "open", mock.mock_open()) as mock_open: + grub._load_config(path) + + mock_open.assert_called_once_with(path, "r", encoding="UTF-8") + mock_parse_config.assert_called_once_with(mock_open.return_value) + + def test_save_config(self): + """Test to save GRUB config file.""" + path = self.tmp_dir / "test-config" + + with mock.patch.object(grub, "open", mock.mock_open()) as mock_open: + grub._save_config(path, {"test": '"1234"'}) + + mock_open.assert_called_once_with(path, "w", encoding="UTF-8") + mock_open.return_value.writelines.assert_called_once_with([mock.ANY, "test='\"1234\"'"]) + + def test_save_config_overwrite(self): + """Test overwriting if GRUB config already exist.""" + path = self.tmp_dir / "test-config" + path.touch() + + with mock.patch.object(grub, "open", mock.mock_open()): + grub._save_config(path, {"test": '"1234"'}) + + self.logger.debug.assert_called_once_with( + "GRUB config %s already exist and it will overwritten", path + ) + + @mock.patch("subprocess.check_call") + @mock.patch("filecmp.cmp") + def test_check_update_grub(self, mock_filecmp, mock_check_call): + """Test check update function.""" + grub.check_update_grub() + mock_check_call.assert_called_once_with( + ["/usr/sbin/grub-mkconfig", "-o", "/tmp/tmp_grub.cfg"], stderr=subprocess.STDOUT + ) + mock_filecmp.assert_called_once_with( + Path("/boot/grub/grub.cfg"), Path("/tmp/tmp_grub.cfg") + ) + + @mock.patch("subprocess.check_call") + @mock.patch("filecmp.cmp") + def test_check_update_grub_failure(self, mock_filecmp, mock_check_call): + """Test check update function.""" + mock_check_call.side_effect = subprocess.CalledProcessError(1, []) + + with self.assertRaises(subprocess.CalledProcessError): + grub.check_update_grub() + + mock_filecmp.assert_not_called() + + +class TestGrubConfig(BaseTestGrubLib): + def setUp(self) -> None: + super().setUp() + # load config + mocker_load_config = mock.patch.object(grub, "_load_config") + self.load_config = mocker_load_config.start() + self.addCleanup(mocker_load_config.stop) + # save config + mocker_save_config = mock.patch.object(grub, "_save_config") + self.save_config = mocker_save_config.start() + self.addCleanup(mocker_save_config.stop) + # is_container + mocker_is_container = mock.patch.object(grub, "is_container") + self.is_container = mocker_is_container.start() + self.is_container.return_value = False + self.addCleanup(mocker_is_container.stop) + # check_update_grub + mocker_check_update_grub = mock.patch.object(grub, "check_update_grub") + self.check_update_grub = mocker_check_update_grub.start() + self.addCleanup(mocker_check_update_grub.stop) + + self.name = "charm-a" + self.path = self.tmp_dir / f"90-juju-{self.name}" + with open(self.path, "w") as file: + # create example of charm-a config + file.write(GRUB_CONFIG_EXAMPLE_BODY) + self.config = grub.Config(self.name) + self.config._lazy_data = EXP_GRUB_CONFIG.copy() + + def test_lazy_data_not_loaded(self): + """Test data not loaded.""" + self.load_config.assert_not_called() + + def test__contains__(self): + """Test config __contains__ function.""" + self.assertIn("GRUB_TIMEOUT", self.config) + + def test__len__(self): + """Test config __len__ function.""" + self.assertEqual(len(self.config), 4) + + def test__iter__(self): + """Test config __iter__ function.""" + self.assertListEqual(list(self.config), list(EXP_GRUB_CONFIG.keys())) + + def test__getitem__(self): + """Test config __getitem__ function.""" + for key, value in EXP_GRUB_CONFIG.items(): + self.assertEqual(self.config[key], value) + + def test_data(self): + """Test data not loaded.""" + self.load_config.return_value = EXP_GRUB_CONFIG + self.config._lazy_data = None + assert "test" not in self.config # this will call config._data once + + self.load_config.assert_called_once_with(grub.GRUB_CONFIG) + self.assertDictEqual(self.config._data, EXP_GRUB_CONFIG) + + def test_data_no_file(self): + """Test data not loaded.""" + self.config._lazy_data = None + self.load_config.side_effect = FileNotFoundError() + assert "test" not in self.config # this will call config._data once + + self.load_config.assert_called_once_with(grub.GRUB_CONFIG) + self.assertDictEqual(self.config._data, {}) + + def test_save_grub_config(self): + """Test save GRUB config.""" + exp_configs = [self.tmp_dir / "90-juju-charm-b", self.tmp_dir / "90-juju-charm-c"] + [path.touch() for path in exp_configs] # create files + exp_data = {"GRUB_TIMEOUT": "1"} + + self.config._lazy_data = exp_data + self.config._save_grub_configuration() + + self.save_config.assert_called_once_with(grub.GRUB_CONFIG, exp_data, mock.ANY) + _, _, header = self.save_config.call_args[0] + self.assertIn(str(self.config.path), header) + for exp_config_path in exp_configs: + self.assertIn(str(exp_config_path), header) + + def test_set_new_value(self): + """Test set new value in config.""" + changed = self.config._set_value("GRUB_NEW_KEY", "1", set()) + self.assertTrue(changed) + + def test_set_existing_value_without_change(self): + """Test set existing key, but with same value.""" + changed = self.config._set_value("GRUB_TIMEOUT", "0", set()) + self.assertFalse(changed) + + def test_set_existing_value_with_change(self): + """Test set existing key with new value.""" + changed = self.config._set_value("GRUB_TIMEOUT", "1", set()) + self.assertTrue(changed) + + def test_set_blocked_key(self): + """Test set new value in config.""" + with pytest.raises(grub.ValidationError): + self.config._set_value("GRUB_TIMEOUT", "1", {"GRUB_TIMEOUT"}) + + def test_update_data(self): + """Test update GrubConfig _data.""" + data = {"GRUB_TIMEOUT": "1", "GRUB_RECORDFAIL_TIMEOUT": "0"} + new_data = {"GRUB_NEW_KEY": "test", "GRUB_RECORDFAIL_TIMEOUT": "0"} + + self.config._lazy_data = data + changed_keys = self.config._update(new_data) + self.assertEqual({"GRUB_NEW_KEY"}, changed_keys) + + def test_applied_configs(self): + """Test applied_configs property.""" + exp_configs = { + self.path: self.load_config.return_value, + self.tmp_dir / "90-juju-charm-b": self.load_config.return_value, + self.tmp_dir / "90-juju-charm-c": self.load_config.return_value, + } + [path.touch() for path in exp_configs] # create files + + self.assertEqual(self.config.applied_configs, exp_configs) + + def test_blocked_keys(self): + """Test blocked_keys property.""" + exp_configs = { + self.path: {"A": "1", "B": "2"}, + self.tmp_dir / "90-juju-charm-b": {"B": "3", "C": "2"}, + self.tmp_dir / "90-juju-charm-c": {"D": "4"}, + } + + with mock.patch.object( + grub.Config, "applied_configs", new=mock.PropertyMock(return_value=exp_configs) + ): + blocked_keys = self.config.blocked_keys + + self.assertSetEqual(blocked_keys, {"B", "C", "D"}) + + def test_path(self): + """Test path property.""" + self.assertEqual(self.config.path, self.tmp_dir / f"90-juju-{self.name}") + + @mock.patch("subprocess.check_call") + def test_apply_without_changes(self, mock_call): + """Test applying GRUB config without any changes.""" + self.check_update_grub.return_value = False + self.config.apply() + + self.check_update_grub.assert_called_once() + mock_call.assert_not_called() + + @mock.patch("subprocess.check_call") + def test_apply_with_new_changes(self, mock_call): + """Test applying GRUB config.""" + self.check_update_grub.return_value = True + self.config.apply() + + self.check_update_grub.assert_called_once() + mock_call.assert_called_once_with(["/usr/sbin/update-grub"], stderr=subprocess.STDOUT) + + @mock.patch("subprocess.check_call") + def test_apply_failure(self, mock_call): + """Test applying GRUB config failure.""" + mock_call.side_effect = subprocess.CalledProcessError(1, []) + with self.assertRaises(grub.ApplyError): + self.config.apply() + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_remove_no_config(self, mock_save, mock_apply): + """Test removing when there is no charm config.""" + self.config.path.unlink() # remove charm config file + changed_keys = self.config.remove() + + self.assertSetEqual(changed_keys, set()) + mock_save.assert_not_called() + mock_apply.assert_not_called() + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_remove_no_apply(self, mock_save, mock_apply): + """Test removing without applying.""" + changed_keys = self.config.remove(apply=False) + + self.assertSetEqual(changed_keys, set(EXP_GRUB_CONFIG.keys())) + mock_save.assert_called_once() + mock_apply.assert_not_called() + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_remove(self, mock_save, mock_apply): + """Test removing config for current charm.""" + exp_changed_keys = {"GRUB_RECORDFAIL_TIMEOUT"} + exp_configs = { + self.name: EXP_GRUB_CONFIG, + "charm-b": {"GRUB_TIMEOUT": "0"}, + "charm-c": { + "GRUB_TERMINAL": "console", + "GRUB_CMDLINE_LINUX_DEFAULT": '"$GRUB_CMDLINE_LINUX_DEFAULT hugepagesz=1G"', + }, + } + self.load_config.side_effect = [exp_configs["charm-b"], exp_configs["charm-c"]] + (self.tmp_dir / "90-juju-charm-b").touch() + (self.tmp_dir / "90-juju-charm-c").touch() + + changed_keys = self.config.remove() + + self.assertFalse((self.tmp_dir / self.name).exists()) + self.load_config.assert_has_calls( + [ + mock.call(self.tmp_dir / f"90-juju-{charm}") + for charm in exp_configs + if charm != self.name + ] + ) + self.assertSetEqual(changed_keys, exp_changed_keys) + self.assertDictEqual( + self.config._data, + { + "GRUB_TIMEOUT": "0", + "GRUB_TERMINAL": "console", + "GRUB_CMDLINE_LINUX_DEFAULT": '"$GRUB_CMDLINE_LINUX_DEFAULT hugepagesz=1G"', + }, + ) + mock_save.assert_called_once() + mock_apply.assert_called_once() + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_update_on_container(self, mock_save, mock_apply): + """Test update current GRUB config on container.""" + self.is_container.return_value = True + + with self.assertRaises(grub.IsContainerError): + self.config.update({"GRUB_TIMEOUT": "0"}) + + mock_save.assert_not_called() + mock_apply.assert_not_called() + self.assertDictEqual(self.config._data, EXP_GRUB_CONFIG, "config was changed") + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + @mock.patch.object( + grub.Config, "_set_value", side_effect=grub.ValidationError("test", "test_message") + ) + def test_update_validation_failure(self, _, mock_save, mock_apply): + """Test update current GRUB config with validation failure.""" + with self.assertRaises(grub.ValidationError): + # trying to set already existing key with different value -> ValidationError + self.config.update({"GRUB_TIMEOUT": "1"}) + + mock_save.assert_not_called() + mock_apply.assert_not_called() + self.assertDictEqual(self.config._data, EXP_GRUB_CONFIG, "config was changed") + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_update_apply_failure(self, mock_save, mock_apply): + """Test update current GRUB config with applied failure.""" + mock_apply.side_effect = grub.ApplyError("failed to apply") + + with self.assertRaises(grub.ApplyError): + self.config.update({"GRUB_NEW_KEY": "1"}) + + mock_save.assert_has_calls( + [mock.call()] * 2, + "it should be called once before apply and one after snapshot restore", + ) + mock_apply.assert_called_once() + self.assertDictEqual(self.config._data, EXP_GRUB_CONFIG, "snapshot was not restored") + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_update_without_changes(self, mock_save, mock_apply): + """Test update current GRUB config without any changes.""" + changed_keys = self.config.update({"GRUB_TIMEOUT": "0"}) + + self.assertSetEqual(changed_keys, set()) + mock_save.assert_not_called() + mock_apply.assert_not_called() + self.save_config.assert_called_once_with(self.path, mock.ANY) + self.assertDictEqual(self.config._data, EXP_GRUB_CONFIG) + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_update(self, mock_save, mock_apply): + """Test update current GRUB config without applying it.""" + new_config = {"GRUB_NEW_KEY": "1"} + exp_config = {**EXP_GRUB_CONFIG, **new_config} + + self.config.update(new_config) + + mock_save.assert_called_once() + mock_apply.assert_called_once() + self.save_config.assert_called_once_with(self.path, new_config) + self.assertDictEqual(self.config._data, exp_config) + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_update_same_charm(self, *_): + """Test update current GRUB config twice with different values. + + This test is simulating the scenario, when same charm want to change it's own + values. + """ + first_config = {"GRUB_NEW_KEY": "0"} + new_config = {"GRUB_NEW_KEY": "1"} + exp_config = {**EXP_GRUB_CONFIG, **new_config} + + self.config.update(first_config) + self.config.update(new_config) + + self.assertDictEqual(self.config._data, exp_config) + + @mock.patch.object(grub.Config, "apply") + @mock.patch.object(grub.Config, "_save_grub_configuration") + def test_update_without_apply(self, mock_save, mock_apply): + """Test update current GRUB config without applying it.""" + self.config.update({"GRUB_NEW_KEY": "1"}, apply=False) + + mock_save.assert_called_once() + mock_apply.assert_not_called() + self.save_config.assert_called_once_with(self.path, mock.ANY)