Spaces:
Running
Running
File size: 7,964 Bytes
e50c979 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """The patch language: what the Medic is allowed to do, and what must be rejected."""
from __future__ import annotations
from daggrstudio.medic.patching import apply_ops, normalise_ops
from daggrstudio.registry.bricks import BrickRegistry, get_registry
from daggrstudio.spec import Binding
from tests.conftest import concept_spec
def _apply(spec, *ops, registry: BrickRegistry | None = None):
return apply_ops(spec, list(ops), registry or get_registry())
def test_normalise_ops_accepts_every_shape_the_model_might_return():
assert normalise_ops({"ops": [{"op": "set_meta", "notes": "x"}]}) == [{"op": "set_meta", "notes": "x"}]
assert normalise_ops({"patches": [{"op": "drop_input"}]}) == [{"op": "drop_input"}]
assert normalise_ops([{"op": "drop_input"}]) == [{"op": "drop_input"}]
assert normalise_ops({"op": "drop_input", "step": "a"}) == [{"op": "drop_input", "step": "a"}]
assert normalise_ops(None) == []
assert normalise_ops({"diagnosis": "nothing wrong"}) == []
assert normalise_ops("a string") == []
def test_set_input_wires_a_user_binding():
spec = concept_spec()
result = _apply(spec, {"op": "set_input", "step": "concept", "param": "prompt",
"from": "user:prompt"})
assert result.changed and not result.skipped
assert spec.steps[0].inputs["prompt"].kind == "user"
def test_set_input_accepts_a_literal_via_value_key():
spec = concept_spec()
_apply(spec, {"op": "set_input", "step": "concept", "param": "height", "value": 768})
assert spec.steps[0].inputs["height"].value == 768
def test_drop_input_removes_a_param_the_space_no_longer_takes():
spec = concept_spec()
result = _apply(spec, {"op": "drop_input", "step": "concept", "param": "num_inference_steps"})
assert result.changed
assert "num_inference_steps" not in spec.steps[0].inputs
def test_unknown_op_is_rejected_with_a_reason():
spec = concept_spec()
result = _apply(spec, {"op": "run_arbitrary_python", "code": "import os"})
assert not result.changed
assert "unknown op" in result.skipped[0]["reason"]
def test_edge_to_a_nonexistent_step_is_rejected():
spec = concept_spec()
result = _apply(spec, {"op": "set_input", "step": "cutout", "param": "image",
"from": "step:ghost.image"})
assert not result.changed
assert "unknown step 'ghost'" in result.skipped[0]["reason"]
def test_edge_to_a_port_that_does_not_exist_is_rejected():
spec = concept_spec()
result = _apply(spec, {"op": "set_input", "step": "cutout", "param": "image",
"from": "step:concept.mask"})
assert not result.changed
assert "no output port 'mask'" in result.skipped[0]["reason"]
def test_binding_to_an_undeclared_user_input_is_rejected():
spec = concept_spec()
result = _apply(spec, {"op": "set_input", "step": "concept", "param": "prompt",
"from": "user:not_declared"})
assert not result.changed
assert "not declared" in result.skipped[0]["reason"]
def test_replace_brick_only_accepts_verified_bricks():
spec = concept_spec()
bad = _apply(spec, {"op": "replace_brick", "step": "concept", "brick_id": "imaginary-space"})
assert not bad.changed
assert "not a verified brick" in bad.skipped[0]["reason"]
good = _apply(spec, {"op": "replace_brick", "step": "concept", "brick_id": "z-image-turbo"})
assert good.changed
assert spec.steps[0].brick_id == "z-image-turbo"
assert spec.steps[0].source == "hf-applications/Z-Image-Turbo"
def test_replace_brick_preserves_edges_and_drops_stale_literals():
spec = concept_spec()
_apply(spec, {"op": "replace_brick", "step": "cutout", "brick_id": "not-lain-bg-removal"})
# the edge coming from the upstream step must survive the swap
assert spec.steps[1].inputs["image"].step_ref == ("concept", "image")
def test_set_outputs_validates_components():
spec = concept_spec()
bad = _apply(spec, {"op": "set_outputs", "step": "concept",
"outputs": {"image": {"component": "hologram"}}})
assert not bad.changed and "unknown component" in bad.skipped[0]["reason"]
good = _apply(spec, {"op": "set_outputs", "step": "concept",
"outputs": {"image": {"component": "image", "label": "Art"}}})
assert good.changed
assert spec.steps[0].outputs["image"]["label"] == "Art"
def test_set_postprocess_rejects_invented_hints():
spec = concept_spec()
bad = _apply(spec, {"op": "set_postprocess", "step": "cutout", "hint": "take_the_last_one"})
assert not bad.changed
assert "unrecognised postprocess hint" in bad.skipped[0]["reason"]
good = _apply(spec, {"op": "set_postprocess", "step": "cutout", "hint": "tuple_index:1"})
assert good.changed and spec.steps[1].postprocess == "tuple_index:1"
def test_remove_step_refuses_to_break_edges():
spec = concept_spec()
result = _apply(spec, {"op": "remove_step", "step": "concept"})
assert not result.changed
assert "still referenced" in result.skipped[0]["reason"]
ok = _apply(spec, {"op": "remove_step", "step": "collect"})
assert ok.changed and len(spec.steps) == 2
def test_add_step_requires_a_verified_brick_and_a_fresh_id():
spec = concept_spec()
dup = _apply(spec, {"op": "add_step", "step": {"id": "concept", "kind": "fn", "fn": "json_report"}})
assert not dup.changed and "already exists" in dup.skipped[0]["reason"]
invented = _apply(spec, {"op": "add_step",
"step": {"id": "x", "kind": "space", "brick_id": "made-up"}})
assert not invented.changed and "not verified" in invented.skipped[0]["reason"]
good = _apply(spec, {"op": "add_step",
"step": {"id": "report", "kind": "fn", "fn": "json_report",
"inputs": {"payload": {"from": "step:cutout.image"}},
"outputs": {"report": {"component": "json"}}}})
assert good.changed and spec.step("report") is not None
def test_set_meta_enforces_field_types_and_posture_values():
spec = concept_spec()
bad = _apply(spec, {"op": "set_meta", "tags": "game-dev"})
assert not bad.changed and "must be list" in bad.skipped[0]["reason"]
bad2 = _apply(spec, {"op": "set_meta", "license_posture": "whatever"})
assert not bad2.changed
good = _apply(spec, {"op": "set_meta", "name": "Better name", "license_posture": "any",
"tags": ["art"]})
assert good.changed and spec.name == "Better name" and spec.license_posture == "any"
def test_declare_inputs_replaces_the_input_list_safely():
spec = concept_spec()
bad = _apply(spec, {"op": "declare_inputs",
"inputs": [{"port": "x", "component": "teleporter"}]})
assert not bad.changed and "unknown component" in bad.skipped[0]["reason"]
good = _apply(spec, {"op": "declare_inputs",
"inputs": [{"port": "prompt", "component": "textbox", "label": "Prompt"}]})
assert good.changed and [i.port for i in spec.inputs] == ["prompt"]
def test_a_batch_with_one_bad_op_still_applies_the_good_ones():
spec = concept_spec()
result = _apply(spec,
{"op": "set_meta", "name": "Mixed batch"},
{"op": "drop_input", "step": "ghost", "param": "x"},
{"op": "set_input", "step": "concept", "param": "height", "value": 512})
assert len(result.applied) == 2 and len(result.skipped) == 1
assert spec.name == "Mixed batch"
assert spec.steps[0].inputs["height"].value == 512
def test_spec_timestamp_moves_only_when_something_changed():
spec = concept_spec()
before = spec.updated_at
_apply(spec, {"op": "drop_input", "step": "ghost", "param": "nope"})
assert spec.updated_at == before
_apply(spec, {"op": "set_meta", "notes": "touched"})
assert spec.updated_at >= before |