Merge pull request canonical#684 from canonical/feature/namespaced-pa…

…rtitions

chore: merge namespaced partitions into main

craft_parts/dirs.py

@@ -1,6 +1,6 @@
 # -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
 #
-# Copyright 2021 Canonical Ltd.
+# Copyright 2021,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
@@ -18,7 +18,7 @@
 
 from pathlib import Path
 from types import MappingProxyType
-from typing import Dict, Optional, Sequence, Union
+from typing import Optional, Sequence, Union
 
 from craft_parts.utils import partition_utils
 
@@ -54,35 +54,34 @@ def __init__(
         self.overlay_mount_dir = self.overlay_dir / "overlay"
         self.overlay_packages_dir = self.overlay_dir / "packages"
         self.overlay_work_dir = self.overlay_dir / "work"
-        self.base_stage_dir = self.work_dir / "stage"
-        self.base_prime_dir = self.work_dir / "prime"
+        self.stage_dir = self.work_dir / "stage"
+        self.prime_dir = self.work_dir / "prime"
         if partitions:
-            self._partitions: Sequence[Optional[str]] = partitions
-            self.stage_dir = self.base_stage_dir / "default"
-            self.prime_dir = self.base_prime_dir / "default"
-            stage_dirs: Dict[Optional[str], Path] = {
-                part: self.base_stage_dir / part for part in partitions
-            }
-            prime_dirs: Dict[Optional[str], Path] = {
-                part: self.base_prime_dir / part for part in partitions
-            }
+            self._partitions: Optional[Sequence[str]] = partitions
+            self.partition_dir: Optional[Path] = self.work_dir / "partitions"
         else:
-            self._partitions = [None]
-            self.stage_dir = self.base_stage_dir
-            self.prime_dir = self.base_prime_dir
-            stage_dirs = {None: self.stage_dir}
-            prime_dirs = {None: self.prime_dir}
-        self.stage_dirs = MappingProxyType(stage_dirs)
-        self.prime_dirs = MappingProxyType(prime_dirs)
+            self._partitions = None
+            self.partition_dir = None
+
+        self.stage_dirs = MappingProxyType(
+            partition_utils.get_partition_dir_map(
+                base_dir=self.work_dir, partitions=partitions, suffix="stage"
+            )
+        )
+        self.prime_dirs = MappingProxyType(
+            partition_utils.get_partition_dir_map(
+                base_dir=self.work_dir, partitions=partitions, suffix="prime"
+            )
+        )
 
     def get_stage_dir(self, partition: Optional[str] = None) -> Path:
         """Get the stage directory for the given partition."""
-        if partition not in self._partitions:
+        if self._partitions and partition not in self._partitions:
             raise ValueError(f"Unknown partition {partition}")
         return self.stage_dirs[partition]
 
     def get_prime_dir(self, partition: Optional[str] = None) -> Path:
         """Get the stage directory for the given partition."""
-        if partition not in self._partitions:
+        if self._partitions and partition not in self._partitions:
             raise ValueError(f"Unknown partition {partition}")
         return self.prime_dirs[partition]

craft_parts/errors.py

@@ -18,7 +18,7 @@
 
 import dataclasses
 import pathlib
-from typing import TYPE_CHECKING, Iterable, List, Optional, Set, Union
+from typing import TYPE_CHECKING, Iterable, List, Optional, Set
 
 if TYPE_CHECKING:
     from pydantic.error_wrappers import ErrorDict, Loc
@@ -52,12 +52,12 @@ def __str__(self) -> str:
 class FeatureError(PartsError):
     """A feature is not configured as expected."""
 
-    def __init__(self, message: str) -> None:
+    def __init__(self, message: str, details: Optional[str] = None) -> None:
         self.message = message
         brief = message
         resolution = "This operation cannot be executed."
 
-        super().__init__(brief=brief, resolution=resolution)
+        super().__init__(brief=brief, details=details, resolution=resolution)
 
 
 class PartDependencyCycle(PartsError):
@@ -389,23 +389,32 @@ class PartFilesConflict(PartsError):
     :param part_name: The name of the part being processed.
     :param other_part_name: The name of the conflicting part.
     :param conflicting_files: The list of conflicting files.
+    :param partition: Optional name of the partition where the conflict occurred.
     """
 
     def __init__(
-        self, *, part_name: str, other_part_name: str, conflicting_files: List[str]
+        self,
+        *,
+        part_name: str,
+        other_part_name: str,
+        conflicting_files: List[str],
+        partition: Optional[str] = None,
     ) -> None:
         self.part_name = part_name
         self.other_part_name = other_part_name
         self.conflicting_files = conflicting_files
+        self.partition = partition
+
         indented_conflicting_files = (f"    {i}" for i in conflicting_files)
         file_paths = "\n".join(sorted(indented_conflicting_files))
+        partition_info = f" for the {partition!r} partition" if partition else ""
         brief = (
             "Failed to stage: parts list the same file "
             "with different contents or permissions."
         )
         details = (
             f"Parts {part_name!r} and {other_part_name!r} list the following "
-            f"files, but with different contents or permissions:\n"
+            f"files{partition_info}, but with different contents or permissions:\n"
             f"{file_paths}"
         )
 
@@ -622,46 +631,51 @@ class PartitionError(PartsError):
 
     def __init__(
         self,
-        partition: str,
         brief: str,
         *,
         details: Optional[str] = None,
         resolution: Optional[str] = None,
     ) -> None:
-        self.partition = partition
         super().__init__(brief=brief, details=details, resolution=resolution)
 
 
-class InvalidPartitionError(PartitionError):
-    """Partition is not valid for this application."""
+class PartitionUsageError(PartitionError):
+    """Error for a list of invalid partition usages.
+
+    :param error_list: Iterable of strings describing the invalid usages.
+    """
 
     def __init__(
-        self,
-        partition: str,
-        path: Union[str, pathlib.Path],
-        valid_partitions: Iterable[str],
+        self, error_list: Iterable[str], partitions: Optional[Iterable[str]]
     ) -> None:
-        self.valid_partitions = valid_partitions
+        valid_partitions = (
+            f"\nValid partitions: {', '.join(partitions)}" if partitions else ""
+        )
+
         super().__init__(
-            partition,
-            brief=f"Invalid partition {partition!r} in path {str(path)!r}",
-            details="Valid partitions: " + ", ".join(valid_partitions),
-            resolution="Correct the invalid partition name and try again.",
+            brief="Invalid usage of partitions",
+            details="\n".join(error_list) + valid_partitions,
+            resolution="Correct the invalid partition name(s) and try again.",
         )
 
 
-class PartitionWarning(PartitionError, Warning):
-    """Warnings for partition-related items."""
+class PartitionUsageWarning(PartitionError, Warning):
+    """Warnings for possibly invalid usages of partitions.
 
-    def __init__(
-        self,
-        partition: str,
-        brief: str,
-        *,
-        details: Optional[str] = None,
-        resolution: Optional[str] = None,
-    ) -> None:
+    :param warning_list: Iterable of strings describing the misuses.
+    """
+
+    def __init__(self, warning_list: Iterable[str]) -> None:
         super().__init__(
-            partition=partition, brief=brief, details=details, resolution=resolution
+            brief="Possible misuse of partitions",
+            details=(
+                "The following entries begin with a valid partition name but are "
+                "not wrapped in parentheses. These entries will go into the "
+                "default partition.\n" + "\n".join(warning_list)
+            ),
+            resolution=(
+                "Wrap the partition name in parentheses, for example "
+                "'default/file' should be written as '(default)/file'"
+            ),
         )
         Warning.__init__(self)

craft_parts/executor/collisions.py

@@ -1,6 +1,6 @@
 # -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
 #
-# Copyright 2015-2021 Canonical Ltd.
+# Copyright 2015-2021,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
@@ -23,14 +23,53 @@
 from craft_parts import errors, permissions
 from craft_parts.executor import filesets
 from craft_parts.executor.filesets import Fileset
+from craft_parts.features import Features
 from craft_parts.parts import Part
 from craft_parts.permissions import Permissions, permissions_are_compatible
 
 
-def check_for_stage_collisions(part_list: List[Part]) -> None:
+def check_for_stage_collisions(
+    part_list: List[Part], partitions: Optional[List[str]]
+) -> None:
     """Verify whether parts have conflicting files to stage.
 
-    :param part_list: The list of parts to be tested.
+    If the partitions feature is enabled, then check if parts have conflicting files to
+        stage for each partition.
+    If the partitions feature is disabled, only check for conflicts in the default
+        stage directory.
+
+    :param part_list: The list of parts to check.
+    :param partitions: An optional list of partition names.
+
+    :raises PartConflictError: If conflicts are found.
+    :raises FeatureError: If partitions are specified but the feature is not enabled or
+        if partitions are not specified and the feature is enabled.
+    """
+    if partitions and not Features().enable_partitions:
+        raise errors.FeatureError(
+            "Partitions specified but partitions feature is not enabled."
+        )
+
+    if partitions is None and Features().enable_partitions:
+        raise errors.FeatureError(
+            "Partitions feature is enabled but no partitions specified."
+        )
+
+    for partition in partitions or [None]:  # type: ignore[list-item]
+        _check_for_stage_collisions_per_partition(part_list, partition)
+
+
+def _check_for_stage_collisions_per_partition(
+    part_list: List[Part], partition: Optional[str]
+) -> None:
+    """Verify whether parts have conflicting files for a stage directory in a partition.
+
+    If no partition is provided, then the default stage directory is checked.
+
+    :param part_list: The list of parts to check.
+    :param partition: If the partitions feature is enabled, then the name of the
+        partition containing the stage directory to check.
+
     :raises PartConflictError: If conflicts are found.
     """
     all_parts_files: Dict[str, Dict[str, Any]] = {}
@@ -41,9 +80,9 @@ def check_for_stage_collisions(part_list: List[Part]) -> None:
 
         # Gather our own files up.
         stage_fileset = Fileset(stage_files, name="stage")
-        srcdir = str(part.part_install_dir)
+        srcdir = str(part.part_install_dirs[partition])
         part_files, part_directories = filesets.migratable_filesets(
-            stage_fileset, srcdir
+            stage_fileset, srcdir, partition
         )
         part_contents = part_files | part_directories
 
@@ -54,7 +93,7 @@ def check_for_stage_collisions(part_list: List[Part]) -> None:
 
             conflict_files = []
             for file in common:
-                this = os.path.join(part.part_install_dir, file)
+                this = os.path.join(part.part_install_dirs[partition], file)
                 other = os.path.join(other_part_files["installdir"], file)
 
                 permissions_this = permissions.filter_permissions(
@@ -73,12 +112,13 @@ def check_for_stage_collisions(part_list: List[Part]) -> None:
                     part_name=part.name,
                     other_part_name=other_part_name,
                     conflicting_files=conflict_files,
+                    partition=partition,
                 )
 
         # And add our files to the list.
         all_parts_files[part.name] = {
             "files": part_contents,
-            "installdir": part.part_install_dir,
+            "installdir": part.part_install_dirs[partition],
             "part": part,
         }
 

craft_parts/executor/environment.py

@@ -165,15 +165,7 @@ def _get_global_environment(info: ProjectInfo) -> Dict[str, str]:
         global_environment["CRAFT_OVERLAY"] = str(info.overlay_mount_dir)
 
     if Features().enable_partitions:
-        if not info.partitions:
-            raise errors.FeatureError("Partitions enabled but no partitions specified.")
-        for partition in info.partitions:
-            global_environment[f"CRAFT_{partition.upper()}_STAGE"] = str(
-                info.get_stage_dir(partition=partition)
-            )
-            global_environment[f"CRAFT_{partition.upper()}_PRIME"] = str(
-                info.get_prime_dir(partition=partition)
-            )
+        global_environment.update(_get_environment_for_partitions(info))
 
     global_environment["CRAFT_STAGE"] = str(info.stage_dir)
     global_environment["CRAFT_PRIME"] = str(info.prime_dir)
@@ -184,6 +176,37 @@ def _get_global_environment(info: ProjectInfo) -> Dict[str, str]:
     return global_environment
 
 
+def _get_environment_for_partitions(info: ProjectInfo) -> Dict[str, str]:
+    """Get environment variables related to partitions.
+
+    Assumes the partition feature is enabled.
+
+    :param info: The project information.
+
+    :returns: A dictionary contain environment variables for partitions.
+
+    :raises FeatureError: If the Project does not specify any partitions.
+    """
+    environment: Dict[str, str] = {}
+
+    if not info.partitions:
+        raise errors.FeatureError("Partitions enabled but no partitions specified.")
+
+    for partition in info.partitions:
+        formatted_partition = partition.upper().translate(
+            {ord("-"): "_", ord("/"): "_"}
+        )
+
+        environment[f"CRAFT_{formatted_partition}_STAGE"] = str(
+            info.get_stage_dir(partition=partition)
+        )
+        environment[f"CRAFT_{formatted_partition}_PRIME"] = str(
+            info.get_prime_dir(partition=partition)
+        )
+
+    return environment
+
+
 def _get_step_environment(step_info: StepInfo) -> Dict[str, str]:
     """Add project and part information variables to the environment.