Add Snap Strategy

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,20 +725,20 @@ 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 (
                 ResourceFileSizeZeroError,
                 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)
 

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