/
devsec
/
anothertap
Обзор
Документация
Войти
/
devsec
/
anothertap
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/gui/execution_widget.py
342 строки
13 KB
devsec
init project
23 авг 2025, 21:01
23 авг 2025, 21:01
5b0e5af
Код
Авторство
О чём код?
""" Execution Widget for running test plans with real-time progress tracking """ from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QFormLayout, QLineEdit, QTextEdit, QPushButton, QLabel, QProgressBar, QTableWidget, QTableWidgetItem, QHeaderView, QSplitter, QComboBox, QMessageBox ) from PySide6.QtCore import Qt, QTimer, Signal from PySide6.QtGui import QFont from models.test_plan import TestPlan, PlanStatus from models.test_step import TestStep, StepStatus from services.test_plan_manager import TestPlanManager from services.device_manager import DeviceManager from services.execution_engine import ExecutionEngine class ExecutionWidget(QWidget): """Widget for executing test plans with real-time monitoring""" execution_started = Signal(str) # plan_id execution_finished = Signal(str, bool) # plan_id, success def __init__(self, test_plan_manager: TestPlanManager, device_manager: DeviceManager): super().__init__() self.test_plan_manager = test_plan_manager self.device_manager = device_manager # Create execution engine self.execution_engine = ExecutionEngine(device_manager, test_plan_manager) self.setup_connections() self.current_plan = None self.setup_ui() self.refresh_test_plans() # Update timer for real-time updates self.update_timer = QTimer() self.update_timer.timeout.connect(self.update_display) self.update_timer.start(1000) # Update every second def setup_ui(self): """Set up the user interface""" layout = QVBoxLayout(self) # Test plan selection selection_group = QGroupBox("Test Plan Selection") selection_layout = QFormLayout(selection_group) layout.addWidget(selection_group) self.plan_combo = QComboBox() self.plan_combo.currentTextChanged.connect(self.on_plan_selected) selection_layout.addRow("Test Plan:", self.plan_combo) # Execution controls controls_group = QGroupBox("Execution Controls") controls_layout = QHBoxLayout(controls_group) layout.addWidget(controls_group) self.start_btn = QPushButton("Start Execution") self.start_btn.clicked.connect(self.start_execution) controls_layout.addWidget(self.start_btn) self.pause_btn = QPushButton("Pause") self.pause_btn.clicked.connect(self.pause_execution) self.pause_btn.setEnabled(False) controls_layout.addWidget(self.pause_btn) self.stop_btn = QPushButton("Stop") self.stop_btn.clicked.connect(self.stop_execution) self.stop_btn.setEnabled(False) controls_layout.addWidget(self.stop_btn) controls_layout.addStretch() # Progress information progress_group = QGroupBox("Execution Progress") progress_layout = QFormLayout(progress_group) layout.addWidget(progress_group) self.status_label = QLabel("Idle") progress_layout.addRow("Status:", self.status_label) self.progress_bar = QProgressBar() progress_layout.addRow("Progress:", self.progress_bar) self.current_step_label = QLabel("None") progress_layout.addRow("Current Step:", self.current_step_label) # Create splitter for results splitter = QSplitter(Qt.Vertical) layout.addWidget(splitter) # Step results table self.setup_results_table(splitter) # Execution log self.setup_execution_log(splitter) # Set splitter proportions splitter.setSizes([400, 200]) def setup_results_table(self, parent): """Set up the results table""" results_group = QGroupBox("Step Results") results_layout = QVBoxLayout(results_group) parent.addWidget(results_group) self.results_table = QTableWidget() self.results_table.setColumnCount(6) self.results_table.setHorizontalHeaderLabels([ "Step", "Status", "Duration", "Response", "Error", "Timestamp" ]) # Set column widths header = self.results_table.horizontalHeader() header.setSectionResizeMode(0, QHeaderView.Stretch) # Step header.setSectionResizeMode(1, QHeaderView.ResizeToContents) # Status header.setSectionResizeMode(2, QHeaderView.ResizeToContents) # Duration header.setSectionResizeMode(3, QHeaderView.Stretch) # Response header.setSectionResizeMode(4, QHeaderView.Stretch) # Error header.setSectionResizeMode(5, QHeaderView.ResizeToContents) # Timestamp results_layout.addWidget(self.results_table) def setup_execution_log(self, parent): """Set up the execution log""" log_group = QGroupBox("Execution Log") log_layout = QVBoxLayout(log_group) parent.addWidget(log_group) self.log_text = QTextEdit() self.log_text.setReadOnly(True) font = QFont("Consolas", 9) font.setStyleHint(QFont.Monospace) self.log_text.setFont(font) log_layout.addWidget(self.log_text) # Log controls log_controls_layout = QHBoxLayout() log_layout.addLayout(log_controls_layout) self.clear_log_btn = QPushButton("Clear Log") self.clear_log_btn.clicked.connect(self.clear_log) log_controls_layout.addWidget(self.clear_log_btn) log_controls_layout.addStretch() def setup_connections(self): """Set up signal connections with execution engine""" self.execution_engine.execution_started.connect(self.on_execution_started) self.execution_engine.execution_finished.connect(self.on_execution_finished) self.execution_engine.execution_paused.connect(self.on_execution_paused) self.execution_engine.execution_resumed.connect(self.on_execution_resumed) self.execution_engine.execution_aborted.connect(self.on_execution_aborted) self.execution_engine.step_started.connect(self.on_step_started) self.execution_engine.step_finished.connect(self.on_step_finished) self.execution_engine.step_progress.connect(self.on_step_progress) self.execution_engine.plan_progress.connect(self.on_plan_progress) self.execution_engine.log_message.connect(self.on_log_message) def refresh_test_plans(self): """Refresh the test plan combo box""" self.plan_combo.clear() self.plan_combo.addItem("(Select Test Plan)", "") plans = self.test_plan_manager.get_all_test_plans() for plan in plans: self.plan_combo.addItem(plan.display_name, plan.id) def on_plan_selected(self): """Handle test plan selection""" plan_id = self.plan_combo.currentData() if plan_id: self.current_plan = self.test_plan_manager.get_test_plan(plan_id) self.load_test_plan_steps() else: self.current_plan = None self.results_table.setRowCount(0) def load_test_plan_steps(self): """Load test plan steps into results table""" if not self.current_plan: return all_steps = self.current_plan.all_steps self.results_table.setRowCount(len(all_steps)) for i, step in enumerate(all_steps): self.results_table.setItem(i, 0, QTableWidgetItem(step.display_name)) self.results_table.setItem(i, 1, QTableWidgetItem(step.status.value)) duration = f"{step.execution_time:.2f}s" if step.execution_time else "" self.results_table.setItem(i, 2, QTableWidgetItem(duration)) self.results_table.setItem(i, 3, QTableWidgetItem(step.response or "")) self.results_table.setItem(i, 4, QTableWidgetItem(step.error_message or "")) self.results_table.setItem(i, 5, QTableWidgetItem(step.timestamp or "")) # Store step ID for updates self.results_table.item(i, 0).setData(Qt.UserRole, step.id) # Color code by status if step.status == StepStatus.PASSED: for col in range(6): if self.results_table.item(i, col): self.results_table.item(i, col).setBackground(Qt.green) elif step.status == StepStatus.FAILED: for col in range(6): if self.results_table.item(i, col): self.results_table.item(i, col).setBackground(Qt.red) def load_test_plan(self, plan_id: str): """Load a specific test plan for execution""" # Set the combo box to the specified plan for i in range(self.plan_combo.count()): if self.plan_combo.itemData(i) == plan_id: self.plan_combo.setCurrentIndex(i) break def start_execution(self): """Start test plan execution""" if not self.current_plan: QMessageBox.warning(self, "Warning", "Please select a test plan first") return if self.execution_engine.start_execution(self.current_plan.id): self.start_btn.setEnabled(False) self.pause_btn.setEnabled(True) self.stop_btn.setEnabled(True) self.log_message("INFO", f"Starting execution of '{self.current_plan.name}'") else: QMessageBox.warning(self, "Error", "Failed to start execution") def pause_execution(self): """Pause test plan execution""" if self.execution_engine.pause_execution(): self.pause_btn.setText("Resume") self.pause_btn.clicked.disconnect() self.pause_btn.clicked.connect(self.resume_execution) def resume_execution(self): """Resume test plan execution""" if self.execution_engine.resume_execution(): self.pause_btn.setText("Pause") self.pause_btn.clicked.disconnect() self.pause_btn.clicked.connect(self.pause_execution) def stop_execution(self): """Stop test plan execution""" if self.execution_engine.stop_execution(): self.log_message("WARNING", "Execution stop requested") def update_display(self): """Update the display with current execution state""" if self.current_plan: # Update progress bar progress = self.current_plan.progress_percentage self.progress_bar.setValue(int(progress)) # Update status self.status_label.setText(self.current_plan.status.value) # Refresh results table if self.execution_engine.is_running: self.load_test_plan_steps() def clear_log(self): """Clear the execution log""" self.log_text.clear() def log_message(self, level: str, message: str): """Add a message to the execution log""" from datetime import datetime timestamp = datetime.now().strftime("%H:%M:%S") formatted_message = f"[{timestamp}] {level}: {message}" self.log_text.append(formatted_message) # Auto-scroll to bottom scrollbar = self.log_text.verticalScrollBar() scrollbar.setValue(scrollbar.maximum()) # Execution engine signal handlers def on_execution_started(self, plan_id: str): """Handle execution started signal""" self.status_label.setText("Running") self.execution_started.emit(plan_id) def on_execution_finished(self, plan_id: str, success: bool): """Handle execution finished signal""" self.start_btn.setEnabled(True) self.pause_btn.setEnabled(False) self.stop_btn.setEnabled(False) self.pause_btn.setText("Pause") status = "Completed Successfully" if success else "Failed" self.status_label.setText(status) self.current_step_label.setText("None") self.execution_finished.emit(plan_id, success) self.load_test_plan_steps() # Final update def on_execution_paused(self, plan_id: str): """Handle execution paused signal""" self.status_label.setText("Paused") def on_execution_resumed(self, plan_id: str): """Handle execution resumed signal""" self.status_label.setText("Running") def on_execution_aborted(self, plan_id: str): """Handle execution aborted signal""" self.status_label.setText("Aborted") def on_step_started(self, plan_id: str, step_id: str): """Handle step started signal""" if self.current_plan: step = self.current_plan.get_step_by_id(step_id) if step: self.current_step_label.setText(step.display_name) def on_step_finished(self, plan_id: str, step_id: str, success: bool): """Handle step finished signal""" # Results table will be updated by update_display pass def on_step_progress(self, plan_id: str, step_id: str, message: str): """Handle step progress signal""" self.log_message("DEBUG", f"Step progress: {message}") def on_plan_progress(self, plan_id: str, percentage: float): """Handle plan progress signal""" self.progress_bar.setValue(int(percentage)) def on_log_message(self, level: str, message: str): """Handle log message signal""" self.log_message(level, message)