"""Regression tests for WorkflowGraph edge/graph constraints + the rule-based audit that mirrors them. Each fixture in `dto_fixtures/` is either a clean workflow or a single category of violation we found in production. We pin two layers: 1. WorkflowGraph — semantic gate used by `/publish`, the SDK API, and both MCP tools. Driven by `NodeSpec.graph_constraints`. If this layer ever stops rejecting one of these fixtures, the production write paths will quietly start accepting bad workflows again. 2. audit_definition (api.services.workflow.audit) — read-only sweep over persisted rows for one-off cleanup tooling that finds legacy/imported breakage. Pinned so refactors of the rule set don't silently change those cleanup verdicts. DTO-level shape validation is covered by `test_dto.py` and isn't re-pinned here. """ import json from pathlib import Path import pytest from api.services.workflow.audit import audit_definition from api.services.workflow.dto import ReactFlowDTO from api.services.workflow.workflow_graph import ( WorkflowGraph, validate_unique_transition_tool_names, ) _FIXTURES_DIR = Path(__file__).parent / "dto_fixtures" def _load(name: str) -> tuple[str, dict]: raw = (_FIXTURES_DIR / f"{name}.json").read_text() return raw, json.loads(raw) # (fixture_name, expected_audit_reasons, expected_graph_messages) # # expected_graph_messages semantics: # None — DTO rejects upstream, WorkflowGraph is never reached. # [] — WorkflowGraph accepts (clean fixture). # [...] — WorkflowGraph rejects; each substring must appear in the # emitted WorkflowError messages. _SCENARIOS = [ ("clean", [], []), # A workflow with just a startCall and no edges is valid — startCall # has no `min_outgoing` constraint, so a "greet then hang up" flow # passes both audit and WorkflowGraph. ("start_only", [], []), ( "bad_edge_into_start", ["target_max_incoming_0:startCall"], ["Start Call cannot have incoming edges"], ), ( "bad_edge_into_webhook", ["target_max_incoming_0:webhook"], ["Webhook cannot have incoming edges"], ), ( "bad_edge_out_of_webhook", ["source_max_outgoing_0:webhook"], ["Webhook cannot have outgoing edges"], ), ( "bad_edge_out_of_globalnode", ["source_max_outgoing_0:globalNode"], ["Global Node cannot have outgoing edges"], ), ("bad_edge_target_missing", ["target_id_missing"], None), ("bad_edge_source_missing", ["source_id_missing"], None), ( "no_start_node", ["no_start_node"], ["Workflow must have at least one Start Call"], ), # Two startCall nodes — surfaced separately from no_start_node so # the editor can show a count-specific message. ( "multiple_start_nodes", ["multiple_start_nodes:2"], ["Workflow can have at most one Start Call"], ), ( "multiple_trigger_nodes", ["max_instances_1:trigger:2"], ["Workflow can have at most one API Trigger"], ), ( "multiple_global_nodes", ["max_instances_1:globalNode:2"], ["Workflow can have at most one Global Node"], ), ] @pytest.mark.parametrize( "name,expected_reasons", [(name, reasons) for name, reasons, _ in _SCENARIOS], ) def test_audit_catches_violations(name, expected_reasons): _, definition = _load(name) violations = audit_definition(definition["nodes"], definition["edges"]) reasons = sorted(v["reason"] for v in violations) assert reasons == sorted(expected_reasons) @pytest.mark.parametrize( "name,expected_graph_messages", [ (name, messages) for name, _, messages in _SCENARIOS if messages is not None # skip fixtures DTO rejects upstream ], ) def test_workflow_graph_rejects_violations(name, expected_graph_messages): """If WorkflowGraph accepts a definition, every save path that goes 'live' will accept it — so this layer is the canonical regression point for the rules in `NodeSpec.graph_constraints`.""" raw, _ = _load(name) dto = ReactFlowDTO.model_validate_json(raw) if not expected_graph_messages: WorkflowGraph(dto) return with pytest.raises(ValueError) as exc_info: WorkflowGraph(dto) actual_messages = [w["message"] for w in exc_info.value.args[0]] for expected in expected_graph_messages: assert any(expected in m for m in actual_messages), ( f"Expected substring {expected!r} not found in graph errors: {actual_messages}" ) def test_workflow_graph_can_skip_duplicate_api_trigger_check_for_runtime(): raw, _ = _load("multiple_trigger_nodes") dto = ReactFlowDTO.model_validate_json(raw) WorkflowGraph(dto, skip_instance_constraints_for={"trigger"}) def test_workflow_graph_start_semantics_come_from_node_type_not_legacy_flag(): dto = ReactFlowDTO.model_validate( { "nodes": [ { "id": "start-1", "type": "startCall", "position": {"x": 0, "y": 0}, "data": { "name": "Start", "prompt": "Greet.", "is_start": False, }, } ], "edges": [], } ) graph = WorkflowGraph(dto) assert graph.start_node_id == "start-1" assert graph.nodes["start-1"].is_start is True def test_workflow_graph_rejects_duplicate_transition_tool_names(): dto = ReactFlowDTO.model_validate( { "nodes": [ { "id": "start-1", "type": "startCall", "position": {"x": 0, "y": 0}, "data": {"name": "Start", "prompt": "Greet."}, }, { "id": "end-1", "type": "endCall", "position": {"x": 0, "y": 200}, "data": {"name": "End one", "prompt": "Bye."}, }, { "id": "end-2", "type": "endCall", "position": {"x": 200, "y": 200}, "data": {"name": "End two", "prompt": "Bye."}, }, ], "edges": [ { "id": "edge-1", "source": "start-1", "target": "end-1", "data": { "label": "Go to sales", "condition": "Caller wants sales.", }, }, { "id": "edge-2", "source": "start-1", "target": "end-2", "data": { "label": "go-to-sales", "condition": "Caller asks for the sales team.", }, }, ], } ) with pytest.raises(ValueError) as exc_info: WorkflowGraph(dto) errors = exc_info.value.args[0] assert {error["id"] for error in errors} == {"edge-1", "edge-2"} assert all(error["field"] == "data.label" for error in errors) assert all('"go_to_sales"' in error["message"] for error in errors) def test_duplicate_transition_tool_names_are_scoped_to_source_node(): errors = validate_unique_transition_tool_names( [ ("edge-1", "source-1", "Continue"), ("edge-2", "source-2", "Continue"), ] ) assert errors == []