"""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_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"