partitions: validate usage of partitions in parts

craft_parts/lifecycle_manager.py

@@ -30,7 +30,7 @@
 from craft_parts.features import Features
 from craft_parts.infos import ProjectInfo
 from craft_parts.overlays import LayerHash
-from craft_parts.parts import Part, part_by_name
+from craft_parts.parts import Part, part_by_name, validate_partition_usage
 from craft_parts.state_manager import states
 from craft_parts.steps import Step
 
@@ -148,6 +148,9 @@ def __init__(
         for name, spec in parts_data.items():
             part_list.append(_build_part(name, spec, project_dirs, strict_mode))
 
+        if partitions:
+            validate_partition_usage(part_list, partitions)
+
         self._has_overlay = any(p.has_overlay for p in part_list)
 
         # a base layer is mandatory if overlays are in use

craft_parts/parts.py

@@ -16,8 +16,9 @@
 
 """Definitions and helpers to handle parts."""
 
+import re
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Sequence, Set
+from typing import Any, Dict, Iterable, List, Optional, Sequence, Set
 
 from pydantic import BaseModel, Field, ValidationError, root_validator, validator
 
@@ -28,6 +29,7 @@
 from craft_parts.permissions import Permissions
 from craft_parts.plugins.properties import PluginProperties
 from craft_parts.steps import Step
+from craft_parts.utils.formatting_utils import humanize_list
 
 
 class PartSpec(BaseModel):
@@ -320,6 +322,62 @@ def has_overlay(self) -> bool:
         """Return whether this part declares overlay content."""
         return self.spec.has_overlay
 
+    def check_partition_usage(self, partitions: List[str]) -> List[str]:
+        """Check if partitions are properly used in a part.
+
+        :param partitions: The list of valid partitions.
+
+        :returns: A list of invalid uses of partitions in the part.
+        """
+        error_list: List[str] = []
+
+        if not Features().enable_partitions:
+            return error_list
+
+        for fileset_name, fileset in [
+            ("overlay", self.spec.overlay_files),
+            # only the destination of organize filepaths use partitions
+            ("organize", self.spec.organize_files.values()),
+            ("stage", self.spec.stage_files),
+            ("prime", self.spec.prime_files),
+        ]:
+            error_list.extend(
+                self._check_partitions_in_filepaths(fileset_name, fileset, partitions)
+            )
+
+        return error_list
+
+    def _check_partitions_in_filepaths(
+        self, fileset_name: str, fileset: Iterable[str], partitions: List[str]
+    ) -> List[str]:
+        """Check if partitions are properly used in a fileset.
+
+        If a filepath begins with a parentheses, then the text inside the parentheses
+        must be a valid partition.
+
+        :param fileset_name: The name of the fileset being checked.
+        :param fileset: The list of filepaths to check.
+        :param partitions: The list of valid partitions.
+
+        :returns: A list of invalid uses of partitions in the fileset.
+        """
+        error_list = []
+        pattern = re.compile("^-?\\((?P<partition>.*?)\\)")
+
+        for filepath in fileset:
+            match = re.match(pattern, filepath)
+            if match:
+                partition = match.group("partition")
+                if partition not in partitions:
+                    error_list.append(
+                        f"    unknown partition {partition!r} in {filepath!r}"
+                    )
+
+        if error_list:
+            error_list.insert(0, f"  parts.{self.name}.{fileset_name}")
+
+        return error_list
+
 
 def part_by_name(name: str, part_list: List[Part]) -> Part:
     """Obtain the part with the given name from the part list.
@@ -469,6 +527,33 @@ def validate_part(data: Dict[str, Any]) -> None:
     _get_part_spec(data)
 
 
+def validate_partition_usage(part_list: List[Part], partitions: List[str]) -> None:
+    """Validate usage of partitions in a list of parts.
+
+    For each part, the use of partitions in filepaths in overlay, stage, prime, and
+    organize keywords are validated.
+
+    :param part_list: The list of parts to validate.
+    :param partitions: The list of valid partitions.
+
+    :raises ValueError: If partitions are not used properly in the list of parts.
+    """
+    if not Features().enable_partitions:
+        return
+
+    error_list = []
+
+    for part in part_list:
+        error_list.extend(part.check_partition_usage(partitions))
+
+    if error_list:
+        raise ValueError(
+            "Error: Invalid usage of partitions:\n"
+            + "\n".join(error_list)
+            + f"\nValid partitions are {humanize_list(partitions, 'and')}."
+        )
+
+
 def part_has_overlay(data: Dict[str, Any]) -> bool:
     """Whether the part described by ``data`` employs the Overlay step.
 

tests/conftest.py

@@ -82,6 +82,18 @@ def enable_partitions_feature():
     Features.reset()
 
 
+@pytest.fixture
+def enable_all_features():
+    assert Features().enable_overlay is False
+    assert Features().enable_partitions is False
+    Features.reset()
+    Features(enable_overlay=True, enable_partitions=True)
+
+    yield
+
+    Features.reset()
+
+
 @pytest.fixture(autouse=True)
 def temp_xdg(tmpdir, mocker):
     """Use a temporary locaction for XDG directories."""

tests/unit/features/partitions/test_lifecycle_manager.py

@@ -17,6 +17,7 @@
 """Unit tests for the lifecycle manager with the partitions feature."""
 
 from string import ascii_lowercase
+from textwrap import dedent
 from typing import Any, Dict
 
 import pytest
@@ -44,6 +45,11 @@ def valid_partitions_strategy():
 class TestPartitionsSupport:
     """Verify LifecycleManager supports partitions."""
 
+    @pytest.fixture
+    def partition_list(self):
+        """Return a list of partitions, 'default' and 'kernel'."""
+        return ["default", "kernel"]
+
     @pytest.fixture
     def parts_data(self) -> Dict[str, Any]:
         return {"parts": {"foo": {"plugin": "nil"}}}
@@ -174,3 +180,48 @@ def test_partitions_duplicates(self, new_dir, parts_data, partitions):
             )
 
         assert str(raised.value) == "Partitions must be unique."
+
+    def test_partitions_usage_valid(self, new_dir, partition_list):
+        """Verify partitions can be used in parts when creating a LifecycleManager."""
+        parts_data = {
+            "parts": {
+                "foo": {
+                    "plugin": "nil",
+                    "stage": ["(default)/foo"],
+                },
+            }
+        }
+
+        # nothing to assert, just ensure an exception is not raised
+        LifecycleManager(
+            parts_data,
+            application_name="test_manager",
+            cache_dir=new_dir,
+            partitions=partition_list,
+        )
+
+    def test_partitions_usage_invalid(self, new_dir, partition_list):
+        """Invalid uses of partitions are raised when creating a LifecycleManager."""
+        parts_data = {
+            "parts": {
+                "foo": {
+                    "plugin": "nil",
+                    "stage": ["(test)/foo"],
+                },
+            }
+        }
+        with pytest.raises(ValueError) as raised:
+            LifecycleManager(
+                parts_data,
+                application_name="test_manager",
+                cache_dir=new_dir,
+                partitions=partition_list,
+            )
+
+        assert str(raised.value) == dedent(
+            """\
+            Error: Invalid usage of partitions:
+              parts.foo.stage
+                unknown partition 'test' in '(test)/foo'
+            Valid partitions are 'default' and 'kernel'."""
+        )