Files
resolutionflow/backend/app/core/tree_validation.py
2026-02-18 01:58:24 -05:00

290 lines
10 KiB
Python

"""Tree validation helper module for draft/published workflow."""
from typing import Any
PROCEDURAL_TREE_TYPES = {"procedural", "maintenance"}
class TreeValidationError(Exception):
"""Custom exception for tree validation errors."""
def __init__(self, field: str, message: str):
self.field = field
self.message = message
super().__init__(f"{field}: {message}")
# --- Troubleshooting Tree Validation ---
def validate_tree_structure(tree_structure: dict[str, Any]) -> tuple[bool, list[dict[str, str]]]:
"""Validate troubleshooting tree structure for publishing.
A valid tree for publishing must have:
- A root node with id, type, and appropriate content fields
- All decision nodes must have a question field
- All decision nodes with children must have at least 2 children
- All action nodes must have an action field
- All solution nodes must have a solution field
- No orphaned nodes (all nodes reachable from root)
Args:
tree_structure: The tree structure dict to validate
Returns:
Tuple of (is_valid, list of errors)
Each error is a dict with 'field' and 'message' keys
"""
errors = []
# Check root node exists
if not tree_structure:
errors.append({"field": "tree_structure", "message": "Tree structure cannot be empty"})
return False, errors
if "id" not in tree_structure:
errors.append({"field": "tree_structure.id", "message": "Root node must have an id"})
if "type" not in tree_structure:
errors.append({"field": "tree_structure.type", "message": "Root node must have a type"})
return False, errors
# Validate root node based on type
_validate_node(tree_structure, "root", errors)
# Validate all child nodes recursively
if "children" in tree_structure:
_validate_children(tree_structure["children"], "root.children", errors)
# Block publish if any answer placeholder nodes remain
if _has_answer_nodes(tree_structure):
errors.append({
"field": "tree_structure",
"message": "Answer placeholders must be resolved to a node type before publishing."
})
return len(errors) == 0, errors
def _validate_node(node: dict[str, Any], path: str, errors: list[dict[str, str]]) -> None:
"""Validate a single node in the tree structure."""
node_type = node.get("type")
if node_type == "decision":
if "question" not in node or not node["question"]:
errors.append({
"field": f"{path}.question",
"message": "Decision nodes must have a non-empty question"
})
# If node has children, must have at least 2 (for decision branches)
children = node.get("children", [])
if children and len(children) < 2:
errors.append({
"field": f"{path}.children",
"message": "Decision nodes with children must have at least 2 branches"
})
elif node_type == "action":
if "action" not in node or not node["action"]:
errors.append({
"field": f"{path}.action",
"message": "Action nodes must have a non-empty action"
})
elif node_type == "solution":
if "solution" not in node or not node["solution"]:
errors.append({
"field": f"{path}.solution",
"message": "Solution nodes must have a non-empty solution"
})
elif node_type == "answer":
# Answer nodes are draft-only placeholders — no structural validation needed
pass
else:
errors.append({
"field": f"{path}.type",
"message": f"Unknown node type: {node_type}"
})
def _validate_children(children: list[dict[str, Any]], path: str, errors: list[dict[str, str]]) -> None:
"""Recursively validate child nodes."""
for i, child in enumerate(children):
child_path = f"{path}[{i}]"
if "id" not in child:
errors.append({"field": f"{child_path}.id", "message": "Child node must have an id"})
if "type" not in child:
errors.append({"field": f"{child_path}.type", "message": "Child node must have a type"})
continue
_validate_node(child, child_path, errors)
# Recursively validate grandchildren
if "children" in child:
_validate_children(child["children"], f"{child_path}.children", errors)
def _has_answer_nodes(node: dict[str, Any]) -> bool:
"""Recursively check if any node in the tree has type 'answer'."""
if node.get("type") == "answer":
return True
for child in node.get("children", []):
if _has_answer_nodes(child):
return True
return False
# --- Procedural Tree Validation ---
VALID_STEP_TYPES = {"procedure_step", "procedure_end", "section_header"}
VALID_CONTENT_TYPES = {"action", "informational", "verification", "warning"}
def validate_procedural_structure(tree_structure: dict[str, Any]) -> tuple[bool, list[dict[str, str]]]:
"""Validate procedural tree structure for publishing.
Procedural trees store steps as a flat ordered array in tree_structure["steps"].
Rules:
- Must have a non-empty "steps" array
- Each step must have: id, type, title
- Only procedure_step and procedure_end types allowed
- Must have exactly one procedure_end (as the last step)
- All other steps must be procedure_step
- No duplicate step IDs
- Steps with content_type must use valid values
Args:
tree_structure: Dict with a "steps" key containing the ordered step array
Returns:
Tuple of (is_valid, list of errors)
"""
errors = []
if not tree_structure:
errors.append({"field": "tree_structure", "message": "Tree structure cannot be empty"})
return False, errors
steps = tree_structure.get("steps")
if not steps or not isinstance(steps, list):
errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have a non-empty steps array"})
return False, errors
# Track IDs for uniqueness
seen_ids: set[str] = set()
end_count = 0
for i, step in enumerate(steps):
path = f"steps[{i}]"
# Required fields
step_id = step.get("id")
if not step_id:
errors.append({"field": f"{path}.id", "message": "Step must have an id"})
elif step_id in seen_ids:
errors.append({"field": f"{path}.id", "message": f"Duplicate step id: {step_id}"})
else:
seen_ids.add(step_id)
step_type = step.get("type")
if not step_type:
errors.append({"field": f"{path}.type", "message": "Step must have a type"})
elif step_type not in VALID_STEP_TYPES:
errors.append({"field": f"{path}.type", "message": f"Invalid step type: {step_type}. Must be one of: {', '.join(VALID_STEP_TYPES)}"})
elif step_type == "procedure_end":
end_count += 1
# procedure_end must be last step
if i != len(steps) - 1:
errors.append({"field": f"{path}.type", "message": "procedure_end must be the last step"})
if not step.get("title"):
errors.append({"field": f"{path}.title", "message": "Step must have a non-empty title"})
# Validate content_type if present
content_type = step.get("content_type")
if content_type and content_type not in VALID_CONTENT_TYPES:
errors.append({"field": f"{path}.content_type", "message": f"Invalid content_type: {content_type}. Must be one of: {', '.join(VALID_CONTENT_TYPES)}"})
# Must have exactly one end step
if end_count == 0:
errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have a procedure_end step as the last step"})
elif end_count > 1:
errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have exactly one procedure_end step"})
return len(errors) == 0, errors
# --- Dispatch ---
def can_publish_tree(
tree_structure: dict[str, Any],
name: str,
description: str | None = None,
tree_type: str = "troubleshooting",
intake_form: list[dict[str, Any]] | None = None,
) -> tuple[bool, list[dict[str, str]]]:
"""Check if a tree can be published.
Dispatches to the appropriate validator based on tree_type.
Args:
tree_structure: The tree structure to validate
name: The tree name
description: Optional tree description
tree_type: 'troubleshooting' or 'procedural'
intake_form: Optional intake form fields (procedural only)
Returns:
Tuple of (can_publish, list of errors)
"""
errors = []
# Validate name
if not name or not name.strip():
errors.append({"field": "name", "message": "Tree must have a name to be published"})
# Validate structure based on tree type
if tree_type in PROCEDURAL_TREE_TYPES:
structure_valid, structure_errors = validate_procedural_structure(tree_structure)
else:
structure_valid, structure_errors = validate_tree_structure(tree_structure)
errors.extend(structure_errors)
# Validate intake form if present (procedural only)
if intake_form and tree_type in PROCEDURAL_TREE_TYPES:
form_valid, form_errors = _validate_intake_form(intake_form)
errors.extend(form_errors)
return len(errors) == 0, errors
def _validate_intake_form(intake_form: list[dict[str, Any]]) -> tuple[bool, list[dict[str, str]]]:
"""Validate intake form field definitions."""
errors = []
variable_names: set[str] = set()
for i, field in enumerate(intake_form):
path = f"intake_form[{i}]"
var_name = field.get("variable_name")
if not var_name:
errors.append({"field": f"{path}.variable_name", "message": "Field must have a variable_name"})
elif var_name in variable_names:
errors.append({"field": f"{path}.variable_name", "message": f"Duplicate variable_name: {var_name}"})
else:
variable_names.add(var_name)
if not field.get("label"):
errors.append({"field": f"{path}.label", "message": "Field must have a label"})
field_type = field.get("field_type")
if field_type in ("select", "multi_select"):
options = field.get("options")
if not options or len(options) == 0:
errors.append({"field": f"{path}.options", "message": f"{field_type} fields must have at least one option"})
return len(errors) == 0, errors