fix: validate schedule update bounds

This commit is contained in:
lda
2026-09-09 21:31:46 +07:00 Verified
parent a7018215b7
commit 479d8e4669
2 changed files with 36 additions and 3 deletions
+3 -3
View File
@@ -498,9 +498,9 @@ class UpdateScheduleParams(RpcParamsModel):
input_bindings: list[dict[str, Any]] | None = None input_bindings: list[dict[str, Any]] | None = None
overlap: str | None = None overlap: str | None = None
misfire: str | None = None misfire: str | None = None
max_active_runs: int | None = None max_active_runs: int | None = Field(default=None, ge=1)
lateness_allowance_s: float | None = None lateness_allowance_s: float | None = Field(default=None, ge=0)
max_steps: int | None = None max_steps: int | None = Field(default=None, ge=1, strict=True)
enabled: bool | None = None enabled: bool | None = None
@@ -12,6 +12,7 @@ from wf_transport_rpc_http.models import (
SetWorkflowOutputBindingsParams, SetWorkflowOutputBindingsParams,
StartRunParams, StartRunParams,
TraceRangeParams, TraceRangeParams,
UpdateScheduleParams,
) )
@@ -61,6 +62,38 @@ def test_run_params_are_explicit_models() -> None:
assert trace.trace_range.limit == 1 assert trace.trace_range.limit == 1
def test_update_schedule_params_match_create_numeric_constraints() -> None:
valid = UpdateScheduleParams(
schedule_id="schedule",
expected_revision=1,
max_active_runs=2,
lateness_allowance_s=0,
max_steps=10,
)
assert valid.max_active_runs == 2
assert valid.lateness_allowance_s == 0
assert valid.max_steps == 10
with pytest.raises(ValidationError):
UpdateScheduleParams(
schedule_id="schedule", expected_revision=1, max_active_runs=0
)
with pytest.raises(ValidationError):
UpdateScheduleParams(
schedule_id="schedule", expected_revision=1, lateness_allowance_s=-1
)
with pytest.raises(ValidationError):
UpdateScheduleParams(schedule_id="schedule", expected_revision=1, max_steps=0)
with pytest.raises(ValidationError):
UpdateScheduleParams(
schedule_id="schedule", expected_revision=1, max_steps=True
)
with pytest.raises(ValidationError):
UpdateScheduleParams(
schedule_id="schedule", expected_revision=1, max_steps="10"
)
def test_step_input_params_accept_composite_bindings_but_workflow_output_does_not() -> ( def test_step_input_params_accept_composite_bindings_but_workflow_output_does_not() -> (
None None
): ):