206 lines
7.8 KiB
Python
206 lines
7.8 KiB
Python
"""HEL-809: import declarations, persistence and both library views agree."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from calliope.config import settings
|
|
from calliope.db import get_db
|
|
from tests.test_workflow_import_batch import _ui_graph
|
|
|
|
|
|
def graph(*classes):
|
|
nodes = {str(i): {"class_type": cls, "inputs": {}} for i, cls in enumerate(classes, 1)}
|
|
nodes["90"] = {
|
|
"class_type": "PrimitiveStringMultiline",
|
|
"inputs": {"value": "HEL809 isolated prompt"},
|
|
"_meta": {"title": "Prompt (Input:prompt)"},
|
|
}
|
|
for i in range(1, len(classes) + 1):
|
|
role = "mask" if classes[i - 1] == "LoadImageMask" else "image"
|
|
nodes[str(i)]["_meta"] = {"title": f"Reference (Input:{role})"}
|
|
return nodes
|
|
|
|
|
|
def create(client, workflow, schema=None):
|
|
analyzed = client.post("/api/workflows/analyze", json={"workflow_json": workflow})
|
|
assert analyzed.status_code == 200
|
|
response = client.post(
|
|
"/api/workflows",
|
|
json={
|
|
"name": "HEL809 input contract",
|
|
"workflow_json": workflow,
|
|
"input_schema": analyzed.json()["inputs"] if schema is None else schema,
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def assert_views(client, wid, expected):
|
|
detail = client.get(f"/api/workflows/{wid}").json()
|
|
assert detail["media_io"] == expected
|
|
for query in ("", "?view=meta"):
|
|
rows = client.get("/api/workflows" + query).json()
|
|
row = next(r for r in rows if r["id"] == wid)
|
|
assert row["media_io"] == expected
|
|
# This is the library's exact media predicate, fed by the meta API.
|
|
for media in ("image", "video", "audio"):
|
|
hits = [
|
|
r["id"]
|
|
for r in rows
|
|
if media in r["media_io"]["in"] or media in r["media_io"]["out"]
|
|
]
|
|
assert (wid in hits) == (media in expected["in"] or media in expected["out"])
|
|
if query:
|
|
assert not {"workflow_json", "input_schema", "output_schema"} & row.keys()
|
|
|
|
|
|
def stored_media(wid):
|
|
with get_db(settings.db_path) as conn:
|
|
return json.loads(
|
|
conn.execute("SELECT media_io FROM workflows WHERE id = ?", (wid,)).fetchone()[
|
|
"media_io"
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"classes", [("LoadImage",), ("LoadImageMask",), ("LoadImage", "LoadImageMask"), ()]
|
|
)
|
|
def test_analyze_create_persists_media_and_agrees_with_library(client, classes):
|
|
row = create(client, graph(*classes))
|
|
expected = {"in": ["image"] if classes else [], "out": []}
|
|
if "LoadImageMask" in classes:
|
|
assert any(e["role"] == "mask" and e["kind"] == "image" for e in row["input_schema"])
|
|
assert stored_media(row["id"]) == expected
|
|
assert row["model_family"] is None # never infer a model or output from kind
|
|
assert_views(client, row["id"], expected)
|
|
|
|
|
|
def test_create_uses_confirmed_schema_not_original_graph(client):
|
|
row = create(client, graph(), [{"nodeId": "90", "kind": "audio_list", "manual": True}])
|
|
assert stored_media(row["id"]) == {"in": ["audio"], "out": []}
|
|
assert_views(client, row["id"], {"in": ["audio"], "out": []})
|
|
|
|
|
|
def test_update_changes_inputs_preserving_explicit_outputs_and_other_metadata(client):
|
|
row = create(client, graph("LoadImage"))
|
|
with get_db(settings.db_path) as conn:
|
|
conn.execute(
|
|
"UPDATE workflows SET media_io = ?, model_family = ?, tags = ? WHERE id = ?",
|
|
(json.dumps({"in": ["image"], "out": []}), "existing", '["tag"]', row["id"]),
|
|
)
|
|
for kind, expected in (("audio_list", ["audio"]), ("text", []), ("video_list", ["video"])):
|
|
response = client.patch(
|
|
f"/api/workflows/{row['id']}",
|
|
json={
|
|
"input_schema": [{"nodeId": "90", "kind": kind, "manual": True}],
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["model_family"] == "existing"
|
|
assert response.json()["tags"] == ["tag"]
|
|
assert stored_media(row["id"]) == {"in": expected, "out": []}
|
|
assert_views(client, row["id"], {"in": expected, "out": []})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
(None, {"in": ["image"], "out": ["image"]}),
|
|
('{"in": [], "out": []}', {"in": [], "out": []}),
|
|
('{"out": []}', {"in": ["image"], "out": []}),
|
|
('{"in": []}', {"in": [], "out": ["image"]}),
|
|
],
|
|
)
|
|
def test_legacy_null_and_missing_keys_are_distinct_from_empty_lists(client, raw, expected):
|
|
workflow = graph("LoadImageMask")
|
|
workflow["99"] = {"class_type": "SaveImage", "inputs": {"images": ["1", 0]}}
|
|
row = create(client, workflow)
|
|
with get_db(settings.db_path) as conn:
|
|
conn.execute("UPDATE workflows SET media_io = ? WHERE id = ?", (raw, row["id"]))
|
|
assert_views(client, row["id"], expected)
|
|
with get_db(settings.db_path) as conn:
|
|
assert (
|
|
conn.execute("SELECT media_io FROM workflows WHERE id = ?", (row["id"],)).fetchone()[
|
|
"media_io"
|
|
]
|
|
== raw
|
|
) # read only fallback
|
|
if raw is None:
|
|
response = client.patch(f"/api/workflows/{row['id']}", json={"favorite": True})
|
|
assert response.status_code == 200
|
|
assert stored_media(row["id"]) == expected
|
|
|
|
|
|
def test_legacy_prompt_only_has_no_invented_media(client):
|
|
row = create(client, graph())
|
|
with get_db(settings.db_path) as conn:
|
|
conn.execute("UPDATE workflows SET media_io = NULL WHERE id = ?", (row["id"],))
|
|
assert_views(client, row["id"], {"in": [], "out": []})
|
|
|
|
|
|
def test_batch_import_metadata_survives_unrelated_update(client):
|
|
response = client.post(
|
|
"/api/workflows/import-batch",
|
|
json={
|
|
"files": [
|
|
{
|
|
"source_path": "3-图像生成/Qwen-Image-2.1/demo.json",
|
|
"workflow_json": _ui_graph(),
|
|
"tags": ["existing batch"],
|
|
}
|
|
]
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
wid = response.json()["imported_ids"][0]
|
|
before = client.get(f"/api/workflows/{wid}").json()
|
|
response = client.patch(f"/api/workflows/{wid}", json={"favorite": True})
|
|
after = response.json()
|
|
for key in ("media_io", "source", "source_path", "category", "model_family", "tags"):
|
|
assert before[key] == after[key]
|
|
assert_views(client, wid, {"in": [], "out": ["image"]})
|
|
|
|
|
|
def test_reanalyze_refreshes_inputs_and_preserves_empty_output_declaration(client):
|
|
row = create(client, graph("LoadImageMask"))
|
|
client.patch(
|
|
f"/api/workflows/{row['id']}",
|
|
json={
|
|
"input_schema": [{"nodeId": "90", "kind": "text"}],
|
|
},
|
|
)
|
|
# Legacy auto schema: read/reanalyze must rederive it before media classification.
|
|
with get_db(settings.db_path) as conn:
|
|
conn.execute(
|
|
"UPDATE workflows SET input_schema = ? WHERE id = ?",
|
|
('[{"nodeId": "90", "kind": "text", "auto": true}]', row["id"]),
|
|
)
|
|
response = client.post(f"/api/workflows/{row['id']}/reanalyze")
|
|
assert response.status_code == 200
|
|
expected = {"in": ["image"], "out": []}
|
|
assert stored_media(row["id"]) == expected
|
|
assert_views(client, row["id"], expected)
|
|
|
|
|
|
def test_batch_import_uses_final_confirmed_input_schema(client):
|
|
response = client.post(
|
|
"/api/workflows/import-batch",
|
|
json={
|
|
"files": [
|
|
{
|
|
"workflow_json": _ui_graph(),
|
|
"input_schema": [{"nodeId": "2", "kind": "image_list", "manual": True}],
|
|
}
|
|
]
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
wid = response.json()["imported_ids"][0]
|
|
assert stored_media(wid) == {"in": ["image"], "out": ["image"]}
|
|
assert_views(client, wid, {"in": ["image"], "out": ["image"]})
|