diff --git a/kernelci/config/job.py b/kernelci/config/job.py index 60d82a2349..f4571c5b1b 100644 --- a/kernelci/config/job.py +++ b/kernelci/config/job.py @@ -5,104 +5,131 @@ """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) + + +_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", - 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._image = image - self._kcidb_test_suite = kcidb_test_suite - self._priority = priority - self._params = ( - self.format_params(params.copy(), 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._config = self._config.model_copy( + update={"params": formatted_params} ) - self._rules = rules + + 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 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 dict(self._params) - - @property - def rules(self): - """Kernel requirements (tree, branch, version...)""" - return self._rules + """Return isolated parameters passed to the template.""" + return copy.deepcopy(self._config.params) - @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", - "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.load_from_yaml(config, name=name) - for name, config in data.get("jobs", {}).items() + name: Job(name=name, config=config) + 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..44ac01b758 100644 --- a/tests/test_configs.py +++ b/tests/test_configs.py @@ -8,10 +8,15 @@ """Unit test for KernelCI YAML config handling""" +import copy + +import pytest import yaml +from pydantic import ValidationError import kernelci.config import kernelci.config.build +from kernelci.config.job import Job, JobConfig # ----------------------------------------------------------------------------- # Legacy @@ -169,6 +174,197 @@ 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 = { + "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"""