Get Pro status in managed LXD instance

craft_providers/lxd/lxc.py

@@ -18,6 +18,7 @@
 """LXC wrapper."""
 import contextlib
 import enum
+import json
 import logging
 import os
 import pathlib
@@ -1173,3 +1174,47 @@
             time.sleep(3)
 
         raise LXDError(brief="Timed out waiting for instance to be ready.")
+
+    def is_pro_enabled(
+        self,
+        *,
+        instance_name: str,
+        project: str = "default",
+        remote: str = "local",
+    ) -> bool:
+        """Check whether Pro is enabled on the instance.
+
+        :param instance_name: Name of instance.
+
+        :returns: True if instance is attached.
+        """
+        command = [
+            "exec",
+            f"{remote}:{instance_name}",
+            "pro",
+            "api",
+            "u.pro.status.is_attached.v1",
+        ]
+
+        try:
+            proc = self._run_lxc(command, capture_output=True, project=project)
+            data = json.loads(proc.stdout)
+            if data.get("result") == "success":
+                if data.get("data", {}).get("attributes", {}).get("is_attached"):
+                    logger.debug("Managed instance is Pro enabled.")
+                    return True
+                logger.debug("Managed instance is not Pro enabled.")
+                return False
+
+            raise LXDError(
+                brief=f"Failed to get a successful response from `pro` command on {instance_name!r}.",
+            )
+        except json.JSONDecodeError as error:
+            raise LXDError(
+                brief=f"Failed to parse JSON response of `pro` command on {instance_name!r}.",
+            ) from error
+        except subprocess.CalledProcessError as error:
+            raise LXDError(
+                brief=f"Failed to run `pro` command on {instance_name!r}.",
+                details=errors.details_from_called_process_error(error),
+            ) from error

craft_providers/lxd/lxd_instance.py

@@ -647,3 +647,16 @@ def info(self) -> Dict[str, Any]:
             project=self.project,
             remote=self.remote,
         )
+
+    def is_pro_enabled(self) -> bool:
+        """Check whether the instance is Pro enabled.
+
+        :returns: True if the instance is Pro enabled.
+
+        :raises: LXDError: On unexpected error.
+        """
+        return self.lxc.is_pro_enabled(
+            instance_name=self.instance_name,
+            project=self.project,
+            remote=self.remote,
+        )

pyproject.toml

@@ -41,7 +41,7 @@ dev = [
     "pytest-rerunfailures==14.0",
     "pytest-subprocess==1.5.2",
     "pytest-xdist==3.6.1",
-    "pytest-time==0.3.1",
+    "pytest-time==0.3.2",
     "responses==0.25.3",
     # types-requests>=2.31.0.7 requires urllib3>=2
     "types-requests==2.31.0.6",
@@ -54,8 +54,8 @@ lint = [
     "yamllint==1.35.1",
 ]
 types = [
-    "mypy[reports]==1.11.1",
-    "pyright==1.1.377",
+    "mypy[reports]==1.11.2",
+    "pyright==1.1.379",
 ]
 docs = [
     "pyspelling==2.10",

tests/integration/lxd/test_lxc.py

@@ -33,6 +33,17 @@ def instance(instance_name, session_project):
         yield tmp_instance
 
 
+@pytest.fixture
+def instance_alma(instance_name, session_project):
+    with conftest.tmp_instance(
+        name=instance_name,
+        project=session_project,
+        image="almalinux/9",
+        image_remote="images",
+    ) as tmp_instance:
+        yield tmp_instance
+
+
 def test_launch_default_config(instance, lxc, session_project):
     """Verify default config values when launching."""
     status = lxc.config_get(
@@ -271,3 +282,25 @@ def test_info(instance, lxc, session_project):
     data = lxc.info(instance_name=instance, project=session_project)
 
     assert data["Name"] == instance
+
+
+def test_is_pro_enabled_ubuntu(instance, lxc, session_project):
+    """Test the scenario where Pro client is installed."""
+    result = lxc.is_pro_enabled(
+        instance_name=instance,
+        project=session_project,
+    )
+
+    # Assert the instance is not Pro enabled
+    assert result is False
+
+
+def test_is_pro_enabled_alma(instance_alma, lxc, session_project):
+    """Test the scenario where Pro client is not installed."""
+    with pytest.raises(LXDError) as raised:
+        lxc.is_pro_enabled(
+            instance_name=instance_alma,
+            project=session_project,
+        )
+
+    assert raised.value.brief == (f"Failed to run `pro` command on {instance_alma!r}.")

tests/unit/lxd/test_lxc.py

@@ -2579,3 +2579,121 @@ def test_yaml_loader_invalid_timestamp():
 
     assert "last_used_at" in obj
     assert isinstance(obj["last_used_at"], str)
+
+
+def test_is_pro_enabled_success(fake_process):
+    fake_process.register_subprocess(
+        [
+            "lxc",
+            "--project",
+            "test-project",
+            "exec",
+            "test-remote:test-instance",
+            "pro",
+            "api",
+            "u.pro.status.is_attached.v1",
+        ],
+        stdout=b"""{"_schema_version": "v1", "data": {"attributes": {"contract_remaining_days": 2912917, "contract_status": "active", "is_attached": true, "is_attached_and_contract_valid": true}, "meta": {"environment_vars": []}, "type": "IsAttached"}, "errors": [], "result": "success", "version": "32.3.1~24.04", "warnings": []}""",
+    )
+
+    assert (
+        LXC().is_pro_enabled(
+            instance_name="test-instance",
+            project="test-project",
+            remote="test-remote",
+        )
+        is True
+    )
+
+    assert len(fake_process.calls) == 1
+
+
+def test_is_pro_enabled_failed(fake_process):
+    fake_process.register_subprocess(
+        [
+            "lxc",
+            "--project",
+            "test-project",
+            "exec",
+            "test-remote:test-instance",
+            "pro",
+            "api",
+            "u.pro.status.is_attached.v1",
+        ],
+        stdout=b"""{"_schema_version": "v1", "data": {"attributes": {"contract_remaining_days": 2912917, "contract_status": "active", "is_attached": true, "is_attached_and_contract_valid": true}, "meta": {"environment_vars": []}, "type": "IsAttached"}, "errors": [], "result": "failure", "version": "32.3.1~24.04", "warnings": []}""",
+        returncode=0,
+    )
+
+    with pytest.raises(LXDError) as exc_info:
+        LXC().is_pro_enabled(
+            instance_name="test-instance",
+            project="test-project",
+            remote="test-remote",
+        )
+
+    assert exc_info.value == LXDError(
+        brief="Failed to get a successful response from `pro` command on 'test-instance'.",
+    )
+
+    assert len(fake_process.calls) == 1
+
+
+def test_is_pro_enabled_json_error(fake_process):
+    fake_process.register_subprocess(
+        [
+            "lxc",
+            "--project",
+            "test-project",
+            "exec",
+            "test-remote:test-instance",
+            "pro",
+            "api",
+            "u.pro.status.is_attached.v1",
+        ],
+        stdout=b"random",
+    )
+
+    with pytest.raises(LXDError) as exc_info:
+        LXC().is_pro_enabled(
+            instance_name="test-instance",
+            project="test-project",
+            remote="test-remote",
+        )
+
+    assert exc_info.value == LXDError(
+        brief="Failed to parse JSON response of `pro` command on 'test-instance'.",
+    )
+
+    assert len(fake_process.calls) == 1
+
+
+def test_is_pro_enabled_process_error(fake_process):
+    fake_process.register_subprocess(
+        [
+            "lxc",
+            "--project",
+            "test-project",
+            "exec",
+            "test-remote:test-instance",
+            "pro",
+            "api",
+            "u.pro.status.is_attached.v1",
+        ],
+        returncode=127,
+    )
+
+    with pytest.raises(LXDError) as exc_info:
+        LXC().is_pro_enabled(
+            instance_name="test-instance",
+            project="test-project",
+            remote="test-remote",
+        )
+
+    assert exc_info.value == LXDError(
+        brief="Failed to run `pro` command on 'test-instance'.",
+        details=errors.details_from_called_process_error(
+            exc_info.value.__cause__  # type: ignore
+        ),
+    )
+
+    assert len(fake_process.calls) == 1

tests/unit/lxd/test_lxd_instance.py

@@ -1050,3 +1050,15 @@ def test_set_instance_name_invalid(mock_lxc, name):
         brief=f"failed to create LXD instance with name {name!r}.",
         details="name must contain at least one alphanumeric character",
     )
+
+
+def test_is_pro_enabled(mock_lxc, instance):
+    instance.is_pro_enabled()
+
+    assert mock_lxc.mock_calls == [
+        mock.call.is_pro_enabled(
+            instance_name=instance.instance_name,
+            project=instance.project,
+            remote=instance.remote,
+        )
+    ]