Use snapd REST api to change snap config

lib/charms/operator_libs_linux/v2/snap.py

@@ -64,6 +64,7 @@
 import socket
 import subprocess
 import sys
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -83,7 +84,7 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 7
+LIBPATCH = 8
 
 
 # Regex to locate 7-bit C1 ANSI sequences
@@ -332,19 +333,17 @@ def get(self, key: Optional[str], *, typed: bool = False) -> Any:
 
         return self._snap("get", [key]).strip()
 
-    def set(self, config: Dict[str, Any], *, typed: bool = False) -> str:
+    def set(self, config: Dict[str, Any], *, typed: bool = False) -> None:
         """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.
         """
-        if typed:
-            kv = [f"{key}={json.dumps(val)}" for key, val in config.items()]
-            return self._snap("set", ["-t"] + kv)
-
-        return self._snap("set", [f"{key}={val}" for key, val in config.items()])
+        if not typed:
+            config = {k: str(v) for k, v in config.items()}
+        self._snap_client.put_snap_config(self._name, config)
 
     def unset(self, key) -> str:
         """Unset a snap configuration value.
@@ -770,7 +769,30 @@ def _request(
             headers["Content-Type"] = "application/json"
 
         response = self._request_raw(method, path, query, headers, data)
-        return json.loads(response.read().decode())["result"]
+        response = json.loads(response.read().decode())
+        if response["type"] == "async":
+            return self._wait(response["change"])
+        return response["result"]
+
+    def _wait(self, change_id: int) -> JSONType:
+        """Wait for an async change to complete.
+
+        The poll time is 100 milliseconds, the same as in snap clients.
+        """
+        while True:
+            response = self._request("GET", f"changes/{change_id}")
+            status = response["status"]
+            if status == "Done":
+                return response.get("data")
+            if status == "Doing":
+                time.sleep(0.1)
+                continue
+            if status == "Wait":
+                logger.warning("snap change %s succeeded with status 'Wait'", change_id)
+                return response.get("data")
+            raise SnapError(
+                f"snap change {response.get('kind')!r} id {change_id} failed with status {status}"
+            )
 
     def _request_raw(
         self,
@@ -818,6 +840,10 @@ def get_installed_snap_apps(self, name: str) -> List:
         """Query the snap server for apps belonging to a named, currently installed snap."""
         return self._request("GET", "apps", {"names": name, "select": "service"})
 
+    def put_snap_config(self, name: str, conf: Dict[str, Any]):
+        """Set the configuration details for an installed snap."""
+        return self._request("PUT", f"snaps/{name}/conf", body=conf)
+
 
 class SnapCache(Mapping):
     """An abstraction to represent installed/available packages.

tests/integration/test_snap.py

@@ -5,6 +5,7 @@
 
 import logging
 import re
+import time
 from datetime import datetime, timedelta
 from subprocess import CalledProcessError, check_output, run
 
@@ -65,7 +66,11 @@ def test_snap_refresh():
 def test_snap_set_and_get_with_typed():
     cache = snap.SnapCache()
     lxd = cache["lxd"]
-    lxd.ensure(snap.SnapState.Latest, channel="latest")
+    try:
+        lxd.ensure(snap.SnapState.Latest, channel="latest")
+    except snap.SnapError:
+        time.sleep(60)
+        lxd.ensure(snap.SnapState.Latest, channel="latest")
     configs = {
         "true": True,
         "false": False,
@@ -137,7 +142,11 @@ def test_snap_set_and_get_with_typed():
 def test_snap_set_and_get_untyped():
     cache = snap.SnapCache()
     lxd = cache["lxd"]
-    lxd.ensure(snap.SnapState.Latest, channel="latest")
+    try:
+        lxd.ensure(snap.SnapState.Latest, channel="latest")
+    except snap.SnapError:
+        time.sleep(60)
+        lxd.ensure(snap.SnapState.Latest, channel="latest")
 
     lxd.set({"foo": "true", "bar": True}, typed=False)
     assert lxd.get("foo", typed=False) == "true"

tests/unit/test_snap.py

@@ -902,28 +902,24 @@ def fake_snap(command: str, optargs: Optional[Iterable[str]] = None) -> str:
         with self.assertRaises(TypeError):
             foo.get(None)  # pyright: ignore[reportArgumentType]
 
-    @patch("charms.operator_libs_linux.v2.snap.subprocess.check_output")
-    def test_snap_set_typed(self, mock_subprocess):
+    @patch("charms.operator_libs_linux.v2.snap.SnapClient.put_snap_config")
+    def test_snap_set_typed(self, put_snap_config):
         foo = snap.Snap("foo", snap.SnapState.Present, "stable", "1", "classic")
 
         config = {"n": 42, "s": "string", "d": {"nested": True}}
 
         foo.set(config, typed=True)
-        mock_subprocess.assert_called_with(
-            ["snap", "set", "foo", "-t", "n=42", 's="string"', 'd={"nested": true}'],
-            universal_newlines=True,
-        )
+        put_snap_config.assert_called_with("foo", {"n": 42, "s": "string", "d": {"nested": True}})
 
-    @patch("charms.operator_libs_linux.v2.snap.subprocess.check_output")
-    def test_snap_set_untyped(self, mock_subprocess):
+    @patch("charms.operator_libs_linux.v2.snap.SnapClient.put_snap_config")
+    def test_snap_set_untyped(self, put_snap_config):
         foo = snap.Snap("foo", snap.SnapState.Present, "stable", "1", "classic")
 
         config = {"n": 42, "s": "string", "d": {"nested": True}}
 
         foo.set(config, typed=False)
-        mock_subprocess.assert_called_with(
-            ["snap", "set", "foo", "n=42", "s=string", "d={'nested': True}"],
-            universal_newlines=True,
+        put_snap_config.assert_called_with(
+            "foo", {"n": "42", "s": "string", "d": "{'nested': True}"}
         )
 
     @patch(