add image tag vaildation

kpops/component_handlers/kafka_connect/model.py

@@ -13,6 +13,7 @@
 from pydantic.json_schema import SkipJsonSchema
 from typing_extensions import override
 
+from kpops.component_handlers.kubernetes.utils import validate_image_tag
 from kpops.components.base_components.helm_app import HelmAppValues
 from kpops.components.base_components.models.topic import KafkaTopic, KafkaTopicStr
 from kpops.utils.pydantic import (
@@ -125,3 +126,8 @@ class KafkaConnectorResetterValues(HelmAppValues):
     connector_type: Literal["source", "sink"]
     config: KafkaConnectorResetterConfig
     image_tag: str = Field(default="latest")
+
+    @pydantic.field_validator("image_tag", mode="before")
+    @classmethod
+    def validate_image_tag_field(cls, image_tag: Any) -> str:
+        return validate_image_tag(image_tag)

kpops/component_handlers/kubernetes/utils.py

@@ -1,5 +1,8 @@
 import hashlib
 import logging
+import re
+
+from kpops.api.exception import ValidationError
 
 log = logging.getLogger("K8sUtils")
 
@@ -26,3 +29,27 @@ def trim(max_len: int, name: str, suffix: str) -> str:
         )
         return new_name
     return name
+
+
+def validate_image_tag(image_tag: str) -> str:
+    """Validate an image tag.
+
+    Image tags consist of lowercase and uppercase letters, digits, underscores (_), periods (.), and dashes (-).
+    It can be up to 128 characters long and must follow the regex pattern: [a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}
+
+    :param image_tag: Docker image tag to be validated.
+    :return: The validated image tag.
+    """
+    if isinstance(image_tag, str) and is_valid_image_tag(image_tag):
+        return image_tag
+    msg = (
+        "Image tag is not valid. "
+        "Image tags consist of lowercase and uppercase letters, digits, underscores (_), periods (.), and dashes (-). "
+        "It can be up to 128 characters long."
+    )
+    raise ValidationError(msg)
+
+
+def is_valid_image_tag(image_tag: str) -> bool:
+    pattern = r"^[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}$"
+    return bool(re.match(pattern, image_tag))

kpops/components/streams_bootstrap/__init__.py

@@ -1,11 +1,12 @@
 import logging
 from abc import ABC
-from typing import Self
+from typing import Any, Self
 
 import pydantic
 from pydantic import Field
 
 from kpops.component_handlers.helm_wrapper.model import HelmRepoConfig
+from kpops.component_handlers.kubernetes.utils import validate_image_tag
 from kpops.components.base_components.helm_app import HelmApp, HelmAppValues
 from kpops.utils.docstring import describe_attr
 
@@ -26,6 +27,11 @@ class StreamsBootstrapValues(HelmAppValues):
 
     image_tag: str = Field(default="latest")
 
+    @pydantic.field_validator("image_tag", mode="before")
+    @classmethod
+    def validate_image_tag_field(cls, image_tag: Any) -> str:
+        return validate_image_tag(image_tag)
+
 
 class StreamsBootstrap(HelmApp, ABC):
     """Base for components with a streams-bootstrap Helm chart.

tests/component_handlers/kubernetes/test_utils.py

@@ -0,0 +1,20 @@
+import pytest
+
+from kpops.api.exception import ValidationError
+from kpops.component_handlers.kubernetes.utils import validate_image_tag
+
+
+def test_is_valid_image_tag():
+    assert validate_image_tag("1.2.3") == "1.2.3"
+    assert validate_image_tag("123") == "123"
+    assert validate_image_tag("1_2_3") == "1_2_3"
+    assert validate_image_tag("1-2-3") == "1-2-3"
+    assert validate_image_tag("latest") == "latest"
+    assert (
+        validate_image_tag(
+            "1ff6c18fbef2045af6b9c16bf034cc421a29027b800e4f9b68ae9b1cb3e9ae07"
+        )
+        == "1ff6c18fbef2045af6b9c16bf034cc421a29027b800e4f9b68ae9b1cb3e9ae07"
+    )
+    with pytest.raises(ValidationError):
+        assert validate_image_tag("la!est") is False

tests/pipeline/resources/resetter_values/defaults.yaml

@@ -9,3 +9,5 @@ helm-app:
 kafka-sink-connector:
   app:
     "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector"
+  resetter_values:
+    imageTag: override-default-image-tag

tests/pipeline/test_generate.py

@@ -840,6 +840,10 @@ def test_substitution_in_inflated_component(self):
             enriched_pipeline[1]["_resetter"]["app"]["label"]
             == "inflated-connector-name"
         )
+        assert (
+            enriched_pipeline[1]["_resetter"]["app"]["imageTag"]
+            == "override-default-image-tag"
+        )
 
     def test_substitution_in_resetter(self):
         pipeline = kpops.generate(
@@ -857,3 +861,7 @@ def test_substitution_in_resetter(self):
         assert enriched_pipeline[0]["name"] == "es-sink-connector"
         assert enriched_pipeline[0]["_resetter"]["name"] == "es-sink-connector"
         assert enriched_pipeline[0]["_resetter"]["app"]["label"] == "es-sink-connector"
+        assert (
+            enriched_pipeline[1]["_resetter"]["app"]["imageTag"]
+            == "override-default-image-tag"
+        )