diff --git a/src/hw_tools.py b/src/hw_tools.py index 4febb159..ef1cf544 100644 --- a/src/hw_tools.py +++ b/src/hw_tools.py @@ -18,6 +18,7 @@ import requests import urllib3 from charms.operator_libs_linux.v0 import apt +from charms.operator_libs_linux.v2 import snap from ops.model import ModelError, Resources import apt_helpers @@ -175,6 +176,41 @@ def remove(self) -> None: # the remove option. +class SnapStrategy(StrategyABC): + """Snap strategy class.""" + + def __init__(self, tool: HWTool, channel: str): + """Snap strategy constructor.""" + self._name = tool + self.snap_name = tool.value + self.channel = channel + self.snap_client = snap.SnapCache()[tool.value] + + def install(self) -> None: + """Install the snap from a channel.""" + try: + snap.add(self.snap_name, channel=self.channel) + # using the snap.SnapError will result into: + # TypeError: catching classes that do not inherit from BaseException is not allowed + except Exception as err: # pylint: disable=broad-except + logger.error( + "Failed to install %s from channel: %s: %s", self.snap_name, self.channel, err + ) + raise err + + logger.info("Installed %s from channel: %s", self.snap_name, self.channel) + # enable services because some might be disabled by default + self.snap_client.start(list(self.snap_client.services.keys()), enable=True) + + def remove(self) -> None: + """Remove the snap.""" + snap.remove([self.snap_name]) + + def check(self) -> bool: + """Check if all services are active.""" + return all(service.get("active", False) for service in self.snap_client.services.values()) + + class TPRStrategyABC(StrategyABC, metaclass=ABCMeta): """Third party resource strategy class.""" @@ -689,13 +725,12 @@ def install(self, resources: Resources, hw_available: Set[HWTool]) -> Tuple[bool if strategy.name not in hw_available: continue try: - # TPRStrategy if isinstance(strategy, TPRStrategyABC): path = fetch_tools.get(strategy.name) # pylint: disable=W0212 if path: strategy.install(path) - # APTStrategy - elif isinstance(strategy, APTStrategyABC): + + elif isinstance(strategy, (APTStrategyABC, SnapStrategy)): strategy.install() # pylint: disable=E1120 logger.info("Strategy %s install success", strategy) except ( @@ -703,6 +738,7 @@ def install(self, resources: Resources, hw_available: Set[HWTool]) -> Tuple[bool OSError, apt.PackageError, ResourceChecksumError, + snap.SnapError, ) as e: logger.warning("Strategy %s install fail: %s", strategy, e) fail_strategies.append(strategy.name) @@ -717,7 +753,7 @@ def remove(self, resources: Resources, hw_available: Set[HWTool]) -> None: for strategy in self.strategies: if strategy.name not in hw_available: continue - if isinstance(strategy, (TPRStrategyABC, APTStrategyABC)): + if isinstance(strategy, (TPRStrategyABC, APTStrategyABC, SnapStrategy)): strategy.remove() logger.info("Strategy %s remove success", strategy) diff --git a/tests/unit/test_hw_tools.py b/tests/unit/test_hw_tools.py index 5784f687..8b946b13 100644 --- a/tests/unit/test_hw_tools.py +++ b/tests/unit/test_hw_tools.py @@ -36,6 +36,7 @@ SAS3IRCUStrategy, SmartCtlExporterStrategy, SmartCtlStrategy, + SnapStrategy, SSACLIStrategy, StorCLIStrategy, StrategyABC, @@ -57,6 +58,7 @@ symlink, ) from keys import HP_KEYS +from lib.charms.operator_libs_linux.v2 import snap def get_mock_path(size: int): @@ -1147,3 +1149,133 @@ def mock_get_response_ipmi(ipmi_call): output = bmc_hw_verifier() mock_apt_helpers.add_pkg_with_candidate_version.assert_called_with("freeipmi-tools") self.assertCountEqual(output, [HWTool.IPMI_SENSOR, HWTool.IPMI_SEL]) + + +@mock.patch("hw_tools.snap") +def test_snap_strategy_name(_): + hwtool = mock.MagicMock() + hwtool.value = "my-snap" + + strategy = SnapStrategy(hwtool, "my-channel") + assert strategy.name == hwtool + + +@mock.patch("hw_tools.snap.SnapCache") +@mock.patch("hw_tools.snap") +def test_snap_strategy_install(mock_snap, mock_snap_cache): + hwtool = mock.MagicMock() + hwtool.value = "my-snap" + channel = "my-channel" + + mock_snap_client = mock.MagicMock() + mock_snap_client.services = {"service1": {}, "service2": {}} + mock_snap_cache.return_value = {"my-snap": mock_snap_client} + + strategy = SnapStrategy(hwtool, channel) + strategy.install() + mock_snap.add.assert_called_with(strategy.snap_name, channel=channel) + mock_snap_client.start.assert_called_once_with(["service1", "service2"], enable=True) + + +@mock.patch("hw_tools.snap.SnapCache") +@mock.patch("hw_tools.snap") +def test_snap_strategy_install_fail(mock_snap, mock_snap_cache): + hwtool = mock.MagicMock() + hwtool.value = "my-snap" + channel = "my-channel" + + mock_snap_client = mock.MagicMock() + mock_snap_cache.return_value = {"my-snap": mock_snap_client} + + strategy = SnapStrategy(hwtool, channel) + mock_snap.add.side_effect = snap.SnapError + with pytest.raises(snap.SnapError): + strategy.install() + mock_snap_client.start.assert_not_called() + + +@mock.patch("hw_tools.snap") +def test_snap_strategy_remove(mock_snap): + hwtool = mock.MagicMock() + hwtool.value = "my-snap" + strategy = SnapStrategy(hwtool, "my-channel") + strategy.remove() + mock_snap.remove.assert_called_with([strategy.snap_name]) + + +@pytest.mark.parametrize( + "services, expected", + [ + # all services active + ( + { + "service_1": { + "daemon": "simple", + "daemon_scope": "system", + "enabled": True, + "active": True, + "activators": [], + }, + "service_2": { + "daemon": "simple", + "daemon_scope": "system", + "enabled": True, + "active": True, + "activators": [], + }, + }, + True, + ), + # at least one services down + ( + { + "service_1": { + "daemon": "simple", + "daemon_scope": "system", + "enabled": True, + "active": False, + "activators": [], + }, + "service_2": { + "daemon": "simple", + "daemon_scope": "system", + "enabled": True, + "active": True, + "activators": [], + }, + }, + False, + ), + # all services down + ( + { + "service_1": { + "daemon": "simple", + "daemon_scope": "system", + "enabled": True, + "active": False, + "activators": [], + }, + "service_2": { + "daemon": "simple", + "daemon_scope": "system", + "enabled": True, + "active": False, + "activators": [], + }, + }, + False, + ), + # snap without service + ({}, True), + ], +) +@mock.patch("hw_tools.snap.SnapCache") +def test_snap_strategy_check(mock_snap_cache, services, expected): + mock_snap_info = mock.MagicMock() + mock_snap_info.services = services + mock_snap_cache.return_value.__getitem__.return_value = mock_snap_info + hwtool = mock.MagicMock() + hwtool.value = "my-snap" + strategy = SnapStrategy(hwtool, "my-channel") + assert strategy.check() is expected