File size: 6,399 Bytes
0c85e62 | 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 186 187 | """Regression tests for scripts/validate_workflow.py.
Anchored by PR #170 review: the validator must report problems rather than
raise on malformed-but-valid JSON. These tests pin that behavior plus a
sampling of the existing happy-path and consistency checks so future edits
cannot silently regress the safety guarantees.
"""
from __future__ import annotations
import importlib.util
import json
from pathlib import Path
import pytest
# Load the script directly — it lives under scripts/, not under a package.
_SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "validate_workflow.py"
_spec = importlib.util.spec_from_file_location("validate_workflow", _SCRIPT)
assert _spec is not None and _spec.loader is not None
_module = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_module)
validate = _module.validate
main = _module.main
def _minimal_clean() -> dict:
"""Smallest workflow that should validate cleanly: two nodes, one link."""
return {
"last_node_id": 2,
"last_link_id": 1,
"nodes": [
{
"id": 1,
"type": "Src",
"inputs": [],
"outputs": [
{"name": "out", "type": "FOO", "links": [1]},
],
},
{
"id": 2,
"type": "Dst",
"inputs": [
{"name": "in", "type": "FOO", "link": 1},
],
"outputs": [],
},
],
"links": [[1, 1, 0, 2, 0, "FOO"]],
}
# --- Happy path -------------------------------------------------------------
def test_minimal_clean_workflow_has_no_problems() -> None:
assert validate(_minimal_clean()) == []
# --- PR #170 review repros: malformed-but-valid JSON must not crash --------
@pytest.mark.parametrize("data", [None, 42, 3.14, "string", [1, 2], (1, 2)])
def test_top_level_not_object_reports_does_not_crash(data) -> None:
problems = validate(data)
assert problems, "expected at least one problem"
assert "top-level JSON must be an object" in problems[0]
def test_node_entry_not_dict_reports_does_not_crash() -> None:
# Reviewer's repro 1: nodes contains an int.
data = {"nodes": [1], "links": [], "last_node_id": 0, "last_link_id": 0}
problems = validate(data)
assert any("nodes[0]" in p and "object" in p for p in problems), problems
def test_nodes_field_not_list_reports() -> None:
data = {"nodes": {"oops": "dict not list"}, "links": []}
problems = validate(data)
assert any("'nodes' must be a list" in p for p in problems), problems
def test_links_field_not_list_reports() -> None:
data = {"nodes": [], "links": "not-a-list"}
problems = validate(data)
assert any("'links' must be a list" in p for p in problems), problems
def test_inputs_field_not_list_reports() -> None:
data = _minimal_clean()
data["nodes"][1]["inputs"] = {"oops": "dict not list"}
problems = validate(data)
assert any("'inputs' must be a list" in p for p in problems), problems
def test_input_entry_not_dict_reports() -> None:
data = _minimal_clean()
data["nodes"][1]["inputs"] = ["string-not-dict"]
problems = validate(data)
assert any("inputs[0]" in p and "object" in p for p in problems), problems
def test_output_entry_not_dict_reports() -> None:
data = _minimal_clean()
data["nodes"][0]["outputs"] = [99]
problems = validate(data)
assert any("outputs[0]" in p and "object" in p for p in problems), problems
def test_malformed_link_not_list_reports() -> None:
data = _minimal_clean()
data["links"].append("not-a-list")
problems = validate(data)
assert any("links[1]" in p for p in problems), problems
def test_malformed_link_wrong_length_reports() -> None:
data = _minimal_clean()
data["links"].append([2, 1, 0]) # too short
problems = validate(data)
assert any("links[1]" in p and "6 elements" in p for p in problems), problems
# --- Existing consistency checks -------------------------------------------
def test_link_to_missing_src_node_reports() -> None:
data = _minimal_clean()
data["links"] = [[1, 99, 0, 2, 0, "FOO"]]
data["nodes"][0]["outputs"][0]["links"] = [] # not referenced from anywhere now
problems = validate(data)
assert any("src node 99 missing" in p for p in problems), problems
def test_type_mismatch_on_slot_reports() -> None:
data = _minimal_clean()
data["links"] = [[1, 1, 0, 2, 0, "BAR"]] # link declares BAR, slots are FOO
problems = validate(data)
assert any("type mismatch" in p for p in problems), problems
def test_wildcard_destination_accepts_concrete_link_type() -> None:
data = _minimal_clean()
data["nodes"][1]["inputs"][0]["type"] = "*"
assert validate(data) == []
def test_duplicate_node_ids_report() -> None:
data = _minimal_clean()
data["nodes"][1]["id"] = 1 # collision with first node
problems = validate(data)
assert any("Duplicate node IDs" in p for p in problems), problems
def test_endpoint_ref_desync_reports() -> None:
data = _minimal_clean()
data["nodes"][0]["outputs"][0]["links"] = [] # source side forgot link 1
problems = validate(data)
assert any("missing from node 1 output[0]" in p for p in problems), problems
def test_last_node_id_too_low_reports() -> None:
data = _minimal_clean()
data["last_node_id"] = 0 # but max is 2
problems = validate(data)
assert any("last_node_id" in p for p in problems), problems
# --- CLI smoke (covers main() exit codes) ----------------------------------
def test_cli_returns_zero_on_clean_workflow(tmp_path: Path) -> None:
p = tmp_path / "clean.json"
p.write_text(json.dumps(_minimal_clean()), encoding="utf-8")
assert main([str(p), "--quiet"]) == 0
def test_cli_returns_one_on_problems(tmp_path: Path) -> None:
p = tmp_path / "broken.json"
p.write_text(json.dumps({"nodes": [1], "links": []}), encoding="utf-8")
assert main([str(p), "--quiet"]) == 1
def test_cli_returns_two_on_missing_file(tmp_path: Path) -> None:
assert main([str(tmp_path / "does-not-exist.json"), "--quiet"]) == 2
def test_cli_returns_two_on_invalid_json(tmp_path: Path) -> None:
p = tmp_path / "bad.json"
p.write_text("{not valid", encoding="utf-8")
assert main([str(p), "--quiet"]) == 2
|