import pytest
from src.linter.schemas import FlowInput, Node, Edge
from src.linter.engine import lint_flow
def test_valid_flow():
"""Test a clean linear flow"""
flow = FlowInput(
nodes=[
Node(id="start", type="Start"),
Node(id="end", type="End")
],
edges=[
Edge(from_node="start", to_node="end")
]
)
result = lint_flow(flow)
assert result.score == 100
assert len(result.errors) == 0
assert len(result.warnings) == 0
def test_unreachable_node():
"""Test detection of disconnected nodes"""
flow = FlowInput(
nodes=[
Node(id="start", type="Start"),
Node(id="orphan", type="Speak", prompt="Nobody hears me"),
Node(id="end", type="End")
],
edges=[
Edge(from_node="start", to_node="end")
]
)
result = lint_flow(flow)
assert result.score < 100
codes = [e.code for e in result.errors]
assert "unreachable_node" in codes
def test_dead_end():
"""Test detection of nodes that don't terminate properly"""
flow = FlowInput(
nodes=[
Node(id="start", type="Start"),
Node(id="stuck", type="Speak", prompt="Iam stuck")
],
edges=[
Edge(from_node="start", to_node="stuck")
]
)
result = lint_flow(flow)
codes = [e.code for e in result.errors]
assert "dead_end_not_end" in codes
def test_missing_fallback():
"""Test missing fallback logic on Ask nodes"""
flow = FlowInput(
nodes=[
Node(id="start", type="Start"),
Node(id="ask_something", type="Ask", prompt="Say yes"),
Node(id="end", type="End")
],
edges=[
Edge(from_node="start", to_node="ask_something"),
Edge(from_node="ask_something", to_node="end", label="yes")
]
)
result = lint_flow(flow)
codes = [e.code for e in result.errors]
assert "missing_fallback_on_ask" in codes
def test_var_defined_check_basic():
"""Test basic variable usage check"""
flow = FlowInput(
nodes=[
Node(id="start", type="Start", vars_defined=["my_var"]),
Node(id="use_it", type="Speak", prompt="Usage", vars_used=["my_var", "unknown_var"]),
Node(id="end", type="End")
],
edges=[
Edge(from_node="start", to_node="use_it"),
Edge(from_node="use_it", to_node="end")
]
)
result = lint_flow(flow)
codes = [e.code for e in result.errors]
error_msgs = [e.message for e in result.errors if e.code == "var_used_not_defined"]
# my_var is defined, so no error for it
if any("my_var" in msg for msg in error_msgs):
pytest.fail("Should not report error for defined variable 'my_var'")
# unknown_var is NOT defined, should error
assert any("unknown_var" in msg for msg in error_msgs)
def test_warnings():
"""Test warning rules"""
long_prompt = "a" * 300
flow = FlowInput(
nodes=[
Node(id="start", type="Start"),
Node(id="longy", type="Speak", prompt=long_prompt),
Node(id="too_many_qs", type="Ask", prompt="Q1? Q2? Q3?"),
Node(id="end", type="End")
],
edges=[
Edge(from_node="start", to_node="longy"),
Edge(from_node="longy", to_node="too_many_qs"),
Edge(from_node="too_many_qs", to_node="end", is_fallback=True) # Prevent missing fallback error
]
)
result = lint_flow(flow)
codes = [w.code for w in result.warnings]
assert "prompt_too_long" in codes
assert "multi_question" in codes
def test_idempotency():
"""Test that result is cached and consistent"""
flow = FlowInput(
nodes=[
Node(id="start", type="Start"),
Node(id="end", type="End")
],
edges=[
Edge(from_node="start", to_node="end")
]
)
res1 = lint_flow(flow)
res2 = lint_flow(flow)
assert res1.flow_hash == res2.flow_hash
assert res1 is res2 # Same object ID from cache
