diff --git a/craft_providers/lxd/installer.py b/craft_providers/lxd/installer.py index 887cc592..71e3c4c8 100644 --- a/craft_providers/lxd/installer.py +++ b/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: diff --git a/pyproject.toml b/pyproject.toml index 98fb787e..29cdb69b 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/unit/lxd/test_installer.py b/tests/unit/lxd/test_installer.py index b572408c..5f49c80e 100644 --- a/tests/unit/lxd/test_installer.py +++ b/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}")