93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
"""Phase 4.2+4.3: YAML workflow + ACA protocol registry tests."""
|
|
import pytest
|
|
from app.core.yaml_workflow import validate_yaml, yaml_to_orchestration, orchestration_to_yaml
|
|
|
|
|
|
# ── YAML workflow tests ──────────────────────────────────────────────────────
|
|
|
|
VALID_YAML = """
|
|
version: "1.0"
|
|
name: "Test"
|
|
nodes:
|
|
- id: a
|
|
type: llm
|
|
config:
|
|
model: auto
|
|
system_prompt: "You are a helper."
|
|
- id: b
|
|
type: llm
|
|
- id: c
|
|
type: reduce
|
|
config:
|
|
strategy: concat
|
|
edges:
|
|
- from: a
|
|
to: b
|
|
- from: b
|
|
to: c
|
|
- from: a
|
|
to: c
|
|
"""
|
|
|
|
|
|
def test_validate_valid():
|
|
import yaml
|
|
data = yaml.safe_load(VALID_YAML)
|
|
r = validate_yaml(data)
|
|
assert r["ok"], r["errors"]
|
|
assert r["errors"] == []
|
|
|
|
|
|
def test_validate_invalid_type():
|
|
import yaml
|
|
data = yaml.safe_load("version: '1.0'\nnodes:\n - id: x\n type: bogus")
|
|
r = validate_yaml(data)
|
|
assert not r["ok"]
|
|
assert any("bogus" in e for e in r["errors"])
|
|
|
|
|
|
def test_validate_cycle():
|
|
import yaml
|
|
data = yaml.safe_load("version: '1.0'\nnodes:\n - id: a\n type: llm\n - id: b\n type: llm\nedges:\n - from: a\n to: b\n - from: b\n to: a")
|
|
r = validate_yaml(data)
|
|
assert not r["ok"]
|
|
assert any("环" in e for e in r["errors"])
|
|
|
|
|
|
def test_yaml_to_json_roundtrip():
|
|
import yaml
|
|
data = yaml.safe_load(VALID_YAML)
|
|
orch = yaml_to_orchestration(data)
|
|
assert len(orch["nodes"]) == 3
|
|
assert len(orch["edges"]) == 3
|
|
# config flattening
|
|
node_a = next(n for n in orch["nodes"] if n["id"] == "a")
|
|
assert node_a.get("model_id") == "auto"
|
|
assert node_a.get("system_prompt") == "You are a helper."
|
|
|
|
|
|
def test_json_to_yaml_to_json():
|
|
"""Roundtrip: JSON → YAML obj → JSON should preserve key fields."""
|
|
orch_json = {
|
|
"nodes": [
|
|
{"id": "x", "type": "llm", "model_id": "auto"},
|
|
{"id": "y", "type": "reduce", "strategy": "concat"},
|
|
],
|
|
"edges": [{"from": "x", "to": "y"}],
|
|
}
|
|
yaml_obj = orchestration_to_yaml(orch_json)
|
|
assert yaml_obj["version"] == "1.0"
|
|
# config nesting
|
|
node_x = next(n for n in yaml_obj["nodes"] if n["id"] == "x")
|
|
assert node_x.get("config", {}).get("model") == "auto"
|
|
node_y = next(n for n in yaml_obj["nodes"] if n["id"] == "y")
|
|
assert node_y.get("config", {}).get("strategy") == "concat"
|
|
|
|
|
|
def test_validate_missing_agent_id():
|
|
import yaml
|
|
data = yaml.safe_load("version: '1.0'\nnodes:\n - id: x\n type: local\n - id: y\n type: llm\nedges:\n - from: x\n to: y")
|
|
r = validate_yaml(data)
|
|
assert not r["ok"]
|
|
assert any("agent_id" in e for e in r["errors"])
|