fix: check if lxd snap is installed (#585)

Check if the lxd snap is installed by querying the snapd socket
instead of looking for an executable in the path, as lxd can use
auto-installer stubs.

Co-authored-by: Callahan Kovacs <callahan.kovacs@canonical.com>
Signed-off-by: Claudio Matsuoka <claudio.matsuoka@canonical.com>

craft_providers/lxd/installer.py

@@ -19,10 +19,14 @@
 
 import logging
 import os
+import pathlib
 import shutil
 import subprocess
 import sys
 
+import requests
+import requests_unixsocket  # type: ignore
+
 from craft_providers.errors import details_from_called_process_error
 
 from . import errors
@@ -52,6 +56,7 @@ def install(sudo: bool = True) -> str:
 
     cmd += ["snap", "install", "lxd"]
 
+    logger.debug("installing LXD")
     try:
         subprocess.run(cmd, check=True)
     except subprocess.CalledProcessError as error:
@@ -62,6 +67,8 @@ def install(sudo: bool = True) -> str:
 
     lxd = LXD()
     lxd.wait_ready(sudo=sudo)
+
+    logger.debug("initialising LXD")
     lxd.init(auto=True, sudo=sudo)
 
     if not is_user_permitted():
@@ -96,11 +103,45 @@ def is_initialized(*, remote: str, lxc: LXC) -> bool:
 
 
 def is_installed() -> bool:
-    """Check if LXD is installed (and found on PATH).
+    """Check if LXD is installed.
 
     :returns: True if lxd is installed.
     """
-    return shutil.which("lxd") is not None
+    logger.debug("Checking if LXD is installed.")
+
+    # check if non-snap lxd socket exists (for Arch or NixOS)
+    if (
+        pathlib.Path("/var/lib/lxd/unix.socket").is_socket()
+        and shutil.which("lxd") is not None
+    ):
+        return True
+
+    # query snapd API
+    url = "http+unix://%2Frun%2Fsnapd.socket/v2/snaps/lxd"
+    try:
+        snap_info = requests_unixsocket.get(url=url, params={"select": "enabled"})
+    except requests.exceptions.ConnectionError as error:
+        raise errors.ProviderError(
+            brief="Unable to connect to snapd service."
+        ) from error
+
+    try:
+        snap_info.raise_for_status()
+    except requests.exceptions.HTTPError as error:
+        logger.debug(f"Could not get snap info for LXD: {error}")
+        return False
+
+    # the LXD snap should be installed and active but check the status
+    # for completeness
+    try:
+        status = snap_info.json()["result"]["status"]
+    except (TypeError, KeyError):
+        raise errors.ProviderError(brief="Unexpected response from snapd service.")
+
+    logger.debug(f"LXD snap status: {status}")
+    # snap status can be "installed" or "active" - "installed" revisions
+    # are filtered from this API call with `select: enabled`
+    return bool(status == "active") and shutil.which("lxd") is not None
 
 
 def is_user_permitted() -> bool:

pyproject.toml

@@ -4,6 +4,8 @@ dynamic = ["version", "readme"]
 dependencies = [
     "packaging>=14.1",
     "pydantic<2.0",
+    # see https://github.com/psf/requests/issues/6707
+    "requests<2.32",
     "pyyaml",
     # see https://github.com/psf/requests/issues/6707
     "requests<2.32",

tests/unit/lxd/test_installer.py

@@ -16,11 +16,13 @@
 #
 
 import os
-import shutil
 import sys
+from typing import Any, Dict
 from unittest import mock
+from unittest.mock import call
 
 import pytest
+from craft_providers.errors import ProviderError
 from craft_providers.lxd import (
     LXC,
     LXD,
@@ -301,13 +303,45 @@ def test_is_initialized_no_disk_device(devices):
     assert not initialized
 
 
+@pytest.mark.parametrize(("has_lxd_executable"), [(True), (False)])
+@pytest.mark.parametrize(("has_nonsnap_socket"), [(True), (False)])
 @pytest.mark.parametrize(
-    ("which", "installed"), [("/path/to/lxd", True), (None, False)]
+    ("status", "exception", "installed"),
+    [
+        ({"result": {"status": "active"}}, None, True),
+        ({"result": {"status": "foo"}}, None, False),
+        ({}, ProviderError, False),
+        (None, ProviderError, False),
+    ],
 )
-def test_is_installed(which, installed, monkeypatch):
-    monkeypatch.setattr(shutil, "which", lambda x: which)
-
-    assert is_installed() == installed
+def test_is_installed(
+    mocker, has_nonsnap_socket, has_lxd_executable, status, exception, installed
+):
+    class FakeSnapInfo:
+        def raise_for_status(self) -> None:
+            pass
+
+        def json(self) -> Dict[str, Any]:
+            return status
+
+    mock_get = mocker.patch("requests_unixsocket.get", return_value=FakeSnapInfo())
+    mocker.patch("pathlib.Path.is_socket", return_value=has_nonsnap_socket)
+    mocker.patch("shutil.which", return_value="lxd" if has_lxd_executable else None)
+
+    if has_nonsnap_socket and has_lxd_executable:
+        assert is_installed()
+        return
+
+    if exception:
+        with pytest.raises(exception):
+            is_installed()
+    else:
+        assert is_installed() == (installed and has_lxd_executable)
+
+    assert mock_get.mock_calls[0] == call(
+        url="http+unix://%2Frun%2Fsnapd.socket/v2/snaps/lxd",
+        params={"select": "enabled"},
+    )
 
 
 @pytest.mark.skipif(sys.platform != "linux", reason=f"unsupported on {sys.platform}")