From a08a051e6acd935d3c9869d0ac12dfa9c7642a6a Mon Sep 17 00:00:00 2001
From: Denys Fedoryshchenko <denys.f@collabora.com>
Date: Thu, 30 Jul 2026 11:18:52 +0300
Subject: [PATCH 1/2] config: validate pipeline job definitions

Pipeline job YAML is currently filtered through a manually maintained attribute set. Unknown fields are silently discarded and values are accepted without type validation, so typos and configuration drift can surface only when a job runs.

Add a strict Pydantic schema at the post-merge loading boundary. Require a template, validate parameter mappings and supported priority values, reject unknown fields, and explicitly preserve the base_name field already used by the live pipeline configuration. Keep the existing Job API as a compatibility adapter and deep-copy parameters to prevent configuration mutation.

Cover invalid fields, required values, symbolic and numeric priorities, base-name serialization, merge ordering, and nested parameter isolation. This is an incremental step toward an authoritative configuration schema while preserving current consumers.

Refs: #2648
Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>
---
 kernelci/config/job.py  |  50 +++++++++++++--
 tests/configs/jobs.yaml |   4 ++
 tests/test_configs.py   | 135 ++++++++++++++++++++++++++++++++++++++++
 3 files changed, 185 insertions(+), 4 deletions(-)

diff --git a/kernelci/config/job.py b/kernelci/config/job.py
index 60d82a2349..dbcbb1b199 100644
--- a/kernelci/config/job.py
+++ b/kernelci/config/job.py
@@ -5,8 +5,41 @@
 
 """KernelCI pipeline job configuration"""
 
+import copy
+from typing import Annotated, Any, Dict, Literal, Optional, Union
+
+from pydantic import BaseModel, ConfigDict, Field
+
 from .base import YAMLConfigObject
 
+JobPriority = Union[
+    Literal["low", "medium", "high"],
+    Annotated[int, Field(ge=0, le=100)],
+]
+
+
+class JobConfig(BaseModel):
+    """Validated YAML representation of a pipeline job."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)
+
+    template: str = Field(min_length=1)
+    kind: str = "node"
+    base_name: Optional[str] = None
+    image: Optional[str] = None
+    params: Dict[str, Any] = Field(default_factory=dict)
+    rules: Any = None
+    kcidb_test_suite: Any = None
+    priority: Optional[JobPriority] = None
+
+
+class JobsConfig(BaseModel):
+    """Validated top-level pipeline jobs configuration."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)
+
+    jobs: Dict[str, JobConfig] = Field(default_factory=dict)
+
 
 class Job(YAMLConfigObject):
     """Pipeline job definition"""
@@ -19,6 +52,7 @@ def __init__(
         template,
         *,
         kind="node",
+        base_name=None,
         image=None,
         params=None,
         rules=None,
@@ -28,11 +62,12 @@ def __init__(
         self._name = name
         self._template = template
         self._kind = kind
+        self._base_name = base_name
         self._image = image
         self._kcidb_test_suite = kcidb_test_suite
         self._priority = priority
         self._params = (
-            self.format_params(params.copy(), params) if params else {}
+            self.format_params(copy.deepcopy(params), params) if params else {}
         )
         self._rules = rules
 
@@ -51,6 +86,11 @@ def kind(self):
         """Job node kind"""
         return self._kind
 
+    @property
+    def base_name(self):
+        """Optional base job name used to group related job variants"""
+        return self._base_name
+
     @property
     def priority(self):
         """Job priority"""
@@ -69,7 +109,7 @@ def image(self, value):
     @property
     def params(self):
         """Arbitrary parameters passed to the template"""
-        return dict(self._params)
+        return copy.deepcopy(self._params)
 
     @property
     def rules(self):
@@ -88,6 +128,7 @@ def _get_yaml_attributes(cls):
             {
                 "template",
                 "kind",
+                "base_name",
                 "image",
                 "params",
                 "rules",
@@ -100,9 +141,10 @@ def _get_yaml_attributes(cls):
 
 def from_yaml(data, _):
     """Create the pipeline job definitions using data loaded from YAML"""
+    validated = JobsConfig.model_validate({"jobs": data.get("jobs", {})})
     jobs = {
-        name: Job.load_from_yaml(config, name=name)
-        for name, config in data.get("jobs", {}).items()
+        name: Job(name=name, **config.model_dump())
+        for name, config in validated.jobs.items()
     }
 
     return {
diff --git a/tests/configs/jobs.yaml b/tests/configs/jobs.yaml
index 9661f79b90..6fa850c8dd 100644
--- a/tests/configs/jobs.yaml
+++ b/tests/configs/jobs.yaml
@@ -3,6 +3,7 @@ jobs:
   kbuild-gcc-10-x86:
     template: 'kbuild.jinja2'
     kind: 'kbuild'
+    base_name:
     image: 'gcc-10:{arch}{fragments}'
     params:
       config: x86_64_defconfig
@@ -13,6 +14,7 @@ jobs:
   kunit: &kunit-job
     template: 'kunit.jinja2'
     kind: 'test'
+    base_name:
     image: 'gcc-10:x86-kunit-kernelci'
     params: {}
     rules:
@@ -21,6 +23,7 @@ jobs:
 
   kunit-x86_64:
     <<: *kunit-job
+    base_name:
     image: 'kernelci/staging-gcc-10:x86-kunit-qemu-kernelci'
     params:
       arch: x86_64
@@ -31,6 +34,7 @@ jobs:
   kver:
     template: 'kver.jinja2'
     kind: 'test'
+    base_name:
     image: 'kernelci'
     params: {}
     rules:
diff --git a/tests/test_configs.py b/tests/test_configs.py
index 545267a137..1cb4000aa9 100644
--- a/tests/test_configs.py
+++ b/tests/test_configs.py
@@ -8,7 +8,11 @@
 
 """Unit test for KernelCI YAML config handling"""
 
+import copy
+
+import pytest
 import yaml
+from pydantic import ValidationError
 
 import kernelci.config
 import kernelci.config.build
@@ -169,6 +173,137 @@ def test_jobs(self):
             == "kernelci/staging-gcc-10:x86-kunit-qemu-kernelci"
         )
 
+    def test_jobs_reject_unknown_fields(self):
+        """Reject misspelled or unsupported job fields."""
+        data = {
+            "jobs": {
+                "broken-job": {
+                    "template": "kbuild.jinja2",
+                    "temlate": "misspelled.jinja2",
+                }
+            }
+        }
+
+        with pytest.raises(ValidationError, match="temlate"):
+            kernelci.config.load_data(data)
+
+    def test_jobs_require_template(self):
+        """Require every job to select a template."""
+        data = {"jobs": {"broken-job": {"kind": "test"}}}
+
+        with pytest.raises(ValidationError, match="template"):
+            kernelci.config.load_data(data)
+
+    @pytest.mark.parametrize("priority", ["low", "medium", "high", 0, 50, 100])
+    def test_jobs_accept_supported_priorities(self, priority):
+        """Accept symbolic and numeric priorities used by pipeline jobs."""
+        data = {
+            "jobs": {
+                "example": {
+                    "template": "kbuild.jinja2",
+                    "priority": priority,
+                }
+            }
+        }
+
+        config = kernelci.config.load_data(data)
+
+        assert config["jobs"]["example"].priority == priority
+
+    def test_jobs_preserve_base_name(self):
+        """Preserve the base job name used by pipeline job variants."""
+        data = {
+            "jobs": {
+                "ltp-timers_qemu": {
+                    "template": "ltp.jinja2",
+                    "base_name": "ltp-timers",
+                }
+            }
+        }
+
+        config = kernelci.config.load_data(data)
+
+        job = config["jobs"]["ltp-timers_qemu"]
+        assert job.base_name == "ltp-timers"
+        assert yaml.safe_load(yaml.dump(job))["base_name"] == "ltp-timers"
+
+    @pytest.mark.parametrize(
+        ("field", "value"),
+        [
+            ("params", []),
+            ("priority", "50"),
+            ("priority", 101),
+        ],
+    )
+    def test_jobs_reject_invalid_field_values(self, field, value):
+        """Reject invalid job field types and values."""
+        data = {
+            "jobs": {
+                "broken-job": {
+                    "template": "kbuild.jinja2",
+                    field: value,
+                }
+            }
+        }
+
+        with pytest.raises(ValidationError, match=field):
+            kernelci.config.load_data(data)
+
+    def test_jobs_do_not_mutate_source_data(self):
+        """Formatting job parameters must not mutate loaded YAML data."""
+        data = {
+            "jobs": {
+                "example": {
+                    "template": "kbuild.jinja2",
+                    "params": {
+                        "arch": "arm64",
+                        "nested": {"artifact": "{arch}/kernel"},
+                    },
+                }
+            }
+        }
+        original = copy.deepcopy(data)
+
+        config = kernelci.config.load_data(data)
+
+        assert data == original
+        assert config["jobs"]["example"].params["nested"]["artifact"] == (
+            "arm64/kernel"
+        )
+        returned_params = config["jobs"]["example"].params
+        returned_params["nested"]["artifact"] = "modified"
+        assert config["jobs"]["example"].params["nested"]["artifact"] == (
+            "arm64/kernel"
+        )
+
+    def test_jobs_validate_after_config_merge(self, tmp_path):
+        """Allow partial overrides by validating after all files are merged."""
+        base = tmp_path / "base.yaml"
+        overlay = tmp_path / "overlay.yaml"
+        base.write_text(
+            """
+jobs:
+  example:
+    template: kbuild.jinja2
+    params:
+      arch: arm64
+""",
+            encoding="utf-8",
+        )
+        overlay.write_text(
+            """
+jobs:
+  example:
+    priority: 50
+""",
+            encoding="utf-8",
+        )
+
+        config = kernelci.config.load([str(base), str(overlay)])
+
+        assert config["jobs"]["example"].template == "kbuild.jinja2"
+        assert config["jobs"]["example"].priority == 50
+
 
 class TestAPIConfigs(ConfigTest):
     """Tests for configs related to the KernelCI API"""

From 9daa8d5f56dd8690cf8eacead0646a034dac695d Mon Sep 17 00:00:00 2001
From: Denys Fedoryshchenko <denys.f@collabora.com>
Date: Thu, 30 Jul 2026 11:29:48 +0300
Subject: [PATCH 2/2] config: make JobConfig authoritative

The initial schema validation still copied every job field into a parallel runtime object. Adding a field therefore required updating the model, Job constructor, private storage, properties, and YAML attribute allowlist, leaving the central maintenance problem from #2648 in place.

Store the validated JobConfig directly and delegate ordinary field access to it. Derive parameter formatting and YAML serialization from model_dump(), while retaining only the mapping-key name, defensive parameter copies, and mutable image override as wrapper behavior. Keep legacy direct Job construction working by validating its keyword values through the same schema.

Add regression coverage for schema-driven access and serialization, legacy construction, invalid extra fields, and isolated image overrides.

Refs: #2648
Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>
---
 kernelci/config/job.py | 133 ++++++++++++++++++-----------------------
 tests/test_configs.py  |  61 +++++++++++++++++++
 2 files changed, 120 insertions(+), 74 deletions(-)

diff --git a/kernelci/config/job.py b/kernelci/config/job.py
index dbcbb1b199..f4571c5b1b 100644
--- a/kernelci/config/job.py
+++ b/kernelci/config/job.py
@@ -41,109 +41,94 @@ class JobsConfig(BaseModel):
     jobs: Dict[str, JobConfig] = Field(default_factory=dict)
 
 
+_IMAGE_NOT_OVERRIDDEN = object()
+
+
 class Job(YAMLConfigObject):
-    """Pipeline job definition"""
+    """Pipeline job definition backed by a validated ``JobConfig``."""
 
     yaml_tag = "!Job"
 
-    def __init__(
-        self,
-        name,
-        template,
-        *,
-        kind="node",
-        base_name=None,
-        image=None,
-        params=None,
-        rules=None,
-        kcidb_test_suite=None,
-        priority=None,
-    ):
+    def __init__(self, name, config=None, **legacy_values):
+        """Create a named job from validated configuration.
+
+        ``legacy_values`` keeps the previous ``Job(name, template=..., ...)``
+        construction API working while ensuring the schema remains the single
+        source of truth for accepted fields.
+        """
+        if isinstance(config, JobConfig):
+            if legacy_values:
+                fields = ", ".join(sorted(legacy_values))
+                raise TypeError(
+                    f"Unexpected fields with validated JobConfig: {fields}"
+                )
+        else:
+            if config is not None:
+                legacy_values["template"] = config
+            config = JobConfig.model_validate(legacy_values)
+
         self._name = name
-        self._template = template
-        self._kind = kind
-        self._base_name = base_name
-        self._image = image
-        self._kcidb_test_suite = kcidb_test_suite
-        self._priority = priority
-        self._params = (
-            self.format_params(copy.deepcopy(params), params) if params else {}
+        self._config = config
+        self._image_override = _IMAGE_NOT_OVERRIDDEN
+        formatted_params = (
+            self.format_params(
+                copy.deepcopy(self._config.params), self._config.params
+            )
+            if self._config.params
+            else {}
         )
-        self._rules = rules
+        self._config = self._config.model_copy(
+            update={"params": formatted_params}
+        )
+
+    def __getattr__(self, name):
+        """Delegate configuration fields to the validated schema."""
+        if name.startswith("_"):
+            raise AttributeError(name)
+        config = self.__dict__.get("_config")
+        if config is None:
+            raise AttributeError(name)
+        return getattr(config, name)
 
     @property
     def name(self):
         """Job name"""
         return self._name
 
-    @property
-    def template(self):
-        """Template file name"""
-        return self._template
-
-    @property
-    def kind(self):
-        """Job node kind"""
-        return self._kind
-
-    @property
-    def base_name(self):
-        """Optional base job name used to group related job variants"""
-        return self._base_name
-
-    @property
-    def priority(self):
-        """Job priority"""
-        return self._priority
-
     @property
     def image(self):
-        """Runtime environment image name"""
-        return self._image
+        """Runtime image, including an optional runtime override."""
+        if self._image_override is _IMAGE_NOT_OVERRIDDEN:
+            return self._config.image
+        return self._image_override
 
     @image.setter
     def image(self, value):
-        """Set the runtime environment image name"""
-        self._image = value
+        """Override the runtime environment image name."""
+        self._image_override = value
 
     @property
     def params(self):
-        """Arbitrary parameters passed to the template"""
-        return copy.deepcopy(self._params)
+        """Return isolated parameters passed to the template."""
+        return copy.deepcopy(self._config.params)
 
-    @property
-    def rules(self):
-        """Kernel requirements (tree, branch, version...)"""
-        return self._rules
-
-    @property
-    def kcidb_test_suite(self):
-        """Mapping of KernelCI test to KCIDB test suite"""
-        return self._kcidb_test_suite
+    def _get_format_map(self):
+        """Derive formatting fields directly from the authoritative schema."""
+        return self._config.model_dump(exclude={"params", "rules"})
 
     @classmethod
-    def _get_yaml_attributes(cls):
-        attrs = super()._get_yaml_attributes()
-        attrs.update(
-            {
-                "template",
-                "kind",
-                "base_name",
-                "image",
-                "params",
-                "rules",
-                "kcidb_test_suite",
-                "priority",
-            }
-        )
-        return attrs
+    def to_yaml(cls, dumper, data):
+        """Serialize all schema fields without a separate attribute list."""
+        values = data._config.model_dump()
+        values["image"] = data.image
+        return dumper.represent_mapping("tag:yaml.org,2002:map", values)
 
 
 def from_yaml(data, _):
     """Create the pipeline job definitions using data loaded from YAML"""
     validated = JobsConfig.model_validate({"jobs": data.get("jobs", {})})
     jobs = {
-        name: Job(name=name, **config.model_dump())
+        name: Job(name=name, config=config)
         for name, config in validated.jobs.items()
     }
 
diff --git a/tests/test_configs.py b/tests/test_configs.py
index 1cb4000aa9..44ac01b758 100644
--- a/tests/test_configs.py
+++ b/tests/test_configs.py
@@ -16,6 +16,7 @@
 
 import kernelci.config
 import kernelci.config.build
+from kernelci.config.job import Job, JobConfig
 
 # -----------------------------------------------------------------------------
 # Legacy
@@ -173,6 +174,66 @@ def test_jobs(self):
             == "kernelci/staging-gcc-10:x86-kunit-qemu-kernelci"
         )
 
+    def test_job_schema_is_authoritative(self):
+        """Expose and serialize schema fields without per-field properties."""
+        schema = JobConfig(
+            template="kbuild.jinja2",
+            kind="kbuild",
+            base_name="kbuild-base",
+            priority="high",
+        )
+        job = Job("kbuild-example", config=schema)
+
+        serialized = yaml.safe_load(yaml.dump(job))
+
+        assert set(serialized) == set(JobConfig.model_fields)
+        assert serialized["base_name"] == job.base_name
+        assert serialized["priority"] == job.priority
+        delegated_fields = {
+            "template",
+            "kind",
+            "base_name",
+            "rules",
+            "kcidb_test_suite",
+            "priority",
+        }
+        assert delegated_fields.isdisjoint(Job.__dict__)
+
+    def test_job_legacy_constructor_uses_schema(self):
+        """Keep direct construction compatible while validating its fields."""
+        job = Job(
+            "legacy-job",
+            "kbuild.jinja2",
+            kind="kbuild",
+            priority="medium",
+        )
+
+        assert job.template == "kbuild.jinja2"
+        assert job.kind == "kbuild"
+        assert job.priority == "medium"
+        with pytest.raises(ValidationError, match="unknown"):
+            Job(
+                "invalid-job",
+                template="kbuild.jinja2",
+                unknown="value",
+            )
+
+    def test_job_image_override_is_runtime_state(self):
+        """Keep image overrides separate from immutable source configuration."""
+        schema = JobConfig(
+            template="kbuild.jinja2",
+            image="original:image",
+        )
+        job = Job("kbuild-example", config=schema)
+
+        job.image = "override:image"
+
+        assert schema.image == "original:image"
+        assert job.image == "override:image"
+        assert yaml.safe_load(yaml.dump(job))["image"] == "override:image"
+        job.image = None
+        assert job.image is None
+
     def test_jobs_reject_unknown_fields(self):
         """Reject misspelled or unsupported job fields."""
         data = {