feat: merge feature/pro-sources into main

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()

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="<pro-services>",
+            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",

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)

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",
 ]