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

@@ -1,56 +1,150 @@
-from re import L, T
+"""
+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 Optional
 from pydantic import BaseModel
 
 
 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 StepPush(BaseModel):
+
+class StepPushDict(BaseModel):
+    """ Push model within a step """
     repository: str
     tags: list[str]
 
+
 class Step(BaseModel):
+    """ Step model """
     build: Optional[StepBuild]
-    push: Optional[StepPush | list[str | StepPush]]
+    push: Optional[StepPushDict | list[str | StepPushDict] | str]
+
+    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: dict[str, Step]
 
     def __init__(self, **data) -> None:
-      super().__init__(**data)
-      self.validate()
+        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
+        """
         for step in self.steps.values():
-            if step.build is not None and \
-                    step.build.platforms is not None and \
-                    isinstance(step.build.platforms, list):
+            if step.is_multi_platform():
                 return True
         return False
 
-    def validate_multi_platform_build(self):
-        push_tags = {}
-        print(self.steps)
+    def validate_push(self,
+                      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():
-            print(step)
-            if step.build is not None and \
-                    step.build.platforms is not None:
+            if step.is_multi_platform():
                 if step.build.platform is not None:
-                    raise ValueError(f'Cannot specify both platform ({step.build.platform}) and platforms ({step.build.platforms}) in build step {step_name}')
-                # check for push tags
-                # step.push
-                return True
-        return True
+                    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):
-        is_valid = True
+        """
+        Validate the config file
+
+        Raises:
+            ValueError | pydantic.ValidationError : If the config file is invalid
+        """
         if self.has_multi_platform_build():
-            is_valid = self.validate_multi_platform_build() and is_valid
+            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():
 
-        return is_valid
+                # 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