"""Validation for strategy DSL."""
from dataclasses import dataclass
from veupath_chatbot.domain.strategy.ast import (
PlanStepNode,
StrategyAST,
)
from veupath_chatbot.domain.strategy.ops import CombineOp
[docs]
@dataclass
class StepValidationIssue:
"""A single validation issue found during step/strategy validation."""
path: str
message: str
code: str
[docs]
@dataclass
class ValidationResult:
"""Result of validation."""
valid: bool
errors: list[StepValidationIssue]
[docs]
@classmethod
def success(cls) -> ValidationResult:
"""Create a successful result."""
return cls(valid=True, errors=[])
[docs]
@classmethod
def failure(cls, errors: list[StepValidationIssue]) -> ValidationResult:
"""Create a failed result.
:param errors: Validation errors list.
"""
return cls(valid=False, errors=errors)
[docs]
class StrategyValidator:
"""Validates strategy AST."""
[docs]
def __init__(
self,
available_searches: dict[str, list[str]] | None = None,
available_transforms: list[str] | None = None,
) -> None:
"""Initialize validator.
:param available_searches: Map of record_type -> list of search names.
:param available_transforms: List of available transform names.
"""
self.available_searches = available_searches or {}
self.available_transforms = available_transforms or []
[docs]
def validate(self, strategy: StrategyAST) -> ValidationResult:
"""Validate a strategy AST.
:param strategy: Strategy AST.
"""
errors: list[StepValidationIssue] = []
# Validate record type
if not strategy.record_type:
errors.append(
StepValidationIssue(
path="recordType",
message="Record type is required",
code="MISSING_RECORD_TYPE",
)
)
# Check for empty strategy
if strategy.root is None:
errors.append(
StepValidationIssue(
path="root",
message="Strategy must have at least one step",
code="EMPTY_STRATEGY",
)
)
else:
# Validate the tree
self._validate_node(strategy.root, "root", strategy.record_type, errors)
return (
ValidationResult.success()
if not errors
else ValidationResult.failure(errors)
)
def _validate_node(
self,
node: PlanStepNode,
path: str,
expected_record_type: str,
errors: list[StepValidationIssue],
) -> None:
"""Validate a single node in the AST.
:param node: Plan step node.
:param path: Node path.
:param expected_record_type: Expected record type.
:param errors: Validation errors list.
"""
kind = node.infer_kind()
if not node.search_name:
errors.append(
StepValidationIssue(
path=f"{path}.searchName",
message="searchName is required",
code="MISSING_SEARCH_NAME",
)
)
if self.available_searches:
rt_searches = self.available_searches.get(expected_record_type, [])
if node.search_name and node.search_name not in rt_searches:
errors.append(
StepValidationIssue(
path=f"{path}.searchName",
message=f"Unknown search: {node.search_name}",
code="UNKNOWN_SEARCH",
)
)
if kind == "combine":
if node.operator is None:
errors.append(
StepValidationIssue(
path=f"{path}.operator",
message="operator is required for combine nodes",
code="MISSING_OPERATOR",
)
)
elif node.operator not in CombineOp:
errors.append(
StepValidationIssue(
path=f"{path}.operator",
message=f"Invalid operator: {node.operator}",
code="INVALID_OPERATOR",
)
)
if node.primary_input is None or node.secondary_input is None:
errors.append(
StepValidationIssue(
path=path,
message="Combine nodes require two inputs",
code="MISSING_INPUT",
)
)
if node.operator == CombineOp.COLOCATE:
if node.colocation_params is None:
errors.append(
StepValidationIssue(
path=f"{path}.colocationParams",
message="colocationParams is required for COLOCATE",
code="MISSING_COLOCATION_PARAMS",
)
)
else:
for err in node.colocation_params.validate():
errors.append(
StepValidationIssue(
path=f"{path}.colocationParams",
message=err,
code="INVALID_COLOCATION_PARAMS",
)
)
if node.secondary_input is not None:
self._validate_node(
node.secondary_input,
f"{path}.secondaryInput",
expected_record_type,
errors,
)
if node.primary_input is not None:
self._validate_node(
node.primary_input, f"{path}.primaryInput", expected_record_type, errors
)
[docs]
def validate_strategy(strategy: StrategyAST) -> ValidationResult:
"""Validate a strategy AST with default validator.
:param strategy: Strategy AST.
"""
return StrategyValidator().validate(strategy)