test(automations/schemas): lock definition + api validation gates

definition/ (29 tests): the envelope (defaults, extra=forbid, empty
plan/name rejection), Inputs schema-alias roundtrip (Python schema_ ↔
wire schema), PlanStep numeric bounds + addressing-field constraints,
Execution production defaults stability (10-min timeout, 2 retries,
exponential backoff, drop_if_running) + closed-set Literal gates,
Metadata's exceptional extra="allow" contract, and TriggerSpec type
requirement.

api/ (9 tests): AutomationCreate/Update cascade-validate into the
nested definition, reject unknown payload fields, enforce name length;
TriggerCreate exposes safe defaults (enabled=True, params={},
static_inputs={}) and rejects unknown TriggerType strings at the
boundary.

All pure unit, no DB.
This commit is contained in:
CREDO23 2026-05-28 19:03:42 +02:00
parent acbeb60a43
commit 822940b09e
11 changed files with 396 additions and 0 deletions

View file

@ -0,0 +1,82 @@
"""Lock the request-side automation API schemas — the public validation gate."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from app.automations.schemas.api.automation import AutomationCreate, AutomationUpdate
pytestmark = pytest.mark.unit
_VALID_DEFINITION = {
"name": "Test",
"plan": [{"step_id": "s1", "action": "agent_task"}],
}
def test_automation_create_accepts_valid_minimal_payload() -> None:
"""Happy path: just search_space_id, name, and a valid definition.
Triggers default to ``[]`` so users can attach them later."""
payload = AutomationCreate.model_validate(
{
"search_space_id": 1,
"name": "Daily digest",
"definition": _VALID_DEFINITION,
}
)
assert payload.name == "Daily digest"
assert payload.description is None
assert payload.triggers == []
def test_automation_create_cascades_validation_into_nested_definition() -> None:
"""A bad ``definition`` (e.g. empty plan) fails at the API boundary,
not at the DB layer. Locks the cascade so corrupt definitions can't
sneak through a misshapen wire payload."""
with pytest.raises(ValidationError):
AutomationCreate.model_validate(
{
"search_space_id": 1,
"name": "Bad",
"definition": {"name": "X", "plan": []}, # empty plan
}
)
def test_automation_create_rejects_unknown_top_level_field() -> None:
"""``extra='forbid'`` catches typos in API payloads at the boundary."""
with pytest.raises(ValidationError):
AutomationCreate.model_validate(
{
"search_space_id": 1,
"name": "X",
"definition": _VALID_DEFINITION,
"owner": "tg", # not allowed
}
)
def test_automation_create_rejects_empty_name() -> None:
"""Name is required and constrained to 1..200 chars."""
with pytest.raises(ValidationError):
AutomationCreate.model_validate(
{
"search_space_id": 1,
"name": "",
"definition": _VALID_DEFINITION,
}
)
def test_automation_update_accepts_partial_payload_with_no_fields() -> None:
"""All fields on ``AutomationUpdate`` are optional. An empty body is
a valid no-op update (the service layer decides what to do with it)."""
update = AutomationUpdate.model_validate({})
assert update.name is None
assert update.description is None
assert update.status is None
assert update.definition is None

View file

@ -0,0 +1,47 @@
"""Lock the request-side trigger API schemas."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from app.automations.persistence.enums.trigger_type import TriggerType
from app.automations.schemas.api.trigger import TriggerCreate, TriggerUpdate
pytestmark = pytest.mark.unit
def test_trigger_create_uses_safe_defaults_for_optional_fields() -> None:
"""Defaults: empty ``params`` and ``static_inputs``, ``enabled=True``.
These let callers create a trigger with just ``type`` + the params
the trigger requires."""
trigger = TriggerCreate(type=TriggerType.SCHEDULE) # type: ignore[arg-type]
assert trigger.type is TriggerType.SCHEDULE
assert trigger.params == {}
assert trigger.static_inputs == {}
assert trigger.enabled is True
def test_trigger_create_rejects_unknown_trigger_type_string() -> None:
"""``type`` is a ``TriggerType`` enum, so any string outside the
enum's known values fails validation at the boundary."""
with pytest.raises(ValidationError):
TriggerCreate.model_validate({"type": "webhook"}) # not in TriggerType
def test_trigger_create_rejects_unknown_field() -> None:
"""``extra='forbid'`` catches typos in trigger payloads."""
with pytest.raises(ValidationError):
TriggerCreate.model_validate(
{"type": "schedule", "param": {}} # typo: param vs params
)
def test_trigger_update_accepts_partial_payload_with_no_fields() -> None:
"""``TriggerUpdate`` is fully optional — empty body is valid (no-op)."""
update = TriggerUpdate()
assert update.enabled is None
assert update.params is None
assert update.static_inputs is None