ledgrab/server/tests/api/test_graph_duplicate.py

"""Integration tests for server-side subgraph duplication.

Exercises ``_duplicate_subgraph`` against the *real* value-source and
colour-strip stores (temp DB), asserting that references *within* the selection
are remapped to the clones while references to entities *outside* the selection
stay shared with the originals — and that the deep-copy clone preserves config.
"""

import pytest

from ledgrab.api import dependencies as deps
from ledgrab.api.graph_schema import serialize_entity
from ledgrab.api.routes.graph import _duplicate_subgraph
from ledgrab.storage.color_strip_store import ColorStripStore
from ledgrab.storage.database import Database
from ledgrab.storage.value_source_store import ValueSourceStore


@pytest.fixture
def stores(tmp_path, monkeypatch):
    db = Database(tmp_path / "dup.db")
    css = ColorStripStore(db)
    vss = ValueSourceStore(db)
    monkeypatch.setattr(deps, "_deps", {"color_strip_store": css, "value_source_store": vss})
    yield css, vss
    db.close()


def _layer(sid: str) -> dict:
    return {"source_id": sid, "blend_mode": "normal", "opacity": 1.0, "enabled": True}


def test_duplicate_composite_remaps_only_in_selection_refs(stores):
    css, _vss = stores
    b = css.create_source(name="B", source_type="single_color", colors=[[255, 0, 0]])
    d = css.create_source(name="D", source_type="single_color", colors=[[0, 255, 0]])
    a = css.create_source(name="A", source_type="composite", layers=[_layer(b.id), _layer(d.id)])

    res = _duplicate_subgraph([a.id, b.id], " (copy)")

    assert set(res["id_map"]) == {a.id, b.id}
    assert res["skipped"] == []
    assert res["warnings"] == []

    a_new = serialize_entity(css.get(res["id_map"][a.id]))
    layer_ids = [ly["source_id"] for ly in a_new["layers"]]
    # B was in the selection -> remapped to the clone; D was not -> stays shared.
    assert layer_ids == [res["id_map"][b.id], d.id]
    assert a_new["name"] == "A (copy)"
    # Original is untouched.
    assert [ly["source_id"] for ly in serialize_entity(css.get(a.id))["layers"]] == [b.id, d.id]


def test_duplicate_preserves_clone_config(stores):
    """The deep-copy clone keeps every field except identity/name/timestamps —
    guarding against the storage-vs-create-schema name mismatch (e.g. single
    color's ``colors``) that a create-schema round-trip would silently drop."""
    css, _vss = stores
    src = css.create_source(name="Solid", source_type="single_color", colors=[[12, 34, 56]])
    res = _duplicate_subgraph([src.id], " (copy)")
    orig = serialize_entity(css.get(src.id))
    clone = serialize_entity(css.get(res["id_map"][src.id]))
    ignore = {"id", "name", "created_at", "updated_at"}
    assert {k: v for k, v in clone.items() if k not in ignore} == {
        k: v for k, v in orig.items() if k not in ignore
    }


def test_duplicate_partial_selection_keeps_layer_refs_shared(stores):
    css, _vss = stores
    b = css.create_source(name="B", source_type="single_color", colors=[[255, 0, 0]])
    a = css.create_source(name="A", source_type="composite", layers=[_layer(b.id)])

    res = _duplicate_subgraph([a.id], " (copy)")  # only the composite, not its layer

    a_new = serialize_entity(css.get(res["id_map"][a.id]))
    assert [ly["source_id"] for ly in a_new["layers"]] == [b.id]  # shared with original


def test_duplicate_remaps_bindable_slot_to_cloned_value_source(stores):
    """Pass-2's dict round-trip path: a colour-strip bindable slot bound to a
    value source that is *also* in the selection must point at the value clone
    (not the original) after duplication."""
    css, vss = stores
    vs = vss.create_source(name="Pulse", source_type="static", value=0.5)
    c = css.create_source(name="Candle", source_type="candlelight")
    css.update_source(c.id, intensity={"source_id": vs.id})  # bind intensity -> vs
    assert serialize_entity(css.get(c.id))["intensity"]["source_id"] == vs.id  # binding took

    res = _duplicate_subgraph([c.id, vs.id], " (copy)")

    assert res["warnings"] == []
    c_new = serialize_entity(css.get(res["id_map"][c.id]))
    assert c_new["intensity"]["source_id"] == res["id_map"][vs.id]


def test_duplicate_remaps_layer_brightness_source(stores):
    """A composite layer's value-source brightness binding (list + value ref) is
    remapped when that value source is also in the selection."""
    css, vss = stores
    vs = vss.create_source(name="Dim", source_type="static", value=0.3)
    leaf = css.create_source(name="Leaf", source_type="single_color", colors=[[9, 9, 9]])
    comp = css.create_source(
        name="Comp",
        source_type="composite",
        layers=[{**_layer(leaf.id), "brightness_source_id": vs.id}],
    )
    res = _duplicate_subgraph([comp.id, leaf.id, vs.id], " (copy)")
    layer = serialize_entity(css.get(res["id_map"][comp.id]))["layers"][0]
    assert layer["source_id"] == res["id_map"][leaf.id]
    assert layer["brightness_source_id"] == res["id_map"][vs.id]


def test_duplicate_safety_net_flags_unremapped_ref(stores, monkeypatch):
    """If a reference somehow isn't remapped, the post-clone safety net reports
    it in `warnings` rather than silently leaving two pipelines sharing a node."""
    css, _vss = stores
    b = css.create_source(name="B", source_type="single_color", colors=[[1, 2, 3]])
    a = css.create_source(name="A", source_type="composite", layers=[_layer(b.id)])
    # Force remap to a no-op so the clone keeps pointing at the original in-set id.
    monkeypatch.setattr("ledgrab.api.routes.graph.remap_refs", lambda *a, **k: 0)
    res = _duplicate_subgraph([a.id, b.id], " (copy)")
    assert any(w["id"] == res["id_map"][a.id] for w in res["warnings"])


def test_duplicate_value_source(stores):
    _css, vss = stores
    v = vss.create_source(name="V", source_type="static", value=0.5)
    res = _duplicate_subgraph([v.id], " (copy)")
    assert list(res["id_map"]) == [v.id]
    new = vss.get(res["id_map"][v.id])
    assert new.id != v.id
    assert new.name == "V (copy)"
    assert getattr(new, "value", None) == 0.5


def test_duplicate_skips_non_duplicable_ids(stores):
    _css, vss = stores
    v = vss.create_source(name="V", source_type="static", value=0.5)
    res = _duplicate_subgraph([v.id, "dev_external", "bogus"], " (copy)")
    assert list(res["id_map"]) == [v.id]
    assert {s["id"] for s in res["skipped"]} == {"dev_external", "bogus"}


def test_duplicate_name_collision_is_suffixed(stores):
    _css, vss = stores
    v = vss.create_source(name="V", source_type="static", value=0.5)
    vss.create_source(name="V (copy)", source_type="static", value=0.1)  # occupy the obvious name
    res = _duplicate_subgraph([v.id], " (copy)")
    new = vss.get(res["id_map"][v.id])
    assert new.name == "V (copy) 2"


def test_clone_allowlist_invariant():
    """Only explicitly-flagged (secret-free) stores are cloneable; the base
    default is off so a new store is never cloneable by accident."""
    from ledgrab.storage.base_sqlite_store import BaseSqliteStore

    assert BaseSqliteStore._cloneable is False
    assert ColorStripStore._cloneable is True
    assert ValueSourceStore._cloneable is True


def test_clone_refuses_non_cloneable_store(stores, monkeypatch):
    """clone() refuses stores not on the allowlist (defence-in-depth even though
    the duplicate endpoint already restricts to the safe kinds)."""
    css, _vss = stores
    src = css.create_source(name="X", source_type="single_color", colors=[[1, 1, 1]])
    monkeypatch.setattr(type(css), "_cloneable", False)
    with pytest.raises(NotImplementedError):
        css.clone(src.id, "X (copy)")