117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
"""Tests for AI delta response parsing and action-type prompt dispatch."""
|
|
from app.core.ai_chat_service import _parse_delta, _build_action_prompt, _find_node_by_id
|
|
|
|
|
|
def test_parse_delta_from_response():
|
|
"""Service extracts [DELTA] markers from AI responses."""
|
|
response = '''Here's a new branch for that node.
|
|
|
|
[DELTA]
|
|
{"action": "add", "target_node_id": "check-dns", "nodes": [{"id": "verify-dns-server", "type": "decision", "question": "Is the DNS server responding?"}], "explanation": "Added DNS verification branch"}
|
|
[/DELTA]
|
|
|
|
Let me know if you'd like to adjust this.'''
|
|
|
|
parsed = _parse_delta(response)
|
|
assert parsed is not None
|
|
assert parsed["action"] == "add"
|
|
assert parsed["target_node_id"] == "check-dns"
|
|
assert len(parsed["nodes"]) == 1
|
|
|
|
|
|
def test_parse_delta_none_when_absent():
|
|
"""Returns None when no delta marker present."""
|
|
response = "Sure, I can explain that node. It checks connectivity."
|
|
parsed = _parse_delta(response)
|
|
assert parsed is None
|
|
|
|
|
|
def test_parse_delta_with_markdown_fences():
|
|
"""Handles delta JSON wrapped in markdown code fences."""
|
|
response = '''[DELTA]
|
|
```json
|
|
{"action": "modify", "target_node_id": "node-1", "nodes": [{"id": "node-1", "type": "action", "title": "Updated"}], "explanation": "Modified title"}
|
|
```
|
|
[/DELTA]'''
|
|
parsed = _parse_delta(response)
|
|
assert parsed is not None
|
|
assert parsed["action"] == "modify"
|
|
|
|
|
|
def test_parse_delta_invalid_json():
|
|
"""Returns None for invalid JSON inside delta markers."""
|
|
response = "[DELTA]not valid json[/DELTA]"
|
|
parsed = _parse_delta(response)
|
|
assert parsed is None
|
|
|
|
|
|
def test_build_action_prompt_generate_branch():
|
|
"""Generate branch action includes focal node context."""
|
|
tree = {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"question": "Is the server up?",
|
|
"children": [],
|
|
"options": [],
|
|
}
|
|
prompt = _build_action_prompt(
|
|
action_type="generate_branch",
|
|
focal_node_id="root",
|
|
tree_structure=tree,
|
|
flow_type="troubleshooting",
|
|
)
|
|
assert "root" in prompt
|
|
assert "generate" in prompt.lower() or "branch" in prompt.lower()
|
|
|
|
|
|
def test_build_action_prompt_open_chat():
|
|
"""Open chat action is general conversation."""
|
|
prompt = _build_action_prompt(
|
|
action_type="open_chat",
|
|
focal_node_id=None,
|
|
tree_structure={"id": "root", "type": "decision"},
|
|
flow_type="troubleshooting",
|
|
)
|
|
assert isinstance(prompt, str)
|
|
assert len(prompt) > 0
|
|
|
|
|
|
def test_find_node_by_id_root():
|
|
"""Finds root node."""
|
|
tree = {"id": "root", "type": "decision", "children": []}
|
|
assert _find_node_by_id(tree, "root") is not None
|
|
|
|
|
|
def test_find_node_by_id_nested():
|
|
"""Finds nested child node."""
|
|
tree = {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"children": [
|
|
{"id": "child-1", "type": "action", "children": []},
|
|
{"id": "child-2", "type": "solution", "children": []},
|
|
],
|
|
}
|
|
found = _find_node_by_id(tree, "child-2")
|
|
assert found is not None
|
|
assert found["id"] == "child-2"
|
|
|
|
|
|
def test_find_node_by_id_not_found():
|
|
"""Returns None for non-existent node."""
|
|
tree = {"id": "root", "type": "decision", "children": []}
|
|
assert _find_node_by_id(tree, "nonexistent") is None
|
|
|
|
|
|
def test_find_node_by_id_in_steps():
|
|
"""Finds node in procedural steps array."""
|
|
tree = {
|
|
"steps": [
|
|
{"id": "step-1", "type": "procedure_step"},
|
|
{"id": "step-2", "type": "procedure_step"},
|
|
]
|
|
}
|
|
found = _find_node_by_id(tree, "step-2")
|
|
assert found is not None
|
|
assert found["id"] == "step-2"
|