ledgrab/server/tests/api/test_graph_wiring_contract.py

"""Contract tests for graph drag-to-wire writes.

The graph editor's ``updateConnection()`` performs a *partial* PUT for a single
dragged edge: it sends only the reference (or bindable) field being wired. Five
entity kinds have ``Body(discriminator=...)`` PUT routes, so such a partial body
is rejected with a 422 unless it also echoes the entity's subtype
(``source_type`` / ``stream_type`` / ``target_type``). ``updateConnection()`` now
reads the target's subtype back and includes it.

These tests lock in that contract from the backend side: each ``(kind, field)``
pair the frontend ``CONNECTION_MAP`` drag-edits must validate as a minimal
``{discriminator, field}`` body — and must be rejected without the discriminator
(the exact failure that silently broke wiring). If a future schema change makes
one of these fields required-with-siblings, or drops a subtype, these tests fail
and flag that graph wiring will break.
"""

import pytest
from pydantic import TypeAdapter, ValidationError

from ledgrab.api.schemas.audio_sources import AudioSourceUpdate
from ledgrab.api.schemas.color_strip_sources import ColorStripSourceUpdate
from ledgrab.api.schemas.output_targets import OutputTargetUpdate
from ledgrab.api.schemas.picture_sources import PictureSourceUpdate
from ledgrab.api.schemas.value_sources import ValueSourceUpdate

# Each row mirrors one drag-editable (target_kind, field) pair from the frontend
# CONNECTION_MAP, paired with a real subtype tag that owns that field and the
# body shape updateConnection() sends: a flat ref id, or a bindable
# ``{parent: {source_id}}`` slot (here keyed by the parent field root).
# (kind, update_union, discriminator_field, subtype_tag, body_field, sample_value)
_WIRING_WRITES = [
    ("value_source", ValueSourceUpdate, "source_type", "audio", "audio_source_id", "as_1"),
    (
        "value_source",
        ValueSourceUpdate,
        "source_type",
        "adaptive_scene",
        "picture_source_id",
        "ps_1",
    ),
    ("value_source", ValueSourceUpdate, "source_type", "gradient_map", "value_source_id", "vs_1"),
    (
        "value_source",
        ValueSourceUpdate,
        "source_type",
        "css_extract",
        "color_strip_source_id",
        "css_1",
    ),
    (
        "color_strip_source",
        ColorStripSourceUpdate,
        "source_type",
        "picture",
        "picture_source_id",
        "ps_1",
    ),
    (
        "color_strip_source",
        ColorStripSourceUpdate,
        "source_type",
        "audio",
        "audio_source_id",
        "as_1",
    ),
    (
        "color_strip_source",
        ColorStripSourceUpdate,
        "source_type",
        "processed",
        "input_source_id",
        "css_2",
    ),
    (
        "color_strip_source",
        ColorStripSourceUpdate,
        "source_type",
        "processed",
        "processing_template_id",
        "cspt_1",
    ),
    # bindable BindableFloat slot — body is {<parent>: {source_id}}
    (
        "color_strip_source",
        ColorStripSourceUpdate,
        "source_type",
        "audio",
        "smoothing",
        {"source_id": "vs_1"},
    ),
    ("audio_source", AudioSourceUpdate, "source_type", "capture", "audio_template_id", "at_1"),
    ("audio_source", AudioSourceUpdate, "source_type", "processed", "audio_source_id", "as_2"),
    ("picture_source", PictureSourceUpdate, "stream_type", "raw", "capture_template_id", "ct_1"),
    ("picture_source", PictureSourceUpdate, "stream_type", "processed", "source_stream_id", "ps_2"),
    (
        "picture_source",
        PictureSourceUpdate,
        "stream_type",
        "processed",
        "postprocessing_template_id",
        "ppt_1",
    ),
    ("output_target", OutputTargetUpdate, "target_type", "led", "device_id", "dev_1"),
    ("output_target", OutputTargetUpdate, "target_type", "led", "color_strip_source_id", "css_1"),
    # bindable slots on output targets
    (
        "output_target",
        OutputTargetUpdate,
        "target_type",
        "led",
        "brightness",
        {"source_id": "vs_1"},
    ),
    (
        "output_target",
        OutputTargetUpdate,
        "target_type",
        "ha_light",
        "transition",
        {"source_id": "vs_1"},
    ),
]

_IDS = [f"{kind}.{field}[{tag}]" for kind, _u, _d, tag, field, _v in _WIRING_WRITES]


@pytest.mark.parametrize("kind,union,disc,tag,field,value", _WIRING_WRITES, ids=_IDS)
def test_partial_wiring_put_validates_with_discriminator(kind, union, disc, tag, field, value):
    """A single dragged wiring edit validates when the subtype is echoed.

    This is exactly what updateConnection() now sends; if it fails, drag-to-wire
    for ``kind.field`` is broken.
    """
    model = TypeAdapter(union).validate_python({disc: tag, field: value})
    assert getattr(model, disc) == tag


@pytest.mark.parametrize("kind,union,disc,tag,field,value", _WIRING_WRITES, ids=_IDS)
def test_partial_wiring_put_rejected_without_discriminator(kind, union, disc, tag, field, value):
    """Without the discriminator the same body is rejected.

    This is the 422 that silently broke drag-to-wire before updateConnection()
    echoed the subtype — guarding against anyone "simplifying" that away.
    """
    with pytest.raises(ValidationError):
        TypeAdapter(union).validate_python({field: value})


@pytest.mark.parametrize("kind,union,disc,tag,field,value", _WIRING_WRITES, ids=_IDS)
def test_partial_wiring_detach_validates_with_discriminator(kind, union, disc, tag, field, value):
    """Detach (clearing a slot) goes through the same partial-PUT path and must
    also validate with the subtype echoed: ``{field: ""}`` for a flat reference,
    ``{parent: {source_id: ""}}`` for a bindable slot — exactly what
    ``detachConnection()`` -> ``updateConnection(..., "")`` sends.
    """
    cleared = {"source_id": ""} if isinstance(value, dict) else ""
    model = TypeAdapter(union).validate_python({disc: tag, field: cleared})
    assert getattr(model, disc) == tag