""" Script Flow Tester for AI Service. [AC-AISVC-101, AC-AISVC-102] Flow simulation and coverage analysis. """ import logging import re import time from dataclasses import dataclass, field from typing import Any from app.models.entities import ScriptFlow logger = logging.getLogger(__name__) @dataclass class MatchedCondition: """Matched condition details.""" type: str goto_step: int keywords: list[str] | None = None pattern: str | None = None @dataclass class FlowSimulationStep: """Single step in flow simulation.""" stepNo: int botMessage: str userInput: str matchedCondition: MatchedCondition | None nextStep: int | None durationMs: int = 0 @dataclass class FlowSimulationResult: """Result of flow simulation.""" flowId: str flowName: str simulation: list[FlowSimulationStep] = field(default_factory=list) result: dict[str, Any] = field(default_factory=dict) coverage: dict[str, Any] = field(default_factory=dict) issues: list[str] = field(default_factory=list) class ScriptFlowTester: """ [AC-AISVC-101, AC-AISVC-102] Script flow simulation and coverage analysis. Features: - Simulate flow execution with user inputs - Calculate step coverage rate - Detect issues: dead loops, low coverage, uncovered branches - No database modification (read-only simulation) """ MIN_COVERAGE_THRESHOLD = 0.8 MAX_SIMULATION_STEPS_MULTIPLIER = 2 def simulate_flow( self, flow: ScriptFlow, user_inputs: list[str], ) -> FlowSimulationResult: """ [AC-AISVC-101] Simulate flow execution and analyze coverage. Args: flow: ScriptFlow entity to simulate user_inputs: List of user inputs to feed into the flow Returns: FlowSimulationResult with simulation steps, coverage, and issues """ logger.info( f"[AC-AISVC-101] Starting flow simulation: flow_id={flow.id}, " f"flow_name={flow.name}, inputs_count={len(user_inputs)}" ) simulation: list[FlowSimulationStep] = [] current_step = 1 visited_steps: set[int] = set() total_steps = len(flow.steps) total_duration_ms = 0 final_message: str | None = None completed = False for user_input in user_inputs: if current_step > total_steps: completed = True break step_start = time.time() step_def = flow.steps[current_step - 1] visited_steps.add(current_step) matched_condition, next_step = self._match_next_step(step_def, user_input) if next_step is None: default_next = step_def.get("default_next") if default_next: next_step = default_next matched_condition = MatchedCondition( type="default", goto_step=default_next, ) else: next_step = current_step simulation.append( FlowSimulationStep( stepNo=current_step, botMessage=step_def.get("content", ""), userInput=user_input, matchedCondition=matched_condition, nextStep=next_step, durationMs=int((time.time() - step_start) * 1000), ) ) total_duration_ms += simulation[-1].durationMs if next_step > total_steps: completed = True final_message = step_def.get("content", "") break current_step = next_step if len(simulation) > total_steps * self.MAX_SIMULATION_STEPS_MULTIPLIER: logger.warning( f"[AC-AISVC-101] Simulation exceeded max steps: " f"flow_id={flow.id}, steps={len(simulation)}" ) break covered_steps = len(visited_steps) coverage_rate = covered_steps / total_steps if total_steps > 0 else 0.0 uncovered_steps = set(range(1, total_steps + 1)) - visited_steps issues = self._detect_issues( simulation=simulation, total_steps=total_steps, coverage_rate=coverage_rate, uncovered_steps=uncovered_steps, ) result = FlowSimulationResult( flowId=str(flow.id), flowName=flow.name, simulation=simulation, result={ "completed": completed, "totalSteps": total_steps, "totalDurationMs": total_duration_ms, "finalMessage": final_message, }, coverage={ "totalSteps": total_steps, "coveredSteps": covered_steps, "coverageRate": round(coverage_rate, 2), "uncoveredSteps": sorted(list(uncovered_steps)), }, issues=issues, ) logger.info( f"[AC-AISVC-101] Flow simulation completed: flow_id={flow.id}, " f"coverage_rate={coverage_rate:.2%}, issues_count={len(issues)}" ) return result def _match_next_step( self, step_def: dict[str, Any], user_input: str, ) -> tuple[MatchedCondition | None, int | None]: """ [AC-AISVC-101] Match user input against next_conditions. Args: step_def: Current step definition user_input: User's input message Returns: Tuple of (matched_condition, goto_step) or (None, None) """ next_conditions = step_def.get("next_conditions", []) if not next_conditions: return None, None user_input_lower = user_input.lower() for condition in next_conditions: keywords = condition.get("keywords", []) for keyword in keywords: if keyword.lower() in user_input_lower: matched = MatchedCondition( type="keyword", keywords=keywords, goto_step=condition.get("goto_step"), ) return matched, condition.get("goto_step") pattern = condition.get("pattern") if pattern: try: if re.search(pattern, user_input, re.IGNORECASE): matched = MatchedCondition( type="pattern", pattern=pattern, goto_step=condition.get("goto_step"), ) return matched, condition.get("goto_step") except re.error: logger.warning(f"Invalid regex pattern: {pattern}") return None, None def _detect_issues( self, simulation: list[FlowSimulationStep], total_steps: int, coverage_rate: float, uncovered_steps: set[int], ) -> list[str]: """ [AC-AISVC-102] Detect issues in flow simulation. Args: simulation: List of simulation steps total_steps: Total number of steps in flow coverage_rate: Coverage rate (0.0 to 1.0) uncovered_steps: Set of uncovered step numbers Returns: List of issue descriptions """ issues: list[str] = [] if coverage_rate < self.MIN_COVERAGE_THRESHOLD: issues.append( f"流程覆盖率低于 {int(self.MIN_COVERAGE_THRESHOLD * 100)}%，建议增加测试用例" ) if len(simulation) > total_steps * self.MAX_SIMULATION_STEPS_MULTIPLIER: issues.append("检测到可能的死循环") if uncovered_steps: issues.append(f"未覆盖步骤：{sorted(list(uncovered_steps))}") step_visit_count: dict[int, int] = {} for step in simulation: step_visit_count[step.stepNo] = step_visit_count.get(step.stepNo, 0) + 1 for step_no, count in step_visit_count.items(): if count > 3: issues.append(f"步骤 {step_no} 被重复访问 {count} 次，可能存在逻辑问题") return issues def to_dict(self, result: FlowSimulationResult) -> dict[str, Any]: """Convert simulation result to API response dict.""" simulation_data = [] for step in result.simulation: step_data = { "stepNo": step.stepNo, "botMessage": step.botMessage, "userInput": step.userInput, "nextStep": step.nextStep, "durationMs": step.durationMs, } if step.matchedCondition: step_data["matchedCondition"] = { "type": step.matchedCondition.type, "gotoStep": step.matchedCondition.goto_step, } if step.matchedCondition.keywords: step_data["matchedCondition"]["keywords"] = step.matchedCondition.keywords if step.matchedCondition.pattern: step_data["matchedCondition"]["pattern"] = step.matchedCondition.pattern else: step_data["matchedCondition"] = None simulation_data.append(step_data) return { "flowId": result.flowId, "flowName": result.flowName, "simulation": simulation_data, "result": result.result, "coverage": result.coverage, "issues": result.issues, }