/
devsec
/
anothertap
Обзор
Документация
Войти
/
devsec
/
anothertap
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/services/execution_engine.py
429 строк
16 KB
devsec
init project
23 авг 2025, 21:01
23 авг 2025, 21:01
5b0e5af
Код
Авторство
О чём код?
""" Test Plan Execution Engine with progress tracking and device coordination """ import time import logging from typing import Optional, Dict, Any, Callable from datetime import datetime from threading import Thread, Lock, Event from PySide6.QtCore import QObject, Signal from models.test_plan import TestPlan, PlanStatus from models.test_step import TestStep, StepStatus, StepType from models.device import Device from services.device_manager import DeviceManager from services.test_plan_manager import TestPlanManager class ExecutionEngine(QObject): """ Engine for executing test plans with device coordination and progress tracking """ # Signals for GUI updates execution_started = Signal(str) # plan_id execution_finished = Signal(str, bool) # plan_id, success execution_paused = Signal(str) # plan_id execution_resumed = Signal(str) # plan_id execution_aborted = Signal(str) # plan_id step_started = Signal(str, str) # plan_id, step_id step_finished = Signal(str, str, bool) # plan_id, step_id, success step_progress = Signal(str, str, str) # plan_id, step_id, message plan_progress = Signal(str, float) # plan_id, percentage log_message = Signal(str, str) # level, message def __init__(self, device_manager: DeviceManager, test_plan_manager: TestPlanManager): super().__init__() self.device_manager = device_manager self.test_plan_manager = test_plan_manager self.logger = logging.getLogger(__name__) # Execution state self._execution_thread: Optional[Thread] = None self._execution_lock = Lock() self._stop_event = Event() self._pause_event = Event() self._current_plan: Optional[TestPlan] = None self._is_running = False # Step execution callbacks self._step_handlers: Dict[StepType, Callable] = { StepType.SCPI_COMMAND: self._execute_scpi_command, StepType.SCPI_QUERY: self._execute_scpi_query, StepType.DELAY: self._execute_delay, StepType.CONDITION: self._execute_condition, StepType.LOOP: self._execute_loop } @property def is_running(self) -> bool: """Check if execution is currently running""" return self._is_running @property def current_plan(self) -> Optional[TestPlan]: """Get currently executing plan""" return self._current_plan def start_execution(self, plan_id: str) -> bool: """ Start executing a test plan Args: plan_id: ID of the test plan to execute Returns: True if execution started successfully, False otherwise """ with self._execution_lock: if self._is_running: self.logger.warning("Execution already running") return False plan = self.test_plan_manager.get_test_plan(plan_id) if not plan: self.logger.error(f"Test plan {plan_id} not found") return False # Validate plan before execution issues = plan.validate_plan() if issues: self.logger.error(f"Test plan validation failed: {issues}") self.log_message.emit("ERROR", f"Test plan validation failed: {', '.join(issues)}") return False # Check required devices required_devices = plan.get_required_devices() for device_id in required_devices: device = self.device_manager.get_device(device_id) if not device: self.logger.error(f"Required device {device_id} not found") self.log_message.emit("ERROR", f"Required device {device_id} not found") return False if not device.is_connected(): # Try to connect if not self.device_manager.connect_device(device_id): self.logger.error(f"Failed to connect to device {device.display_name}") self.log_message.emit("ERROR", f"Failed to connect to device {device.display_name}") return False # Reset execution state plan.reset_execution_state() self._current_plan = plan self._stop_event.clear() self._pause_event.clear() self._is_running = True # Start execution thread self._execution_thread = Thread(target=self._execute_plan, daemon=True) self._execution_thread.start() self.execution_started.emit(plan_id) self.logger.info(f"Started execution of test plan: {plan.display_name}") return True def stop_execution(self) -> bool: """ Stop current execution Returns: True if execution was stopped, False otherwise """ if not self._is_running: return False self._stop_event.set() self.logger.info("Execution stop requested") return True def pause_execution(self) -> bool: """ Pause current execution Returns: True if execution was paused, False otherwise """ if not self._is_running or self._pause_event.is_set(): return False self._pause_event.set() if self._current_plan: self._current_plan.status = PlanStatus.PAUSED self.execution_paused.emit(self._current_plan.id) self.logger.info("Execution paused") return True def resume_execution(self) -> bool: """ Resume paused execution Returns: True if execution was resumed, False otherwise """ if not self._is_running or not self._pause_event.is_set(): return False self._pause_event.clear() if self._current_plan: self._current_plan.status = PlanStatus.RUNNING self.execution_resumed.emit(self._current_plan.id) self.logger.info("Execution resumed") return True def _execute_plan(self): """Execute the current test plan (runs in separate thread)""" if not self._current_plan: return plan = self._current_plan success = True try: plan.status = PlanStatus.RUNNING plan.start_time = datetime.now().isoformat() self.log_message.emit("INFO", f"Starting execution of '{plan.display_name}'") # Execute for each iteration for iteration in range(plan.repeat_count): if self._stop_event.is_set(): break plan.current_iteration = iteration + 1 self.log_message.emit("INFO", f"Starting iteration {plan.current_iteration}/{plan.repeat_count}") # Execute setup steps if not self._execute_step_list(plan.setup_steps, "Setup"): if plan.stop_on_failure: success = False break # Execute main steps if not self._execute_step_list(plan.steps, "Main"): if plan.stop_on_failure: success = False break # Execute teardown steps (always run, even on failure) self._execute_step_list(plan.teardown_steps, "Teardown") # Update final status if self._stop_event.is_set(): plan.status = PlanStatus.ABORTED self.execution_aborted.emit(plan.id) self.log_message.emit("WARNING", f"Execution aborted: {plan.display_name}") elif success: plan.status = PlanStatus.COMPLETED self.log_message.emit("INFO", f"Execution completed successfully: {plan.display_name}") else: plan.status = PlanStatus.FAILED self.log_message.emit("ERROR", f"Execution failed: {plan.display_name}") plan.end_time = datetime.now().isoformat() # Calculate total execution time if plan.start_time and plan.end_time: start = datetime.fromisoformat(plan.start_time) end = datetime.fromisoformat(plan.end_time) plan.total_execution_time = (end - start).total_seconds() # Save plan state self.test_plan_manager.update_test_plan(plan) self.execution_finished.emit(plan.id, success) except Exception as e: self.logger.error(f"Execution error: {e}") plan.status = PlanStatus.FAILED plan.error_message = str(e) plan.end_time = datetime.now().isoformat() self.test_plan_manager.update_test_plan(plan) self.execution_finished.emit(plan.id, False) self.log_message.emit("ERROR", f"Execution error: {e}") finally: self._is_running = False self._current_plan = None def _execute_step_list(self, steps: list[TestStep], phase_name: str) -> bool: """Execute a list of steps""" if not steps: return True self.log_message.emit("INFO", f"Executing {phase_name} phase ({len(steps)} steps)") success = True for i, step in enumerate(steps): if self._stop_event.is_set(): break # Handle pause while self._pause_event.is_set() and not self._stop_event.is_set(): time.sleep(0.1) if not step.enabled: step.status = StepStatus.SKIPPED self._current_plan.skipped_steps += 1 continue step_success = self._execute_step(step) if step_success: self._current_plan.passed_steps += 1 else: self._current_plan.failed_steps += 1 success = False if self._current_plan.stop_on_failure: self.log_message.emit("WARNING", f"Stopping execution due to step failure: {step.display_name}") break # Update progress progress = ((i + 1) / len(steps)) * 100 self.plan_progress.emit(self._current_plan.id, self._current_plan.progress_percentage) return success def _execute_step(self, step: TestStep) -> bool: """Execute a single test step""" step.status = StepStatus.RUNNING step.timestamp = datetime.now().isoformat() step.response = None step.error_message = None self.step_started.emit(self._current_plan.id, step.id) self.step_progress.emit(self._current_plan.id, step.id, f"Executing: {step.display_name}") start_time = time.time() try: # Pre-execution delay if step.delay_before > 0: self.step_progress.emit(self._current_plan.id, step.id, f"Waiting {step.delay_before}s before execution") time.sleep(step.delay_before) # Execute step based on type handler = self._step_handlers.get(step.step_type) if not handler: raise ValueError(f"Unknown step type: {step.step_type}") success = handler(step) # Post-execution delay if step.delay_after > 0: self.step_progress.emit(self._current_plan.id, step.id, f"Waiting {step.delay_after}s after execution") time.sleep(step.delay_after) step.execution_time = time.time() - start_time if success: step.status = StepStatus.PASSED self.step_progress.emit(self._current_plan.id, step.id, "Completed successfully") self.log_message.emit("INFO", f"Step passed: {step.display_name}") else: step.status = StepStatus.FAILED self.step_progress.emit(self._current_plan.id, step.id, "Failed") self.log_message.emit("ERROR", f"Step failed: {step.display_name}") self.step_finished.emit(self._current_plan.id, step.id, success) return success except Exception as e: step.execution_time = time.time() - start_time step.status = StepStatus.FAILED step.error_message = str(e) self.step_progress.emit(self._current_plan.id, step.id, f"Error: {str(e)}") self.step_finished.emit(self._current_plan.id, step.id, False) self.log_message.emit("ERROR", f"Step error: {step.display_name} - {str(e)}") return False def _execute_scpi_command(self, step: TestStep) -> bool: """Execute SCPI command step""" device = self.device_manager.get_device(step.device_id) if not device: step.error_message = f"Device {step.device_id} not found" return False if not device.is_connected(): step.error_message = f"Device {device.display_name} not connected" return False success, response = self.device_manager.scpi_service.send_command( device, step.scpi_command, step.timeout ) if not success: step.error_message = f"Failed to send command: {step.scpi_command}" return False # SCPI commands don't expect responses return True def _execute_scpi_query(self, step: TestStep) -> bool: """Execute SCPI query step""" device = self.device_manager.get_device(step.device_id) if not device: step.error_message = f"Device {step.device_id} not found" return False if not device.is_connected(): step.error_message = f"Device {device.display_name} not connected" return False success, response = self.device_manager.scpi_service.send_command( device, step.scpi_command, step.timeout ) if not success: step.error_message = f"Failed to send query: {step.scpi_command}" return False if response is None: step.error_message = "No response received from query" return False step.response = response # Validate response if expected response is specified if step.expected_response and response != step.expected_response: step.error_message = f"Response mismatch. Expected: '{step.expected_response}', Got: '{response}'" return False # Apply validation rules if not step.validate_response(response): step.error_message = f"Response validation failed: '{response}'" return False return True def _execute_delay(self, step: TestStep) -> bool: """Execute delay step""" delay_time = step.parameters.get('delay_time', 1.0) self.step_progress.emit(self._current_plan.id, step.id, f"Waiting {delay_time}s") # Implement interruptible delay for _ in range(int(delay_time * 10)): # 100ms intervals if self._stop_event.is_set(): return False time.sleep(0.1) return True def _execute_condition(self, step: TestStep) -> bool: """Execute condition step""" # This would implement conditional logic based on parameters # For now, just return True self.step_progress.emit(self._current_plan.id, step.id, "Condition evaluation not implemented") return True def _execute_loop(self, step: TestStep) -> bool: """Execute loop step""" # This would implement loop logic based on parameters # For now, just return True self.step_progress.emit(self._current_plan.id, step.id, "Loop execution not implemented") return True