fix: check resolved host requirement capability names at job creation

src/openjd/model/v2023_09/_model.py

@@ -3066,6 +3066,18 @@ def _check_constraints(self, value: Any) -> None:
 _STANDARD_AMOUNT_CAPABILITIES_NAMES = list(STANDARD_AMOUNT_CAPABILITIES.keys())
 
 
+_CAPABILITY_NAME_MAX_LENGTH = 100
+
+
+def _validate_resolved_capability_name_length(name: str) -> None:
+    """Apply the 100-character limit of §3.3.1.1 / §3.3.2.1 to a resolved
+    capability name. The limit applies after the name's format string has
+    been resolved, so a name that is a format string is checked at job
+    creation."""
+    if len(name) > _CAPABILITY_NAME_MAX_LENGTH:
+        raise ValueError(f"String must be at most {_CAPABILITY_NAME_MAX_LENGTH} characters long")
+
+
 class AmountCapabilityName(FormatString):
     """The name of an amount capability."""
 
@@ -3121,6 +3133,10 @@ class AmountRequirement(OpenJDModel_v2023_09):
     @field_validator("name")
     @classmethod
     def _validate_name(cls, v: str, info: ValidationInfo) -> str:
+        # The name is the resolved <AmountCapabilityName>, so its §3.3.1.1
+        # length limit applies here even when the template's name was a
+        # format string.
+        _validate_resolved_capability_name_length(v)
         validate_amount_capability_name(
             capability_name=v, standard_capabilities=_STANDARD_AMOUNT_CAPABILITIES_NAMES
         )
@@ -3274,6 +3290,10 @@ class AttributeRequirement(OpenJDModel_v2023_09):
     @field_validator("name")
     @classmethod
     def _validate_name(cls, v: str) -> str:
+        # The name is the resolved <AttributeCapabilityName>, so its §3.3.2.1
+        # length limit applies here even when the template's name was a
+        # format string.
+        _validate_resolved_capability_name_length(v)
         validate_attribute_capability_name(
             capability_name=v, standard_capabilities=_STANDARD_ATTRIBUTE_CAPABILITIES_NAMES
         )
@@ -3410,6 +3430,28 @@ class HostRequirements(OpenJDModel_v2023_09):
     amounts: Optional[list[AmountRequirement]] = None
     attributes: Optional[list[AttributeRequirement]] = None
 
+    # §3.3: no two amounts, and no two attributes, may have the same name
+    # after the name format strings have been resolved. The template only
+    # compares the raw names, so two different format strings that resolve
+    # to the same capability are caught here.
+    @field_validator("amounts")
+    @classmethod
+    def _validate_amounts(
+        cls, v: Optional[list[AmountRequirement]]
+    ) -> Optional[list[AmountRequirement]]:
+        if v is None:
+            return v
+        return validate_unique_elements(v, item_value=lambda v: v.name.lower(), property="name")
+
+    @field_validator("attributes")
+    @classmethod
+    def _validate_attributes(
+        cls, v: Optional[list[AttributeRequirement]]
+    ) -> Optional[list[AttributeRequirement]]:
+        if v is None:
+            return v
+        return validate_unique_elements(v, item_value=lambda v: v.name.lower(), property="name")
+
 
 class HostRequirementsTemplate(OpenJDModel_v2023_09):
     amounts: Optional[list[AmountRequirementTemplate]] = None

test/openjd/model_v0/test_create_job.py

@@ -1068,3 +1068,127 @@ def test_queue_environment_parameters_reach_the_tables(self) -> None:
             "string",
             "fromQueueEnv",
         )
+
+
+class TestCreateJobResolvedCapabilityNames:
+    """The §3.3.1.1 / §3.3.2.1 capability name constraints and the §3.3
+    uniqueness constraints apply to the names after their format strings have
+    been resolved, so job creation checks the resolved names."""
+
+    @staticmethod
+    def _create_job(host_requirements: dict[str, Any], a: str, b: str) -> None:
+        job_template = decode_job_template(
+            template={
+                "specificationVersion": "jobtemplate-2023-09",
+                "name": "Job",
+                "parameterDefinitions": [
+                    {"name": "A", "type": "STRING"},
+                    {"name": "B", "type": "STRING"},
+                ],
+                "steps": [
+                    {
+                        "name": "Step",
+                        "hostRequirements": host_requirements,
+                        "script": {"actions": {"onRun": {"command": "do something"}}},
+                    }
+                ],
+            },
+        )
+        parameter_values = {
+            "A": ParameterValue(type=ParameterValueType.STRING, value=a),
+            "B": ParameterValue(type=ParameterValueType.STRING, value=b),
+        }
+        create_job(job_template=job_template, job_parameter_values=parameter_values)
+
+    @pytest.mark.parametrize(
+        "host_requirements,a,b,expected_message",
+        [
+            pytest.param(
+                {
+                    "attributes": [
+                        {"name": "{{Param.A}}", "anyOf": ["v"]},
+                        {"name": "{{Param.B}}", "anyOf": ["v"]},
+                    ]
+                },
+                "attr.custom.x",
+                "ATTR.CUSTOM.X",
+                "1 validation errors for JobTemplate\n"
+                "steps[0] -> hostRequirements -> attributes:\n"
+                "\tDuplicate values for name are not allowed. Duplicate values: attr.custom.x",
+                id="attribute names resolve to the same name, case-insensitively",
+            ),
+            pytest.param(
+                {
+                    "amounts": [
+                        {"name": "{{Param.A}}", "min": 1},
+                        {"name": "{{Param.B}}", "min": 1},
+                    ]
+                },
+                "amount.custom.x",
+                "amount.custom.x",
+                "1 validation errors for JobTemplate\n"
+                "steps[0] -> hostRequirements -> amounts:\n"
+                "\tDuplicate values for name are not allowed. Duplicate values: amount.custom.x",
+                id="amount names resolve to the same name",
+            ),
+            pytest.param(
+                {"attributes": [{"name": "{{Param.A}}", "anyOf": ["v"]}]},
+                "attr.custom." + "a" * 89,
+                "unused",
+                "1 validation errors for JobTemplate\n"
+                "steps[0] -> hostRequirements -> attributes[0] -> name:\n"
+                "\tString must be at most 100 characters long",
+                id="attribute name resolves to 101 characters",
+            ),
+            pytest.param(
+                {"amounts": [{"name": "{{Param.A}}", "min": 1}]},
+                "amount.custom." + "a" * 87,
+                "unused",
+                "1 validation errors for JobTemplate\n"
+                "steps[0] -> hostRequirements -> amounts[0] -> name:\n"
+                "\tString must be at most 100 characters long",
+                id="amount name resolves to 101 characters",
+            ),
+        ],
+    )
+    def test_resolved_names_checked(
+        self, host_requirements: dict[str, Any], a: str, b: str, expected_message: str
+    ) -> None:
+        # WHEN
+        with pytest.raises(DecodeValidationError) as excinfo:
+            self._create_job(host_requirements, a, b)
+
+        # THEN
+        assert str(excinfo.value) == expected_message
+
+    @pytest.mark.parametrize(
+        "host_requirements,a,b",
+        [
+            pytest.param(
+                {
+                    "attributes": [
+                        {"name": "{{Param.A}}", "anyOf": ["v"]},
+                        {"name": "{{Param.B}}", "anyOf": ["v"]},
+                    ]
+                },
+                "attr.custom.x",
+                "attr.custom.y",
+                id="attribute names resolve to different names",
+            ),
+            pytest.param(
+                {"attributes": [{"name": "{{Param.A}}", "anyOf": ["v"]}]},
+                "attr.custom." + "a" * 88,
+                "unused",
+                id="attribute name resolves to exactly 100 characters",
+            ),
+            pytest.param(
+                {"amounts": [{"name": "{{Param.A}}", "min": 1}]},
+                "amount.custom." + "a" * 86,
+                "unused",
+                id="amount name resolves to exactly 100 characters",
+            ),
+        ],
+    )
+    def test_valid_resolved_names(self, host_requirements: dict[str, Any], a: str, b: str) -> None:
+        # WHEN / THEN (no error)
+        self._create_job(host_requirements, a, b)