Add config validation for multi-platform images

buildrunner/config.py

@@ -21,6 +21,8 @@
 
 import jinja2
 
+from pydantic import ValidationError
+
 from buildrunner.errors import (
     BuildRunnerConfigurationError,
     BuildRunnerVersionError,
@@ -34,6 +36,8 @@
     load_config,
 )
 
+from buildrunner import config_model
+
 from . import fetch
 
 MASTER_GLOBAL_CONFIG_FILE = '/etc/buildrunner/buildrunner.yaml'
@@ -370,6 +374,12 @@ def load_config(self, cfg_file, ctx=None, log_file=True):
 
         config = self._reorder_dependency_steps(config)
 
+        # Validate the config
+        try:
+            config_model.Config(**config)
+        except (ValidationError, ValueError) as err:
+            raise BuildRunnerConfigurationError(f"Invalid configuration: {err}")    # pylint: disable=raise-missing-from
+
         return config
 
     def get_temp_dir(self):

buildrunner/config_model.py

@@ -0,0 +1,155 @@
+"""
+Copyright 2023 Adobe
+All Rights Reserved.
+
+NOTICE: Adobe permits you to use, modify, and distribute this file in accordance
+with the terms of the Adobe license agreement accompanying it.
+"""
+
+from typing import Dict, List, Optional, Set
+
+from pydantic import BaseModel  # pylint: disable=no-name-in-module
+
+
+class StepBuild(BaseModel):
+    """ Build model within a step """
+    path: Optional[str]
+    dockerfile: Optional[str]
+    pull: Optional[bool]
+    platform: Optional[str]
+    platforms: Optional[List[str]]
+
+
+class StepPushDict(BaseModel):
+    """ Push model within a step """
+    repository: str
+    tags: List[str]
+
+
+class Step(BaseModel):
+    """ Step model """
+    build: Optional[StepBuild]
+    push: Optional[StepPushDict | List[str | StepPushDict] | str]   # pylint: disable=unsubscriptable-object
+
+    def is_multi_platform(self):
+        """
+        Check if the step is a multi-platform build step
+        """
+        return self.build is not None and \
+            self.build.platforms is not None
+
+
+class Config(BaseModel):
+    """ Top level config model """
+    version: Optional[float]
+    steps: Optional[Dict[str, Step]]
+
+    def __init__(self, **data) -> None:
+        super().__init__(**data)
+        self.validate()
+
+    def has_multi_platform_build(self):
+        """
+        Check if the config file has multi-platform build steps
+
+        Returns:
+            bool: True if the config file has multi-platform build steps, False otherwise
+        """
+        if self.steps is None:
+            return False
+
+        for step in self.steps.values():
+            if step.is_multi_platform():
+                return True
+        return False
+
+    def validate_push(self,
+                      # pylint: disable=unsubscriptable-object,unsubscriptable-object
+                      push: StepPushDict | list[str | StepPushDict] | str,
+                      mp_push_tags: Set[str],
+                      step_name: str,
+                      update_mp_push_tags: bool = True):
+        """
+        Validate push step
+
+        Args:
+            push (StepPushDict | list[str | StepPushDict] | str): Push step
+            mp_push_tags (Set[str]): Set of all tags used in multi-platform build steps
+            step_name (str): Name of the step
+            update_mp_push_tags (bool, optional): Whether to update the set of tags used in multi-platform steps.
+
+        Raises:
+            ValueError: If the config file is invalid
+        """
+        # Check for valid push section, duplicate mp tags are not allowed
+        if push is not None:
+            name = None
+            names = None
+            if isinstance(push, str):
+                name = push
+                if ":" not in name:
+                    name = f'{name}:latest'
+
+            if isinstance(push, StepPushDict):
+                names = [f"{push.repository}:{tag}" for tag in push.tags]
+
+            if names is not None:
+                for current_name in names:
+                    if current_name in mp_push_tags:
+                        # raise ValueError(f'Cannot specify duplicate tag {current_name} in build step {step_name}')
+                        raise ValueError(f'Cannot specify duplicate tag {current_name} in build step {step_name}')
+
+            if name is not None and name in mp_push_tags:
+                # raise ValueError(f'Cannot specify duplicate tag {name} in build step {step_name}')
+                raise ValueError(f'Cannot specify duplicate tag {name} in build step {step_name}')
+
+            if update_mp_push_tags and names is not None:
+                mp_push_tags.update(names)
+
+            if update_mp_push_tags and name is not None:
+                mp_push_tags.add(name)
+
+    def validate_multi_platform_build(self, mp_push_tags: Set[str]):
+        """
+        Validate multi-platform build steps
+
+        Args:
+            mp_push_tags (Set[str]): Set of all tags used in multi-platform build steps
+
+        Raises:
+            ValueError | pydantic.ValidationError: If the config file is invalid
+        """
+        # Iterate through each step
+        for step_name, step in self.steps.items():
+            if step.is_multi_platform():
+                if step.build.platform is not None:
+                    raise ValueError(f'Cannot specify both platform ({step.build.platform}) and '
+                                     f'platforms ({step.build.platforms}) in build step {step_name}')
+
+                if not isinstance(step.build.platforms, list):
+                    raise ValueError(f'platforms must be a list in build step {step_name}')
+
+                # Check for valid push section, duplicate mp tags are not allowed
+                self.validate_push(step.push, mp_push_tags, step_name)
+
+    def validate(self):
+        """
+        Validate the config file
+
+        Raises:
+            ValueError | pydantic.ValidationError : If the config file is invalid
+        """
+        if self.has_multi_platform_build():
+            mp_push_tags = set()
+            self.validate_multi_platform_build(mp_push_tags)
+
+            # Validate that all tags are unique across all multi-platform step
+            for step_name, step in self.steps.items():
+
+                # Check that there are no single platform tags that match multi-platform tags
+                if not step.is_multi_platform():
+                    if step.push is not None:
+                        self.validate_push(push=step.push,
+                                           mp_push_tags=mp_push_tags,
+                                           step_name=step_name,
+                                           update_mp_push_tags=False)

requirements.in

@@ -11,3 +11,5 @@ vcsinfo>=2.1.105
 graphlib-backport>=1.0.3
 timeout-decorator>=0.5.0
 python-on-whales>=0.61.0
+# python-on-whales requires pydantic 1.10.11 08/2023
+pydantic>=1.10.11

requirements.txt

@@ -73,7 +73,9 @@ pkginfo==1.9.6
 pycparser==2.21
     # via cffi
 pydantic==1.10.11
-    # via python-on-whales
+    # via
+    #   -r requirements.in
+    #   python-on-whales
 pygments==2.15.1
     # via
     #   readme-renderer