diff --git a/craft_application/application.py b/craft_application/application.py index 36ab639a6..19ba4ad4b 100644 --- a/craft_application/application.py +++ b/craft_application/application.py @@ -33,11 +33,13 @@ import craft_cli import craft_platforms import craft_providers +import craft_providers.lxd from craft_parts.errors import PartsError from platformdirs import user_cache_path from craft_application import _config, commands, errors, models, util -from craft_application.errors import PathInvalidError +from craft_application.errors import InvalidUbuntuProStatusError, PathInvalidError +from craft_application.util import ProServices, ValidatorOptions if TYPE_CHECKING: import argparse @@ -162,6 +164,10 @@ def __init__( # Set a globally usable project directory for the application. # This may be overridden by specific application implementations. self.project_dir = pathlib.Path.cwd() + # Ubuntu ProServices instance containing relevant pro services specified by the user. + # Storage of this instance may change in the future as we migrate Pro operations towards + # an application service. + self._pro_services: ProServices | None = None if self.is_managed(): self._work_dir = pathlib.Path("/root") @@ -344,6 +350,7 @@ def _configure_services(self, provider_name: str | None) -> None: "lifecycle", cache_dir=self.cache_dir, work_dir=self._work_dir, + use_host_sources=bool(self._pro_services), ) self.services.update_kwargs( "provider", @@ -387,6 +394,41 @@ def is_managed(self) -> bool: """Shortcut to tell whether we're running in managed mode.""" return self.services.get_class("provider").is_managed() + def _configure_instance_with_pro(self, instance: craft_providers.Executor) -> None: + """Configure an instance with Ubuntu Pro. Currently we only support LXD instances.""" + # TODO: Remove craft_provider typing ignores after feature/pro-sources # noqa: FIX002 + # has been merged into main. + + # Check if the instance has pro services enabled and if they match the requested services. + # If not, raise an Exception and bail out. + if ( + isinstance(instance, craft_providers.lxd.LXDInstance) + and instance.pro_services is not None # type: ignore # noqa: PGH003 + and instance.pro_services != self._pro_services # type: ignore # noqa: PGH003 + ): + raise InvalidUbuntuProStatusError(self._pro_services, instance.pro_services) # type: ignore # noqa: PGH003 + + # if pro services are required, ensure the pro client is + # installed, attached and the correct services are enabled + if self._pro_services: + # Suggestion: create a Pro abstract class used to ensure minimum support by instances. + # we can then check for pro support by inheritance. + if not isinstance(instance, craft_providers.lxd.LXDInstance): + raise errors.UbuntuProNotSupportedError( + "Ubuntu Pro builds are only supported with LXC backend." + ) + + craft_cli.emit.debug( + f"Enabling Ubuntu Pro Services {self._pro_services}, {set(self._pro_services)}" + ) + instance.install_pro_client() # type: ignore # noqa: PGH003 + instance.attach_pro_subscription() # type: ignore # noqa: PGH003 + instance.enable_pro_service(self._pro_services) # type: ignore # noqa: PGH003 + + # Cache the current pro services, for prior checks in reentrant calls. + if self._pro_services is not None: + instance.pro_services = self._pro_services # type: ignore # noqa: PGH003 + def run_managed(self, platform: str | None, build_for: str | None) -> None: """Run the application in a managed instance.""" build_planner = self.services.get("build_plan") @@ -571,6 +613,37 @@ def _pre_run(self, dispatcher: craft_cli.Dispatcher) -> None: resolution="Ensure the path entered is correct.", ) + @staticmethod + def _check_pro_requirement( + pro_services: ProServices | None, + run_managed: bool, # noqa: FBT001 + is_managed: bool, # noqa: FBT001 + ) -> None: + craft_cli.emit.debug( + f"pro_services: {pro_services}, run_managed: {run_managed}, is_managed: {is_managed}" + ) + if pro_services is not None: # should not be None for all lifecycle commands. + # Validate requested pro services on the host if we are running in destructive mode. + if not run_managed and not is_managed: + craft_cli.emit.debug( + f"Validating requested Ubuntu Pro status on host: {pro_services}" + ) + pro_services.validate_environment() + # Validate requested pro services running in managed mode inside a managed instance. + elif run_managed and is_managed: + craft_cli.emit.debug( + f"Validating requested Ubuntu Pro status in managed instance: {pro_services}" + ) + pro_services.validate_environment() + # Validate pro attachment and service names on the host before starting a managed instance. + elif run_managed and not is_managed: + craft_cli.emit.debug( + f"Validating requested Ubuntu Pro attachment on host: {pro_services}" + ) + pro_services.validate_environment( + options=ValidatorOptions.AVAILABILITY, + ) + fetch_service_policy: str | None = getattr(args, "fetch_service_policy", None) if fetch_service_policy: self._enable_fetch_service = True @@ -591,63 +664,63 @@ def get_arg_or_config(self, parsed_args: argparse.Namespace, item: str) -> Any: def _run_inner(self) -> int: """Actual run implementation.""" dispatcher = self._get_dispatcher() - command = cast( - commands.AppCommand, - dispatcher.load_command(self.app_config), - ) - parsed_args = dispatcher.parsed_args() - platform = self.get_arg_or_config(parsed_args, "platform") - build_for = self.get_arg_or_config(parsed_args, "build_for") - - # Some commands (e.g. remote build) can allow multiple platforms - # or build-fors, comma-separated. In these cases, we create the - # project using the first defined platform. - if platform and "," in platform: - platform = platform.split(",", maxsplit=1)[0] - if build_for and "," in build_for: - build_for = build_for.split(",", maxsplit=1)[0] - craft_cli.emit.debug(f"Build plan: platform={platform}, build_for={build_for}") - - self._pre_run(dispatcher) - - if command.needs_project(parsed_args): - project_service = self.services.get("project") - # This branch always runs, except during testing. - if not project_service.is_configured: - project_service.configure(platform=platform, build_for=build_for) - - managed_mode = command.run_managed(parsed_args) - provider_name = command.provider_name(parsed_args) - self._configure_services(provider_name) + craft_cli.emit.debug("Preparing application...") return_code = 1 # General error - if not managed_mode: - # command runs in the outer instance - craft_cli.emit.debug(f"Running {self.app.name} {command.name} on host") - return_code = dispatcher.run() or os.EX_OK - elif not self.is_managed(): - # command runs in inner instance, but this is the outer instance - self.run_managed(platform, build_for) - return_code = os.EX_OK - else: - # command runs in inner instance - return_code = dispatcher.run() or 0 + try: + command = cast( + commands.AppCommand, + dispatcher.load_command(self.app_config), + ) - return return_code + platform = getattr(dispatcher.parsed_args(), "platform", None) + build_for = getattr(dispatcher.parsed_args(), "build_for", None) - def run(self) -> int: - """Bootstrap and run the application.""" - self._setup_logging() - self._configure_early_services() - self._initialize_craft_parts() - self._load_plugins() + run_managed = command.run_managed(dispatcher.parsed_args()) + is_managed = self.is_managed() - craft_cli.emit.debug("Preparing application...") + # Some commands (e.g. remote build) can allow multiple platforms + # or build-fors, comma-separated. In these cases, we create the + # project using the first defined platform. + if platform and "," in platform: + platform = platform.split(",", maxsplit=1)[0] + if build_for and "," in build_for: + build_for = build_for.split(",", maxsplit=1)[0] + craft_cli.emit.debug( + f"Build plan: platform={platform}, build_for={build_for}" + ) - debug_mode = self.services.get("config").get("debug") + # A ProServices instance will only be available for lifecycle commands, + # which may consume pro packages, + self._pro_services = getattr(dispatcher.parsed_args(), "pro", None) + # Check that pro services are correctly configured if available + self._check_pro_requirement( + self._pro_services, run_managed, self.is_managed() + ) - try: - return_code = self._run_inner() + if run_managed or command.needs_project(dispatcher.parsed_args()): + self.services.project = self.get_project( + platform=platform, build_for=build_for + ) + + craft_cli.emit.debug( + f"Build plan: platform={platform}, build_for={build_for}" + ) + self._pre_run(dispatcher) + + self._configure_services(provider_name) + + if not run_managed: + # command runs in the outer instance + craft_cli.emit.debug(f"Running {self.app.name} {command.name} on host") + return_code = dispatcher.run() or os.EX_OK + elif not is_managed: + # command runs in inner instance, but this is the outer instance + self.run_managed(platform, build_for) + return_code = os.EX_OK + else: + # command runs in inner instance + return_code = dispatcher.run() or 0 except craft_cli.ArgumentParsingError as err: print(err, file=sys.stderr) # to stderr, as argparse normally does craft_cli.emit.ended_ok() diff --git a/craft_application/commands/lifecycle.py b/craft_application/commands/lifecycle.py index edc4244a3..997d9c958 100644 --- a/craft_application/commands/lifecycle.py +++ b/craft_application/commands/lifecycle.py @@ -28,6 +28,7 @@ from craft_application import errors, models, util from craft_application.commands import base +from craft_application.util import ProServices _PACKED_FILE_LIST_PATH = ".craft/packed-files" @@ -141,6 +142,23 @@ def _fill_parser(self, parser: argparse.ArgumentParser) -> None: help="Shell into the environment after the step has run.", ) + supported_pro_services = ", ".join( + [f"'{name}'" for name in ProServices.supported_services] + ) + + parser.add_argument( + "--pro", + type=ProServices.from_csv, + metavar="", + help=( + "Enable Ubuntu Pro services for this command. " + f"Supported values include: {supported_pro_services}. " + "Multiple values can be passed separated by commas. " + "Note: This feature requires an Ubuntu Pro compatible host and build base." + ), + default=ProServices(), + ) + parser.add_argument( "--debug", action="store_true", diff --git a/craft_application/errors.py b/craft_application/errors.py index e7f9c6b08..3f97640d0 100644 --- a/craft_application/errors.py +++ b/craft_application/errors.py @@ -384,3 +384,110 @@ class ArtifactCreationError(CraftError): class StateServiceError(CraftError): """Errors related to the state service.""" + + +class UbuntuProError(CraftError): + """Base Exception class for ProServices.""" + + +class UbuntuProApiError(UbuntuProError): + """Base class for exceptions raised during Ubuntu Pro Api calls.""" + + +class InvalidUbuntuProStateError(UbuntuProError): + """Base class for exceptions raised during Ubuntu Pro validation.""" + + # TODO: some of the resolution strings may not sense in a managed # noqa: FIX002 + # environment. What is the best way to get the is_managed method here? + + +class UbuntuProNotSupportedError(UbuntuProError): + """Raised when Ubuntu Pro client is not supported on the base or build base.""" + + +class UbuntuProClientNotFoundError(UbuntuProApiError): + """Raised when Ubuntu Pro client was not found on the system.""" + + def __init__(self, path: str) -> None: + message = f'The Ubuntu Pro client was not found on the system at "{path}"' + + super().__init__(message=message) + + +class UbuntuProDetachedError(InvalidUbuntuProStateError): + """Raised when Ubuntu Pro is not attached, but Pro services were requested.""" + + def __init__(self) -> None: + message = "Ubuntu Pro is requested, but was found detached." + resolution = 'Attach Ubuntu Pro to continue. See "pro" command for details.' + + super().__init__(message=message, resolution=resolution) + + +class UbuntuProAttachedError(InvalidUbuntuProStateError): + """Raised when Ubuntu Pro is attached, but Pro services were not requested.""" + + def __init__(self) -> None: + message = "Ubuntu Pro is not requested, but was found attached." + resolution = 'Detach Ubuntu Pro to continue. See "pro" command for details.' + + super().__init__(message=message, resolution=resolution) + + +class InvalidUbuntuProServiceError(InvalidUbuntuProStateError): + """Raised when the requested Ubuntu Pro service is not supported or invalid.""" + + def __init__(self, invalid_services: set[str] | None) -> None: + invalid_services_set = invalid_services or set() + invalid_services_str = "".join(invalid_services_set) + + message = "Invalid Ubuntu Pro Services were requested." + resolution = ( + "The services listed are either not supported by this application " + "or are invalid Ubuntu Pro Services.\n" + f"Invalid Services: {invalid_services_str}\n" + 'See "--pro" argument details for supported services.' + ) + + super().__init__(message=message, resolution=resolution) + + +class InvalidUbuntuProBaseError(InvalidUbuntuProStateError): + """Raised when the requested base, (or build_base) do not support Ubuntu Pro Builds.""" + + def __init__(self, base_type: str, base_name: str) -> None: + message = f'Ubuntu Pro builds are not supported on "{base_name}" {base_type}.' + resolution = f"Remove --pro argument or set {base_type} to a supported base." + + super().__init__(message=message, resolution=resolution) + + +class InvalidUbuntuProStatusError(InvalidUbuntuProStateError): + """Raised when the incorrect set of Pro Services are enabled.""" + + def __init__( + self, + requested_services: set[str] | None, + available_services: set[str] | None, + ) -> None: + requested_services_set = requested_services or set() + available_services_set = available_services or set() + + enable_services_str = str(requested_services_set - available_services_set) + disable_services_str = str(available_services_set - requested_services_set) + message = "Incorrect Ubuntu Pro Services were enabled." + + if "container" in os.environ: + resolution = ( + "Please enable or disable the following services.\n" + f"Enable: {enable_services_str}\n" + f"Disable: {disable_services_str}\n" + 'See "pro" command for details.' + ) + else: + app_name = os.environ.get("SNAP_INSTANCE_NAME", "*craft") + resolution = ( + f'Please run "{app_name} clean" to reset Ubuntu Pro Services.\n' + ) + + super().__init__(message=message, resolution=resolution) diff --git a/craft_application/util/__init__.py b/craft_application/util/__init__.py index aade87a8c..bd6e7a9d3 100644 --- a/craft_application/util/__init__.py +++ b/craft_application/util/__init__.py @@ -43,6 +43,7 @@ from craft_application.util.system import get_parallel_build_count from craft_application.util.yaml import dump_yaml, safe_yaml_load from craft_application.util.cli import format_timestamp +from craft_application.util.pro_services import ProServices, ValidatorOptions __all__ = [ "get_unique_callbacks", @@ -69,4 +70,8 @@ "get_hostname", "is_managed_mode", "format_timestamp", + "ProServices", + "ValidatorOptions", + "get_hostname", + "is_managed_mode", ] diff --git a/craft_application/util/pro_services.py b/craft_application/util/pro_services.py new file mode 100644 index 000000000..2d04fff39 --- /dev/null +++ b/craft_application/util/pro_services.py @@ -0,0 +1,251 @@ +# This file is part of craft_application. +# +# 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 warranties of MERCHANTABILITY, +# SATISFACTORY QUALITY, 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, see . +"""Handling of Ubuntu Pro Services.""" + +from __future__ import annotations + +import json +import logging +import subprocess as sub +from enum import Flag, auto +from io import TextIOWrapper +from pathlib import Path +from typing import Any + +import yaml + +from craft_application import models +from craft_application.errors import ( + InvalidUbuntuProBaseError, + InvalidUbuntuProServiceError, + InvalidUbuntuProStatusError, + UbuntuProApiError, + UbuntuProAttachedError, + UbuntuProClientNotFoundError, + UbuntuProDetachedError, +) + +logger = logging.getLogger(__name__) + + +# check for pro client in these paths for backwards compatibility. +PRO_CLIENT_PATHS = [ + Path("/usr/bin/ubuntu-advantage"), + Path("/usr/bin/ua"), + Path("/usr/bin/pro"), +] + + +class ValidatorOptions(Flag): + """Options for ProServices.validate method. + + SUPPORT: Check names in ProServices set against supported services. + AVAILABILITY: Check Ubuntu Pro is attached if ProServices are valid. + ATTACHMENT: Check Ubuntu Pro is attached or detached to match ProServices set. + ENABLEMENT: Check enabled Ubuntu Pro enablement matches ProServices set. + """ + + SUPPORT = auto() + ATTACHED = auto() + DETACHED = auto() + ATTACHMENT = ATTACHED | DETACHED + ENABLEMENT = auto() + DEFAULT = SUPPORT | ATTACHMENT | ENABLEMENT + AVAILABILITY = ATTACHED | SUPPORT + + +class ProServices(set[str]): + """Class for managing pro-services within the lifecycle.""" + + # placeholder for empty sets + empty_placeholder = "None" + + supported_services: set[str] = { + "esm-apps", + "esm-infra", + "fips", + # TODO: fips-preview is not part of the spec, but # noqa: FIX002 + # it should be added. Bring this up at regular sync. + "fips-preview", + "fips-updates", + } + + # location of pro client + pro_executable: Path | None = next( + (path for path in PRO_CLIENT_PATHS if path.exists()), None + ) + # locations to check for pro client + + def __str__(self) -> str: + """Convert to string for display to user.""" + services = ", ".join(self) if self else self.empty_placeholder + return f"" + + @classmethod + def load_yaml(cls, f: TextIOWrapper) -> ProServices: + """Create a new ProServices instance from a yaml file.""" + serialized_data = yaml.safe_load(f) + return cls(serialized_data) + + def save_yaml(self, f: TextIOWrapper) -> None: + """Save the ProServices instance to a yaml file.""" + yaml.safe_dump(set(self), f) + + @classmethod + def from_csv(cls, services: str) -> ProServices: + """Create a new ProServices instance from a csv string.""" + split = [service.strip() for service in services.split(",")] + return cls(split) + + @classmethod + def pro_client_exists(cls) -> bool: + """Check if Ubuntu Pro executable exists or not.""" + return cls.pro_executable is not None and cls.pro_executable.exists() + + @classmethod + def _log_processes(cls, process: sub.CompletedProcess[str]) -> None: + logger.error( + "Ubuntu Pro Client Response: \n" + f"Return Code: {process.returncode}\n" + f"StdOut:\n{process.stdout}\n" + f"StdErr:\n{process.stderr}\n" + ) + + @classmethod + def _pro_api_call(cls, endpoint: str) -> dict[str, Any]: + """Call Ubuntu Pro executable and parse response.""" + if not cls.pro_client_exists(): + raise UbuntuProClientNotFoundError(str(cls.pro_executable)) + + try: + proc = sub.run( + [str(cls.pro_executable), "api", endpoint], + capture_output=True, + text=True, + check=False, + ) + except Exception as exc: + raise UbuntuProApiError( + f'An error occurred while executing "{cls.pro_executable}"' + ) from exc + + if proc.returncode != 0: + cls._log_processes(proc) + raise UbuntuProApiError( + f"The Pro Client returned a non-zero status: {proc.returncode}. " + "See log for more details" + ) + + try: + result = json.loads(proc.stdout) + + except json.decoder.JSONDecodeError: + cls._log_processes(proc) + raise UbuntuProApiError( + "Could not parse JSON response from Ubuntu Pro client. " + "See log for more details" + ) + + if result["result"] != "success": + cls._log_processes(proc) + raise UbuntuProApiError( + "Ubuntu Pro API returned an error response. See log for more details" + ) + + # Ignore typing for this private method. The returned object is variable in type, but types are declared in the API docs: + # https://canonical-ubuntu-pro-client.readthedocs-hosted.com/en/v32/references/api/ + return result # type: ignore [no-any-return] + + @classmethod + def is_pro_attached(cls) -> bool: + """Return True if environment is attached to Ubuntu Pro.""" + response = cls._pro_api_call("u.pro.status.is_attached.v1") + + # Ignore typing here. This field's type is static according to: + # https://canonical-ubuntu-pro-client.readthedocs-hosted.com/en/v32/references/api/#u-pro-status-is-attached-v1 + return response["data"]["attributes"]["is_attached"] # type: ignore [no-any-return] + + @classmethod + def get_pro_services(cls) -> ProServices: + """Return set of enabled Ubuntu Pro services in the environment. + + The returned set only includes services relevant to lifecycle commands. + """ + response = cls._pro_api_call("u.pro.status.enabled_services.v1") + enabled_services = response["data"]["attributes"]["enabled_services"] + + service_names = {service["name"] for service in enabled_services} + + # remove any services that aren't relevant to build services + service_names = service_names.intersection(cls.supported_services) + + return cls(service_names) + + def validate_project(self, project: models.Project) -> None: + """Ensure no unsupported interim bases are used in Ubuntu Pro builds.""" + invalid_bases = ["devel"] + if bool(self): + if project.base is not None and project.base in invalid_bases: + raise InvalidUbuntuProBaseError("base", project.base) + if project.build_base is not None and project.build_base in invalid_bases: + raise InvalidUbuntuProBaseError("build-base", project.build_base) + + def validate_environment( + self, + options: ValidatorOptions = ValidatorOptions.DEFAULT, + ) -> None: + """Validate the environment against pro services specified in this ProServices instance.""" + # raise exception if any service was requested outside of build_service_scope + if ValidatorOptions.SUPPORT in options and ( + invalid_services := self - self.supported_services + ): + raise InvalidUbuntuProServiceError(invalid_services) + + try: + # first, check Ubuntu Pro status + # Since we extend the set class, cast ourselves to bool to check if we empty. if we are not + # empty, this implies we require pro services. + + is_pro_attached = self.is_pro_attached() + + if ( + ValidatorOptions.ATTACHED in options + and bool(self) + and not is_pro_attached + ): + # Ubuntu Pro is requested but not attached + raise UbuntuProDetachedError + + if ( + ValidatorOptions.DETACHED in options + and not bool(self) + and is_pro_attached + ): + # Ubuntu Pro is not requested but attached + raise UbuntuProAttachedError + + # second, check that the set of enabled pro services in the environment matches + # the services specified in this set + if ValidatorOptions.ENABLEMENT in options and ( + (available_services := self.get_pro_services()) != self + ): + raise InvalidUbuntuProStatusError(self, available_services) + + except UbuntuProClientNotFoundError: + # If The pro client was not found, we may be on a non Ubuntu + # system, but if Pro services were requested, re-raise error + if self and not self.pro_client_exists(): + raise diff --git a/tests/conftest.py b/tests/conftest.py index 12015110f..5e28c6f03 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -101,6 +101,12 @@ def fake_platform(request: pytest.FixtureRequest) -> str: return request.param +@pytest.fixture(autouse=True) +def reset_services(): + yield + service_factory.ServiceFactory.reset() + + @pytest.fixture def platform_independent_project(fake_project_file, fake_project_dict): """Turn the fake project into a platform-independent project. @@ -590,3 +596,42 @@ def repository_with_unannotated_tag( subprocess.run(["git", "tag", test_tag], check=True) repository_with_commit.tag = test_tag return repository_with_commit + + +@pytest.fixture +def mock_pro_api_call(mocker): + mock_responses = { + "u.pro.status.is_attached.v1": { + "data": { + "attributes": {"is_attached": False}, + "result": "success", + } + }, + "u.pro.status.enabled_services.v1": { + "data": {"attributes": {"enabled_services": []}}, + "result": "success", + }, + } + + def set_is_attached(value: bool): # noqa: FBT001 + response = mock_responses["u.pro.status.is_attached.v1"] + response["data"]["attributes"]["is_attached"] = value + + def set_enabled_services(service_names: list[str]): + enabled_services = [ + {"name": name, "variant_enabled": False, "variant_name": None} + for name in service_names + ] + + response = mock_responses["u.pro.status.enabled_services.v1"] + response["data"]["attributes"]["enabled_services"] = enabled_services + + def mock_pro_api_call(endpoint: str): + return mock_responses[endpoint] + + mocker.patch( + "craft_application.util.ProServices._pro_api_call", + new=mock_pro_api_call, + ) + + return set_is_attached, set_enabled_services diff --git a/tests/unit/commands/test_lifecycle.py b/tests/unit/commands/test_lifecycle.py index 2a6fc5782..66eb65272 100644 --- a/tests/unit/commands/test_lifecycle.py +++ b/tests/unit/commands/test_lifecycle.py @@ -19,6 +19,7 @@ import pathlib import subprocess from unittest import mock +from contextlib import nullcontext import craft_parts import craft_platforms @@ -44,6 +45,17 @@ from craft_parts import Features pytestmark = [pytest.mark.usefixtures("fake_project_file")] +from craft_application.errors import ( + InvalidUbuntuProBaseError, + InvalidUbuntuProServiceError, + InvalidUbuntuProStatusError, + UbuntuProAttachedError, + UbuntuProDetachedError, +) +from craft_application.util import ProServices + +# disable black reformat for improve readability on long parameterisations +# fmt: off PARTS_LISTS = [[], ["my-part"], ["my-part", "your-part"]] SHELL_PARAMS = [ @@ -61,6 +73,40 @@ ({"destructive_mode": True, "use_lxd": False}, ["--destructive-mode"]), ({"destructive_mode": False, "use_lxd": True}, ["--use-lxd"]), ] + +# test paring --pro argument with various pro services, and whitespace +PRO_SERVICE_COMMANDS = [ + ({"pro": ProServices()}, []), + ({"pro": ProServices(["fips-updates"])}, ["--pro", "fips-updates"]), + ({"pro": ProServices(["fips-updates", "esm-infra"])}, ["--pro", "fips-updates,esm-infra"]), + ({"pro": ProServices(["fips-updates", "esm-infra"])}, ["--pro", "fips-updates , esm-infra"]), + ({"pro": ProServices(["fips-updates"])}, ["--pro", "fips-updates,fips-updates"]), +] + +PRO_SERVICE_CONFIGS = [ + # is_attached, enabled_services, pro_services_args, expected_exception + (False, [], [], None), + (True, ["esm-apps"], ["esm-apps"], None), + (True, ["esm-apps", "fips-updates"],["esm-apps", "fips-updates"], None), + (True, ["esm-apps"], [], UbuntuProAttachedError), + (False, [], ["esm-apps"], UbuntuProDetachedError), + (True, ["esm-apps", "fips-updates"],["fips-updates"], InvalidUbuntuProStatusError), + (True, ["esm-apps"], ["fips-updates", "fips-updates"], InvalidUbuntuProStatusError), + (True, ["esm-apps"], ["esm-apps", "invalid-service"], InvalidUbuntuProServiceError), +] + +PRO_PROJECT_CONFIGS = [ + # base build_base pro_services_args expected_exception + ("ubuntu@20.04", None, "esm-apps", None), + ("ubuntu@20.04", "ubuntu@20.04", "esm-apps", None), + ("devel", None, "esm-apps", InvalidUbuntuProBaseError), + ("ubuntu@20.04", "devel", "esm-apps", InvalidUbuntuProBaseError), + ("devel", "ubuntu@20.04", "esm-apps", InvalidUbuntuProBaseError), + ("devel", None, None, None), + ("ubuntu@20.04", "devel", None, None), + ("devel", "ubuntu@20.04", None, None), +] + STEP_NAMES = [step.name.lower() for step in craft_parts.Step] MANAGED_LIFECYCLE_COMMANDS = ( PullCommand, @@ -73,6 +119,8 @@ ALL_LIFECYCLE_COMMANDS = MANAGED_LIFECYCLE_COMMANDS + UNMANAGED_LIFECYCLE_COMMANDS NON_CLEAN_COMMANDS = (*MANAGED_LIFECYCLE_COMMANDS, PackCommand) +# fmt: on + def get_fake_command_class(parent_cls, managed): """Create a fully described fake command based on a partial class.""" @@ -89,6 +137,58 @@ def run_managed(self, parsed_args: argparse.Namespace) -> bool: return FakeCommand +@pytest.mark.parametrize( + ("is_attached", "enabled_services", "pro_services_args", "expected_exception"), + PRO_SERVICE_CONFIGS, +) +def pro_services_validate_environment( + mock_pro_api_call, + is_attached, + enabled_services, + pro_services_args, + expected_exception, +): + # configure api state + set_is_attached, set_enabled_service = mock_pro_api_call + set_is_attached(is_attached) + set_enabled_service(enabled_services) + + exception_context = ( + pytest.raises(expected_exception) if expected_exception else nullcontext() + ) + + with exception_context: + # create and validate pro services + pro_services = ProServices(pro_services_args) + pro_services.validate_environment() + + +@pytest.mark.parametrize( + ("base", "build_base", "pro_services_args", "expected_exception"), + PRO_PROJECT_CONFIGS, +) +def pro_services_validate_project( + mocker, + base, + build_base, + pro_services_args, + expected_exception, +): + # configure project + project = mocker.Mock() + project.base = base + project.build_base = build_base + + exception_context = ( + pytest.raises(expected_exception) if expected_exception else nullcontext() + ) + + with exception_context: + # create and validate pro services + pro_services = ProServices(pro_services_args) + pro_services.validate_project(project) + + @pytest.mark.parametrize( ("enable_overlay", "commands"), [ @@ -117,12 +217,15 @@ def test_get_lifecycle_command_group(enable_overlay, commands): Features.reset() +@pytest.mark.parametrize(("pro_service_dict", "pro_service_args"), PRO_SERVICE_COMMANDS) @pytest.mark.parametrize(("build_env_dict", "build_env_args"), BUILD_ENV_COMMANDS) @pytest.mark.parametrize(("debug_dict", "debug_args"), DEBUG_PARAMS) @pytest.mark.parametrize(("shell_dict", "shell_args"), SHELL_PARAMS) def test_lifecycle_command_fill_parser( default_app_metadata, fake_services, + pro_service_dict, + pro_service_args, build_env_dict, build_env_args, debug_dict, @@ -139,11 +242,16 @@ def test_lifecycle_command_fill_parser( **shell_dict, **debug_dict, **build_env_dict, + **pro_service_dict, } command.fill_parser(parser) - args_dict = vars(parser.parse_args([*build_env_args, *debug_args, *shell_args])) + args_dict = vars( + parser.parse_args( + [*pro_service_args, *build_env_args, *debug_args, *shell_args] + ) + ) assert args_dict == expected @@ -291,6 +399,7 @@ def test_run_manager_for_build_plan( mock_run_managed.assert_called_once_with(build, fetch) +@pytest.mark.parametrize(("pro_service_dict", "pro_service_args"), PRO_SERVICE_COMMANDS) @pytest.mark.parametrize(("build_env_dict", "build_env_args"), BUILD_ENV_COMMANDS) @pytest.mark.parametrize(("debug_dict", "debug_args"), DEBUG_PARAMS) @pytest.mark.parametrize(("shell_dict", "shell_args"), SHELL_PARAMS) @@ -298,6 +407,8 @@ def test_run_manager_for_build_plan( def test_step_command_fill_parser( default_app_metadata, fake_services, + pro_service_dict, + pro_service_args, parts_args, build_env_dict, build_env_args, @@ -315,13 +426,16 @@ def test_step_command_fill_parser( **shell_dict, **debug_dict, **build_env_dict, + **pro_service_dict, } command = cls({"app": default_app_metadata, "services": fake_services}) command.fill_parser(parser) args_dict = vars( - parser.parse_args([*build_env_args, *shell_args, *debug_args, *parts_args]) + parser.parse_args( + [*pro_service_args, *build_env_args, *shell_args, *debug_args, *parts_args] + ) ) assert args_dict == expected @@ -460,6 +574,44 @@ def test_clean_run_without_parts( assert mock_services.provider.clean_instances.called == expected_provider +@pytest.mark.parametrize( + ("destructive", "build_env", "parts", "expected_run_managed"), + [ + # destructive mode or CRAFT_BUILD_ENV==host should not run managed + (True, "lxd", [], False), + (True, "host", [], False), + (False, "host", [], False), + (True, "lxd", ["part1"], False), + (True, "host", ["part1"], False), + (False, "host", ["part1"], False), + (True, "lxd", ["part1", "part2"], False), + (True, "host", ["part1", "part2"], False), + (False, "host", ["part1", "part2"], False), + # destructive mode==False and CRAFT_BUILD_ENV!=host: depends on "parts" + # clean specific parts: should run managed + (False, "lxd", ["part1"], True), + (False, "lxd", ["part1", "part2"], True), + # "part-less" clean: shouldn't run managed + (False, "lxd", [], False), + ], +) +def test_clean_run_managed( + app_metadata, + mock_services, + destructive, + build_env, + parts, + expected_run_managed, + monkeypatch, +): + monkeypatch.setenv("CRAFT_BUILD_ENVIRONMENT", build_env) + parsed_args = argparse.Namespace(parts=parts, destructive_mode=destructive) + command = CleanCommand({"app": app_metadata, "services": mock_services}) + + assert command.run_managed(parsed_args) == expected_run_managed + + +@pytest.mark.parametrize(("pro_service_dict", "pro_service_args"), PRO_SERVICE_COMMANDS) @pytest.mark.parametrize(("build_env_dict", "build_env_args"), BUILD_ENV_COMMANDS) @pytest.mark.parametrize(("shell_dict", "shell_args"), SHELL_PARAMS) @pytest.mark.parametrize(("debug_dict", "debug_args"), DEBUG_PARAMS) @@ -467,6 +619,8 @@ def test_clean_run_without_parts( def test_pack_fill_parser( app_metadata, mock_services, + pro_service_dict, + pro_service_args, build_env_dict, build_env_args, shell_dict, @@ -486,6 +640,7 @@ def test_pack_fill_parser( **shell_dict, **debug_dict, **build_env_dict, + **pro_service_dict, } command = PackCommand({"app": app_metadata, "services": mock_services}) @@ -493,7 +648,13 @@ def test_pack_fill_parser( args_dict = vars( parser.parse_args( - [*build_env_args, *shell_args, *debug_args, f"--output={output_arg}"] + [ + *pro_service_args, + *build_env_args, + *shell_args, + *debug_args, + f"--output={output_arg}", + ] ) ) assert args_dict == expected diff --git a/tests/unit/models/test_project.py b/tests/unit/models/test_project.py index 192acb7f2..dc793f914 100644 --- a/tests/unit/models/test_project.py +++ b/tests/unit/models/test_project.py @@ -24,6 +24,7 @@ import craft_platforms import craft_providers.bases import pytest + from craft_application import util from craft_application.errors import CraftValidationError from craft_application.models import ( diff --git a/tests/unit/test_application.py b/tests/unit/test_application.py index e8db292ab..7b67e003e 100644 --- a/tests/unit/test_application.py +++ b/tests/unit/test_application.py @@ -35,6 +35,7 @@ import craft_providers import pytest import pytest_check + from craft_application import ( application, commands, @@ -48,12 +49,16 @@ ) from craft_application.util import ( get_host_architecture, # pyright: ignore[reportGeneralTypeIssues] + ValidatorOptions, ) from craft_cli import emit from craft_parts.plugins.plugins import PluginType +from craft_providers import bases, lxd +from overrides import override from tests.conftest import FakeApplication + EMPTY_COMMAND_GROUP = craft_cli.CommandGroup("FakeCommands", []) @@ -297,6 +302,80 @@ def test_run_managed_failure(app, fake_project): assert exc_info.value.brief == "Failed to execute testcraft in instance." +def test_run_managed_configure_pro(mocker, app, fake_project, fake_build_plan): + """Ensure that Pro is installed and configured in a managed instance.""" + mock_provider = mocker.MagicMock(spec_set=services.ProviderService) + app.services.provider = mock_provider + # provide spec to pass type check for pro support + mock_instance = mocker.MagicMock(spec=lxd.LXDInstance) + + # TODO: these methods are currently in review, https://github.com/canonical/craft-providers/pull/664/files # noqa: FIX002 + # Remove when craft-providers with pro support in lxd is merged to main. + mock_instance.install_pro_client = mocker.Mock() + mock_instance.attach_pro_subscription = mocker.Mock() + mock_instance.enable_pro_service = mocker.Mock() + # pretend to be a fresh instance w/o any services installed + mock_instance.pro_services = None + + mock_provider.instance.return_value.__enter__.return_value = mock_instance + app.project = fake_project + app._build_plan = fake_build_plan + app._pro_services = util.ProServices(["esm-apps"]) + arch = get_host_architecture() + + app.run_managed(None, arch) + + assert mock_instance.install_pro_client.call_count == 1 + assert mock_instance.attach_pro_subscription.call_count == 1 + assert mock_instance.enable_pro_service.call_count == 1 + + +def test_run_managed_pro_not_supported(mocker, app, fake_project, fake_build_plan): + """Ensure that providers that do not support Ubuntu Pro cause an exception + when pro services are requested.""" + mock_provider = mocker.MagicMock(spec_set=services.ProviderService) + app.services.provider = mock_provider + # Provide an unsupported instance type to raise exception + mock_instance = mocker.MagicMock() + + mock_provider.instance.return_value.__enter__.return_value = mock_instance + app.project = fake_project + app._build_plan = fake_build_plan + app._pro_services = util.ProServices(["esm-apps"]) + arch = get_host_architecture() + + with pytest.raises(errors.UbuntuProNotSupportedError): + app.run_managed(None, arch) + + +def test_run_managed_skip_configure_pro(mocker, app, fake_project, fake_build_plan): + """Ensure that Pro is not installed and configured when it is not required.""" + mock_provider = mocker.MagicMock(spec_set=services.ProviderService) + app.services.provider = mock_provider + # provide spec to pass type check for pro support + mock_instance = mocker.MagicMock(spec=lxd.LXDInstance) + + # TODO: Remove when these mocks when methods are present in main. # noqa: FIX002 + # see TODO in test_run_managed_configure_pro for details. + mock_instance.install_pro_client = mocker.Mock() + mock_instance.attach_pro_subscription = mocker.Mock() + mock_instance.enable_pro_service = mocker.Mock() + # pretend to be a fresh instance w/o any services installed + mock_instance.pro_services = None + + mock_provider.instance.return_value.__enter__.return_value = mock_instance + app.project = fake_project + app._build_plan = fake_build_plan + app._pro_services = util.ProServices([]) + arch = get_host_architecture() + + app.run_managed(None, arch) + + assert mock_instance.install_pro_client.call_count == 0 + assert mock_instance.attach_pro_subscription.call_count == 0 + assert mock_instance.enable_pro_service.call_count == 0 + + def test_run_managed_multiple(app, fake_host_architecture): mock_provider = mock.MagicMock(spec_set=services.ProviderService) app.services._services["provider"] = mock_provider @@ -462,7 +541,15 @@ def test_craft_lib_log_level(app_metadata, fake_services): assert logger.level == logging.DEBUG -def test_gets_project(monkeypatch, fake_project_file, app_metadata, fake_services): +def test_gets_project( + monkeypatch, + tmp_path, + app_metadata, + fake_services, + mock_pro_api_call, +): + project_file = tmp_path / "testcraft.yaml" + project_file.write_text(BASIC_PROJECT_YAML) monkeypatch.setattr(sys, "argv", ["testcraft", "pull", "--destructive-mode"]) app = FakeApplication(app_metadata, fake_services) @@ -474,7 +561,13 @@ def test_gets_project(monkeypatch, fake_project_file, app_metadata, fake_service def test_fails_without_project( - monkeypatch, capsys, tmp_path, app_metadata, fake_services, app, debug_mode + monkeypatch, + capsys, + tmp_path, + app_metadata, + fake_services, + app, + mock_pro_api_call, # noqa: ARG001 ): # Set up a real project service - the fake one for testing gets a fake project! del app.services._services["project"] @@ -562,7 +655,11 @@ def test_pre_run_project_dir_success_unmanaged(app, fs, project_dir): @pytest.mark.parametrize("project_dir", ["relative/file", "/absolute/file"]) -def test_pre_run_project_dir_not_a_directory(app, fs, project_dir): +def test_pre_run_project_dir_not_a_directory( + app, + fs, + project_dir, +): fs.create_file(project_dir) dispatcher = mock.Mock(spec_set=craft_cli.Dispatcher) dispatcher.parsed_args.return_value.project_dir = project_dir @@ -574,7 +671,14 @@ def test_pre_run_project_dir_not_a_directory(app, fs, project_dir): @pytest.mark.parametrize("load_project", [True, False]) @pytest.mark.parametrize("return_code", [None, 0, 1]) def test_run_success_unmanaged( - monkeypatch, emitter, check, app, fake_project, return_code, load_project + monkeypatch, + emitter, + check, + app, + fake_project, + return_code, + load_project, + mock_pro_api_call, ): class UnmanagedCommand(commands.AppCommand): name = "pass" @@ -597,9 +701,101 @@ def run(self, parsed_args: argparse.Namespace): emitter.assert_debug("Running testcraft pass on host") +def test_run_success_managed( + monkeypatch, + app, + fake_project, + mocker, + mock_pro_api_call, +): + mocker.patch.object(app, "get_project", return_value=fake_project) + app.run_managed = mock.Mock() + monkeypatch.setattr(sys, "argv", ["testcraft", "pull"]) + + pytest_check.equal(app.run(), 0) + + app.run_managed.assert_called_once_with(None, None) # --build-for not used + + +def test_run_success_managed_with_arch( + monkeypatch, + app, + fake_project, + mocker, + mock_pro_api_call, +): + mocker.patch.object(app, "get_project", return_value=fake_project) + app.run_managed = mock.Mock() + arch = get_host_architecture() + monkeypatch.setattr(sys, "argv", ["testcraft", "pull", f"--build-for={arch}"]) + + pytest_check.equal(app.run(), 0) + + app.run_managed.assert_called_once() + + +def test_run_success_managed_with_platform( + monkeypatch, + app, + fake_project, + mocker, + mock_pro_api_call, +): + mocker.patch.object(app, "get_project", return_value=fake_project) + app.run_managed = mock.Mock() + monkeypatch.setattr(sys, "argv", ["testcraft", "pull", "--platform=foo"]) + + pytest_check.equal(app.run(), 0) + + app.run_managed.assert_called_once_with("foo", None) + + +@pytest.mark.parametrize( + ("params", "expected_call"), + [ + ([], mock.call(None, None)), + (["--platform=s390x"], mock.call("s390x", None)), + ( + ["--platform", get_host_architecture()], + mock.call(get_host_architecture(), None), + ), + ( + ["--build-for", get_host_architecture()], + mock.call(None, get_host_architecture()), + ), + (["--build-for", "s390x"], mock.call(None, "s390x")), + (["--platform", "s390x,riscv64"], mock.call("s390x", None)), + (["--build-for", "s390x,riscv64"], mock.call(None, "s390x")), + ], +) +def test_run_passes_platforms( + monkeypatch, + app, + fake_project, + mocker, + params, + expected_call, + mock_pro_api_call, +): + mocker.patch.object(app, "get_project", return_value=fake_project) + app.run_managed = mock.Mock(return_value=False) + monkeypatch.setattr(sys, "argv", ["testcraft", "pull", *params]) + + pytest_check.equal(app.run(), 0) + + assert app.run_managed.mock_calls == [expected_call] + + @pytest.mark.parametrize("return_code", [None, 0, 1]) def test_run_success_managed_inside_managed( - monkeypatch, check, app, fake_project, mock_dispatcher, return_code, mocker + monkeypatch, + check, + app, + fake_project, + mock_dispatcher, + return_code, + mocker, + mock_pro_api_call, ): mocker.patch.object(app, "get_project", return_value=fake_project) mocker.patch.object( @@ -1126,3 +1322,49 @@ def test_doc_url_in_command_help(monkeypatch, capsys, app): expected = "For more information, check out: www.testcraft.example/docs/3.14159/reference/commands/app-config\n\n" _, err = capsys.readouterr() assert err.endswith(expected) + + +# fmt: off +@pytest.mark.parametrize( + ( "run_managed", "is_managed", "val_env_calls", "validator_options"), + [ + (False, False, 1, None), + (True, True, 1, None), + (True, False, 1, ValidatorOptions.AVAILABILITY), + (False, True, 0, None), + ], +) +# fmt: on +def test_check_pro_requirement( + mocker, app, run_managed, is_managed, val_env_calls, validator_options +): + """Test that _check_pro_requirement validates Pro Services in the correct situations""" + pro_services = mocker.Mock() + app._check_pro_requirement(pro_services, run_managed, is_managed) + + assert pro_services.validate_environment.call_count == val_env_calls + + for call in pro_services.validate.call_args_list: + if validator_options is not None: # skip assert if default value is passed + assert call.kwargs["options"] == validator_options + + +# fmt: off +@pytest.mark.parametrize( + ( "is_pro", "val_prj_calls"), + [ + (False, 0), + (True, 1), + ], +) +# fmt: on +@pytest.mark.usefixtures("fake_project_file") +def test_validate_project_pro_requirements(mocker, is_pro, val_prj_calls, app_metadata, fake_services): + """Test validate_project is called only when ProServices are available.""" + app = application.Application(app_metadata, fake_services) + + app._pro_services = mocker.MagicMock() + + app._pro_services.__bool__.return_value = is_pro # type: ignore # noqa: PGH003 + app.get_project(build_for=get_host_architecture()) + assert app._pro_services.validate_project.call_count == val_prj_calls # type: ignore # noqa: PGH003