diff --git a/craft_providers/hookutil.py b/craft_providers/hookutil.py new file mode 100644 index 00000000..bbc6742c --- /dev/null +++ b/craft_providers/hookutil.py @@ -0,0 +1,273 @@ +# Copyright 2024 Canonical Ltd. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License version 3 as published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with this program; if not, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. +# + +"""Utilities for use in snap hooks. + +A base instance's full name may look like this: + base-instance-whatevercraft-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24 + +From that, the thing we care most about is the compatibility tag: + whatevercraft-buildd-base-v7 +""" + +import dataclasses +import json +import re +import subprocess +import sys +from typing import Any + +from typing_extensions import Self + +from craft_providers import Base, lxd + +_BASE_INSTANCE_START_STRING = "base-instance" +_CURRENT_COMPATIBILITY_TAG_REGEX = re.compile( + f"^{_BASE_INSTANCE_START_STRING}.*-{Base.compatibility_tag}-.*" +) + + +class HookError(Exception): + """Hook logic cannot continue. Hooks themselves should not exit nonzero.""" + + +@dataclasses.dataclass +class LXDInstance: + """Represents an lxc instance.""" + + name: str + expanded_config: dict[str, str] + + def base_instance_name(self) -> str: + """Get the full name of the base instance this instance was created from.""" + try: + return self.expanded_config["image.description"] + except KeyError as e: + # Unexpected, cannot continue + raise HookError("Could not get full base name from {self.name}") from e + + def is_current_base_instance(self) -> bool: + """Return true if this is a base instance with the current compat tag.""" + return bool(re.match(_CURRENT_COMPATIBILITY_TAG_REGEX, self.name)) + + def is_base_instance(self) -> bool: + """Return true if this is a base instance.""" + return self.name.startswith(_BASE_INSTANCE_START_STRING) + + @classmethod + def unmarshal( + cls, + src: dict[str, str], + ) -> Self: + """Use this rather than init - the lxc output has a lot of extra fields.""" + return cls( + **{ # type: ignore[arg-type] + k: v + for k, v in src.items() + if k in {f.name for f in dataclasses.fields(cls)} + } + ) + + +class HookHelper: + """Hook business logic.""" + + def __init__(self, *, project_name: str, simulate: bool, debug: bool) -> None: + self.simulate = simulate + self.debug = debug + self._project_name = project_name + + self._check_has_lxd() + self._check_project_exists() + + def _check_has_lxd(self) -> None: + """Check if LXD is installed before doing anything. + + On recent Ubuntu systems, "lxc" might be "/usr/sbin/lxc", which is provided by the + "lxd-installer" package and will install the LXD snap if it's not installed. This + installation can then take a long time if the store is having issues. For the + purposes of the configure and remove hooks we don't want to install LXD just to + check that it has no stale images. + """ + if not lxd.is_installed(): + raise HookError("LXD is not installed.") + + def _check_project_exists(self) -> None: + """Raise HookError if lxc doesn't know about this app.""" + for project in self.lxc("project", "list", proj=False): + if project["name"] == self._project_name: + return + + # Didn't find our project name + raise HookError(f"Project {self._project_name} does not exist in LXD.") + + def dprint(self, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 + """Print messages to stderr if debug=True. + + Can treat this like normal print(), except can also pass an instance + dict as the first argument for some automatic formatting. + """ + if not self.debug: + return + if "file" not in kwargs: + kwargs["file"] = sys.stderr + + print_args = list(args) + if len(args) >= 1 and isinstance(args[0], LXDInstance): + # First arg quacks like an instance object + instance = print_args.pop(0) + print_args += [":", instance.name] + + print(*print_args, **kwargs) + + def lxc( + self, + *args: Any, # noqa: ANN401 + fail_msg: str | None = None, + proj: bool = True, + json_out: bool = True, + ) -> Any: # noqa: ANN401 + """Run lxc commands specified in *args. + + :param fail_msg: Print this if the command returns nonzero. + :param proj: Set to False to not specify lxc project. + :param json_out: If set to False, don't ask lxc for JSON output. + """ + lxc_args = ["lxc"] + if json_out: + lxc_args += ["--format", "json"] + if proj: + lxc_args += ["--project", self._project_name] + lxc_args += args + + try: + out = subprocess.run( + lxc_args, + check=True, + text=True, + capture_output=True, + ).stdout + except FileNotFoundError: + raise HookError("LXD is not installed.") + except subprocess.CalledProcessError as e: + if not fail_msg: + fail_msg = e.stderr + raise HookError(fail_msg) + else: + if not json_out: + return out + try: + return json.loads(out) + except json.decoder.JSONDecodeError as e: + raise HookError(f"Didn't get back JSON: {out}") from e + + def delete_instance(self, instance: LXDInstance) -> None: + """Delete the specified lxc instance.""" + print( + f" > Removing instance {instance.name} in LXD {self._project_name} project..." + ) + if self.simulate: + return + self.lxc( + "delete", + "--force", + instance.name, + fail_msg=f"Failed to remove LXD instance {instance.name}.", + json_out=False, + ) + + def _delete_image(self, image_fingerprint: str) -> None: + """Remove the image.""" + self.lxc("image", "delete", image_fingerprint, json_out=False) + + def delete_all_images(self) -> None: + """Delete all images of the lxc project.""" + for image_fingerprint in self._list_images(): + self._delete_image(image_fingerprint) + + def delete_project(self) -> None: + """Delete this lxc project.""" + print(f"Removing project {self._project_name}") + if self.simulate: + return + self.lxc( + "project", + "delete", + self._project_name, + proj=False, + json_out=False, + ) + + def _list_images(self) -> list[str]: + """Return fingerprints of all images associated with the lxc project.""" + return [image["fingerprint"] for image in self.lxc("image", "list")] + + def list_instances(self) -> list[LXDInstance]: + """Return a list of all instance objects for the project.""" + return [LXDInstance.unmarshal(instance) for instance in self.lxc("list")] + + def list_base_instances(self) -> list[LXDInstance]: + """Return a list of all base instance objects for the project.""" + base_instances = [] + for instance in self.list_instances(): + if not instance.is_base_instance(): + self.dprint(instance, "Not a base instance") + continue + + base_instances.append(instance) + return base_instances + + +def configure_hook(lxc: HookHelper) -> None: + """Cleanup hook run on snap configure.""" + # Keep the newest base instance with the most recent compatibility tag. + delete_base_full_names = set() + for instance in lxc.list_base_instances(): + if instance.is_current_base_instance(): + lxc.dprint(instance, "Base instance is current") + continue + + # This is a base instance but it doesn't match the compat tag, assume it's + # old (not future) and delete it. + lxc.dprint(instance, "Base instance uses old compatibility tag, deleting") + lxc.delete_instance(instance) + delete_base_full_names.add(instance.base_instance_name()) + + if not delete_base_full_names: + lxc.dprint("No base instances were deleted, so no derived instances to delete") + return + + # Find the child instances of the bases we deleted and delete them too + did_delete = False + for instance in lxc.list_instances(): + if instance.base_instance_name() not in delete_base_full_names: + continue + lxc.dprint(instance, "Base instance was deleted, deleting derived instance") + lxc.delete_instance(instance) + did_delete = True + if not did_delete: + lxc.dprint("Found no instances derived from deleted base instances") + + +def remove_hook(lxc: HookHelper) -> None: + """Cleanup hook run on snap removal.""" + for instance in lxc.list_instances(): + lxc.delete_instance(instance) + + # Project deletion will fail if images aren't all deleted first + lxc.delete_all_images() + + lxc.delete_project() diff --git a/tests/integration/test_hookutil.py b/tests/integration/test_hookutil.py new file mode 100644 index 00000000..d2247bcf --- /dev/null +++ b/tests/integration/test_hookutil.py @@ -0,0 +1,95 @@ +# +# Copyright 2025 Canonical Ltd. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License version 3 as published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with this program; if not, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. +# + +"""Tests for snap hook utilities.""" + +import pytest +from craft_providers import lxd +from craft_providers.bases import ubuntu +from craft_providers.hookutil import HookError, HookHelper, configure_hook, remove_hook + +FAKE_PROJECT = "boopcraft" + + +@pytest.fixture +def spawn_lxd_instance(installed_lxd): + base_config = ubuntu.BuilddBase(alias=ubuntu.BuilddBaseAlias.JAMMY) + + def spawn_lxd_instance(name, *, is_base_instance): + """Create a long-lived LXD instance under our fake project.""" + return lxd.launch( + name=name, + base_configuration=base_config, + image_name="22.04", + image_remote="ubuntu", + project=FAKE_PROJECT, + auto_create_project=True, + use_base_instance=not is_base_instance, + ) + + return spawn_lxd_instance + + +def test_configure_hook(spawn_lxd_instance): + # Create a current non-base instance (the base instance is also created internally) + current_instance = spawn_lxd_instance( + "boopcraft-myproject-on-amd64-for-amd64-59510339", + is_base_instance=False, + ) + + # Create an outdated instance that would have been created by craft-providers>=1.7.0<1.8.0 + outdated_base_instance = spawn_lxd_instance( + "base-instance-buildd-base-v00--be83d276b0c767e3ad60", + is_base_instance=True, + ) + + helper = HookHelper(project_name=FAKE_PROJECT, simulate=False, debug=True) + configure_hook(helper) + + assert current_instance.exists(), "Current non-base instance should exist" + assert ( + not outdated_base_instance.exists() + ), "Outdated base instance should not exist" + + current_instance.delete() + helper._check_project_exists() # raises exception if project doesn't exist + + +def test_remove_hook(spawn_lxd_instance): + # Create a current non-base instance (the base instance is also created internally) + current_instance = spawn_lxd_instance( + "boopcraft-myproject-on-amd64-for-amd64-59510339", + is_base_instance=False, + ) + + # Create an outdated instance that would have been created by craft-providers>=1.7.0<1.8.0 + outdated_base_instance = spawn_lxd_instance( + "base-instance-buildd-base-v00--be83d276b0c767e3ad60", + is_base_instance=True, + ) + + helper = HookHelper(project_name=FAKE_PROJECT, simulate=False, debug=True) + remove_hook(helper) + + assert not current_instance.exists(), "Current non-base instance should not exist" + assert ( + not outdated_base_instance.exists() + ), "Outdated base instance should not exist" + + with pytest.raises(HookError) as e: + helper._check_project_exists() + assert e == HookError(f"Project {FAKE_PROJECT} does not exist in LXD.") diff --git a/tests/unit/test_hookutil.py b/tests/unit/test_hookutil.py new file mode 100644 index 00000000..890eb30d --- /dev/null +++ b/tests/unit/test_hookutil.py @@ -0,0 +1,175 @@ +# Copyright 2024 Canonical Ltd. +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License version 3 as published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with this program; if not, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. +# + +from unittest.mock import MagicMock, call + +import pytest +from craft_providers.hookutil import ( + HookError, + HookHelper, + LXDInstance, + configure_hook, + remove_hook, +) + +PROJECT_NAME = "fakeproj" + + +def test_no_projects(): + """Make sure HookError is raised if there is no corresponding lxc project.""" + # Use our own mock HookHelper rather than the fixture, we need to do things a little + # differently here + HookHelper._check_has_lxd = MagicMock() + original_lxc_func = HookHelper.lxc + + def fake_lxc(self, *args, **kwargs): + if len(args) == 2 and args[0:2] == ("project", "list"): + return [] + return original_lxc_func(*args, **kwargs) + + HookHelper.lxc = fake_lxc + + with pytest.raises(HookError) as e: + HookHelper(project_name=PROJECT_NAME, simulate=False, debug=True) + assert f"Project {PROJECT_NAME} does not exist in LXD" in str(e) + + +@pytest.fixture +def fake_hookhelper(): + def fake_hookhelper(instance_list): + HookHelper._check_project_exists = MagicMock() # raise nothing + HookHelper._check_has_lxd = MagicMock() + helper = HookHelper(project_name=PROJECT_NAME, simulate=False, debug=True) + + original_lxc_func = helper.lxc + + def fake_lxc(*args, **kwargs): + if len(args) == 1 and args[0] == "list": + return instance_list + return original_lxc_func(*args, **kwargs) + + helper.lxc = fake_lxc + + helper.delete_instance = MagicMock() + helper.delete_project = MagicMock() + helper.delete_all_images = MagicMock() + return helper + + return fake_hookhelper + + +def assert_instances_deleted(helper, instances): + """Transform json list to instance calls for passing to assert_has_calls.""" + helper.delete_instance.assert_has_calls( + [call(LXDInstance.unmarshal(instance)) for instance in instances], + any_order=True, + ) + + +def test_configure_nothing_to_delete(fake_hookhelper): + """Test the configure hook logic with mocked lxc calls.""" + instances = [ + { + "name": f"base-instance-{PROJECT_NAME}-buildd-base-v7-c-a839ea97c42df2065713", + "created_at": "2024-11-15T03:14:36.041502388Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24", + "image.version": "24.04", + }, + }, + { + "name": f"{PROJECT_NAME}-busybox-gadget-on-amd64-for-amd64-13389833", + "created_at": "2024-11-15T03:15:33.48330342Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24", + "image.version": "24.04", + }, + }, + ] + helper = fake_hookhelper(instances) + + configure_hook(helper) + + helper.delete_instance.assert_not_called() + helper.delete_project.assert_not_called() + + +def test_configure_simple_delete_superseded(fake_hookhelper): + """Test a simple case where some images with out-of-date compat tags are deleted.""" + instances = [ + { + "name": f"base-instance-{PROJECT_NAME}-buildd-base-v7-c-a839ea97c42df2065713", + "created_at": "2024-11-15T03:14:36.041502388Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24", + "image.version": "24.04", + }, + }, + { + "name": f"{PROJECT_NAME}-busybox-gadget-on-amd64-for-amd64-13389833", + "created_at": "2024-11-15T03:15:33.48330342Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24", + "image.version": "24.04", + }, + }, + { + "name": f"base-instance-{PROJECT_NAME}-buildd-base-v6-c-a839ea97c42df2065712", + "created_at": "2024-11-15T02:14:36.041502388Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v6-craft-com.ubuntu.cloud-buildd-daily-core22", + "image.version": "22.04", + }, + }, + { + "name": f"{PROJECT_NAME}-busybox-gadget-on-amd64-for-amd64-13389832", + "created_at": "2024-11-15T02:15:33.48330342Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v6-craft-com.ubuntu.cloud-buildd-daily-core22", + "image.version": "22.04", + }, + }, + ] + helper = fake_hookhelper(instances) + configure_hook(helper) + assert_instances_deleted(helper, instances[2:2]) + + +def test_remove_simple_delete(fake_hookhelper): + """Test the remove hook logic with mocked lxc calls.""" + instances = [ + { + "name": f"base-instance-{PROJECT_NAME}-buildd-base-v7-c-a839ea97c42df2065713", + "created_at": "2024-11-15T03:14:36.041502388Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24", + "image.version": "24.04", + }, + }, + { + "name": f"{PROJECT_NAME}-busybox-gadget-on-amd64-for-amd64-13389833", + "created_at": "2024-11-15T03:15:33.48330342Z", + "expanded_config": { + "image.description": f"base-instance-{PROJECT_NAME}-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24", + "image.version": "24.04", + }, + }, + ] + helper = fake_hookhelper(instances) + remove_hook(helper) + assert_instances_deleted(helper, instances) + helper.delete_all_images.assert_called_once() + helper.delete_project.assert_called_once()