Fix the typing issue in snap configs.

Currently, setting boolean config with `Snap.set` will result in setting
the snap config to a string of `true` or `false`, and the openstack
hypervisor snap will read those configs as string as well. This makes
the condition checking in openstack-hypervisor snap behaves incorrectly
[1]. This PR is to update the snap library will proper typing support.

- Update snap library
- Fix the data types used in `Snap.set()`

[1] https://github.com/openstack-snaps/snap-openstack-hypervisor/blob/main/openstack_hypervisor/hooks.py#L740

Closes-Bug: #2033272
Change-Id: I7bec4599b23500aaad9e008fce648793c104b642

charms/openstack-hypervisor/lib/charms/operator_libs_linux/v2/snap.py

@@ -83,7 +83,7 @@ LIBAPI = 2
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 0
+LIBPATCH = 2
 
 
 # Regex to locate 7-bit C1 ANSI sequences
@@ -273,13 +273,13 @@ class Snap(object):
           SnapError if there is a problem encountered
         """
         optargs = optargs or []
-        _cmd = ["snap", command, self._name, *optargs]
+        args = ["snap", command, self._name, *optargs]
         try:
-            return subprocess.check_output(_cmd, universal_newlines=True)
+            return subprocess.check_output(args, universal_newlines=True)
         except CalledProcessError as e:
             raise SnapError(
                 "Snap: {!r}; command {!r} failed with output = {!r}".format(
-                    self._name, _cmd, e.output
+                    self._name, args, e.output
                 )
             )
 
@@ -303,30 +303,45 @@ class Snap(object):
         else:
             services = [self._name]
 
-        _cmd = ["snap", *command, *services]
+        args = ["snap", *command, *services]
 
         try:
-            return subprocess.run(_cmd, universal_newlines=True, check=True, capture_output=True)
+            return subprocess.run(args, universal_newlines=True, check=True, capture_output=True)
         except CalledProcessError as e:
-            raise SnapError("Could not {} for snap [{}]: {}".format(_cmd, self._name, e.stderr))
+            raise SnapError("Could not {} for snap [{}]: {}".format(args, self._name, e.stderr))
 
-    def get(self, key) -> str:
+    def get(self, key: Optional[str], *, typed: bool = False) -> Any:
-        """Fetch a snap configuration value.
+        """Fetch snap configuration values.
 
         Args:
-            key: the key to retrieve
+            key: the key to retrieve. Default to retrieve all values for typed=True.
+            typed: set to True to retrieve typed values (set with typed=True).
+                Default is to return a string.
         """
+        if typed:
+            config = json.loads(self._snap("get", ["-d", key]))
+            if key:
+                return config.get(key)
+            return config
+
+        if not key:
+            raise TypeError("Key must be provided when typed=False")
+
         return self._snap("get", [key]).strip()
 
-    def set(self, config: Dict) -> str:
+    def set(self, config: Dict[str, Any], *, typed: bool = False) -> str:
         """Set a snap configuration value.
 
         Args:
            config: a dictionary containing keys and values specifying the config to set.
+           typed: set to True to convert all values in the config into typed values while
+                configuring the snap (set with typed=True). Default is not to convert.
         """
-        args = ['{}="{}"'.format(key, val) for key, val in config.items()]
+        if typed:
+            kv = [f"{key}={json.dumps(val)}" for key, val in config.items()]
+            return self._snap("set", ["-t"] + kv)
 
-        return self._snap("set", [*args])
+        return self._snap("set", [f"{key}={val}" for key, val in config.items()])
 
     def unset(self, key) -> str:
         """Unset a snap configuration value.
@@ -387,11 +402,11 @@ class Snap(object):
         elif slot:
             command = command + [slot]
 
-        _cmd = ["snap", *command]
+        args = ["snap", *command]
         try:
-            subprocess.run(_cmd, universal_newlines=True, check=True, capture_output=True)
+            subprocess.run(args, universal_newlines=True, check=True, capture_output=True)
         except CalledProcessError as e:
-            raise SnapError("Could not {} for snap [{}]: {}".format(_cmd, self._name, e.stderr))
+            raise SnapError("Could not {} for snap [{}]: {}".format(args, self._name, e.stderr))
 
     def hold(self, duration: Optional[timedelta] = None) -> None:
         """Add a refresh hold to a snap.
@@ -409,6 +424,25 @@ class Snap(object):
         """Remove the refresh hold of a snap."""
         self._snap("refresh", ["--unhold"])
 
+    def alias(self, application: str, alias: Optional[str] = None) -> None:
+        """Create an alias for a given application.
+
+        Args:
+            application: application to get an alias.
+            alias: (optional) name of the alias; if not provided, the application name is used.
+        """
+        if alias is None:
+            alias = application
+        args = ["snap", "alias", f"{self.name}.{application}", alias]
+        try:
+            subprocess.check_output(args, universal_newlines=True)
+        except CalledProcessError as e:
+            raise SnapError(
+                "Snap: {!r}; command {!r} failed with output = {!r}".format(
+                    self._name, args, e.output
+                )
+            )
+
     def restart(
         self, services: Optional[List[str]] = None, reload: Optional[bool] = False
     ) -> None:
@@ -879,11 +913,11 @@ def add(
     if not channel and not revision:
         channel = "latest"
 
-    snap_names = [snap_names] if type(snap_names) is str else snap_names
+    snap_names = [snap_names] if isinstance(snap_names, str) else snap_names
     if not snap_names:
         raise TypeError("Expected at least one snap to add, received zero!")
 
-    if type(state) is str:
+    if isinstance(state, str):
         state = SnapState(state)
 
     return _wrap_snap_operations(snap_names, state, channel, classic, cohort, revision)
@@ -899,7 +933,7 @@ def remove(snap_names: Union[str, List[str]]) -> Union[Snap, List[Snap]]:
     Raises:
         SnapError if some snaps failed to install.
     """
-    snap_names = [snap_names] if type(snap_names) is str else snap_names
+    snap_names = [snap_names] if isinstance(snap_names, str) else snap_names
     if not snap_names:
         raise TypeError("Expected at least one snap to add, received zero!")
 
@@ -992,17 +1026,17 @@ def install_local(
     Raises:
         SnapError if there is a problem encountered
     """
-    _cmd = [
+    args = [
         "snap",
         "install",
         filename,
     ]
     if classic:
-        _cmd.append("--classic")
+        args.append("--classic")
     if dangerous:
-        _cmd.append("--dangerous")
+        args.append("--dangerous")
     try:
-        result = subprocess.check_output(_cmd, universal_newlines=True).splitlines()[-1]
+        result = subprocess.check_output(args, universal_newlines=True).splitlines()[-1]
         snap_name, _ = result.split(" ", 1)
         snap_name = ansi_filter.sub("", snap_name)
 
@@ -1026,9 +1060,9 @@ def _system_set(config_item: str, value: str) -> None:
         config_item: name of snap system setting. E.g. 'refresh.hold'
         value: value to assign
     """
-    _cmd = ["snap", "set", "system", "{}={}".format(config_item, value)]
+    args = ["snap", "set", "system", "{}={}".format(config_item, value)]
     try:
-        subprocess.check_call(_cmd, universal_newlines=True)
+        subprocess.check_call(args, universal_newlines=True)
     except CalledProcessError:
         raise SnapError("Failed setting system config '{}' to '{}'".format(config_item, value))
 

charms/openstack-hypervisor/src/charm.py

@@ -21,7 +21,6 @@ This charm provide hypervisor services as part of an OpenStack deployment
 """
 
 import base64
-import json
 import logging
 import os
 import secrets
@@ -142,7 +141,7 @@ class HypervisorOperatorCharm(sunbeam_charm.OSBaseOperatorCharm):
                 new_settings[k] = snap_data[k]
         if new_settings:
             logger.debug(f"Applying new snap settings {new_settings}")
-            hypervisor.set(new_settings)
+            hypervisor.set(new_settings, typed=True)
         else:
             logger.debug("Snap settings do not need updating")
 
@@ -185,10 +184,10 @@ class HypervisorOperatorCharm(sunbeam_charm.OSBaseOperatorCharm):
                 "identity.user-domain-id": contexts.identity_credentials.user_domain_id,
                 "identity.user-domain-name": contexts.identity_credentials.user_domain_name,
                 "identity.username": contexts.identity_credentials.username,
-                "logging.debug": json.dumps(config("debug")),
+                "logging.debug": config("debug"),
                 "network.dns-domain": config("dns-domain"),
                 "network.dns-servers": config("dns-servers"),
-                "network.enable-gateway": json.dumps(config("enable-gateway")),
+                "network.enable-gateway": config("enable-gateway"),
                 "network.external-bridge": config("external-bridge"),
                 "network.external-bridge-address": config("external-bridge-address")
                 or "10.20.20.1/24",

charms/openstack-hypervisor/tests/unit/test_charm.py

@@ -112,10 +112,10 @@ class TestCharm(test_utils.CharmTestCase):
             "identity.user-domain-id": "udomain-id",
             "identity.user-domain-name": "udomain-name",
             "identity.username": "username",
-            "logging.debug": "false",
+            "logging.debug": False,
             "network.dns-domain": "openstack.local",
             "network.dns-servers": "8.8.8.8",
-            "network.enable-gateway": "false",
+            "network.enable-gateway": False,
             "network.external-bridge": "br-ex",
             "network.external-bridge-address": "10.20.20.1/24",
             "network.ip-address": "10.0.0.10",
@@ -128,4 +128,4 @@ class TestCharm(test_utils.CharmTestCase):
             "node.ip-address": "10.0.0.10",
             "rabbitmq.url": "rabbit://hypervisor:rabbit.pass@10.0.0.13:5672/openstack",
         }
-        hypervisor_snap_mock.set.assert_any_call(expect_settings)
+        hypervisor_snap_mock.set.assert_any_call(expect_settings, typed=True)