src/models/test_plan.py - master - devsec/anothertap

src/models/test_plan.py
264 строки10 KB
devsec
init project
23 авг 2025, 21:01
23 авг 2025, 21:015b0e5af
"""
Test plan model for organizing and managing sequences of test steps
"""

from dataclasses import dataclass, field
from typing import Dict, Any, Optional, List
from dataclasses_json import dataclass_json
from enum import Enum
import uuid
from datetime import datetime

from models.test_step import TestStep, StepStatus


class PlanStatus(Enum):
    """Execution status of a test plan"""
    IDLE = "IDLE"
    RUNNING = "RUNNING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"
    ABORTED = "ABORTED"
    PAUSED = "PAUSED"


@dataclass_json
@dataclass
class TestPlan:
    """
    Represents a test plan containing multiple test steps
    """
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    description: str = ""
    version: str = "1.0.0"
    author: str = ""
    created_date: str = field(default_factory=lambda: datetime.now().isoformat())
    modified_date: str = field(default_factory=lambda: datetime.now().isoformat())
    
    # Test configuration
    steps: List[TestStep] = field(default_factory=list)
    setup_steps: List[TestStep] = field(default_factory=list)  # Run before main steps
    teardown_steps: List[TestStep] = field(default_factory=list)  # Run after main steps
    
    # Execution settings
    stop_on_failure: bool = True
    max_execution_time: Optional[float] = None  # Maximum total execution time
    repeat_count: int = 1  # Number of times to repeat the test plan
    parallel_execution: bool = False  # Execute steps in parallel where possible
    
    # Metadata and tags
    tags: List[str] = field(default_factory=list)
    category: str = ""
    priority: str = "Normal"  # Low, Normal, High, Critical
    requirements: List[str] = field(default_factory=list)  # Required devices, software, etc.
    custom_properties: Dict[str, Any] = field(default_factory=dict)
    
    # Execution state
    status: PlanStatus = PlanStatus.IDLE
    current_step_index: int = 0
    start_time: Optional[str] = None
    end_time: Optional[str] = None
    total_execution_time: Optional[float] = None
    current_iteration: int = 0
    
    # Results
    passed_steps: int = 0
    failed_steps: int = 0
    skipped_steps: int = 0
    error_message: Optional[str] = None
    execution_log: List[Dict[str, Any]] = field(default_factory=list)
    
    def __post_init__(self):
        """Post-initialization processing"""
        if isinstance(self.status, str):
            self.status = PlanStatus(self.status)
        
        # Ensure steps are TestStep objects
        self.steps = [step if isinstance(step, TestStep) else TestStep.from_dict(step) 
                     for step in self.steps]
        self.setup_steps = [step if isinstance(step, TestStep) else TestStep.from_dict(step) 
                           for step in self.setup_steps]
        self.teardown_steps = [step if isinstance(step, TestStep) else TestStep.from_dict(step) 
                              for step in self.teardown_steps]
    
    @property
    def display_name(self) -> str:
        """Get a user-friendly display name for the test plan"""
        return self.name if self.name else f"Test Plan {self.id[:8]}"
    
    @property
    def total_steps(self) -> int:
        """Get total number of steps in the plan"""
        return len(self.setup_steps) + len(self.steps) + len(self.teardown_steps)
    
    @property
    def all_steps(self) -> List[TestStep]:
        """Get all steps including setup and teardown"""
        return self.setup_steps + self.steps + self.teardown_steps
    
    @property
    def progress_percentage(self) -> float:
        """Get execution progress as percentage"""
        if not self.all_steps:
            return 0.0
        completed = sum(1 for step in self.all_steps 
                       if step.status in [StepStatus.PASSED, StepStatus.FAILED, StepStatus.SKIPPED])
        return (completed / len(self.all_steps)) * 100
    
    @property
    def success_rate(self) -> float:
        """Get success rate as percentage"""
        if not self.all_steps:
            return 0.0
        executed = sum(1 for step in self.all_steps 
                      if step.status in [StepStatus.PASSED, StepStatus.FAILED])
        if executed == 0:
            return 0.0
        return (self.passed_steps / executed) * 100
    
    def add_step(self, step: TestStep, index: Optional[int] = None):
        """Add a step to the test plan"""
        if index is None:
            self.steps.append(step)
        else:
            self.steps.insert(index, step)
        self.update_modified_date()
    
    def remove_step(self, step_id: str) -> bool:
        """Remove a step from the test plan"""
        for step_list in [self.setup_steps, self.steps, self.teardown_steps]:
            for i, step in enumerate(step_list):
                if step.id == step_id:
                    step_list.pop(i)
                    self.update_modified_date()
                    return True
        return False
    
    def get_step_by_id(self, step_id: str) -> Optional[TestStep]:
        """Get a step by its ID"""
        for step in self.all_steps:
            if step.id == step_id:
                return step
        return None
    
    def reset_execution_state(self):
        """Reset execution state for re-running"""
        self.status = PlanStatus.IDLE
        self.current_step_index = 0
        self.start_time = None
        self.end_time = None
        self.total_execution_time = None
        self.current_iteration = 0
        self.passed_steps = 0
        self.failed_steps = 0
        self.skipped_steps = 0
        self.error_message = None
        self.execution_log.clear()
        
        # Reset all steps
        for step in self.all_steps:
            step.reset_execution_state()
    
    def update_modified_date(self):
        """Update the modified date to current time"""
        self.modified_date = datetime.now().isoformat()
    
    def get_required_devices(self) -> List[str]:
        """Get list of device IDs required by this test plan"""
        device_ids = set()
        for step in self.all_steps:
            if step.device_id:
                device_ids.add(step.device_id)
        return list(device_ids)
    
    def validate_plan(self) -> List[str]:
        """Validate the test plan and return list of issues"""
        issues = []
        
        if not self.name:
            issues.append("Test plan name is required")
        
        if not self.steps:
            issues.append("Test plan must contain at least one test step")
        
        # Check for steps without device assignment
        for i, step in enumerate(self.all_steps):
            if not step.device_id:
                issues.append(f"Step {i+1} '{step.display_name}' has no device assigned")
            if not step.scpi_command and step.step_type.value.startswith("SCPI"):
                issues.append(f"Step {i+1} '{step.display_name}' has no SCPI command")
        
        return issues
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert test plan to dictionary for JSON serialization"""
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'version': self.version,
            'author': self.author,
            'created_date': self.created_date,
            'modified_date': self.modified_date,
            'steps': [step.to_dict() for step in self.steps],
            'setup_steps': [step.to_dict() for step in self.setup_steps],
            'teardown_steps': [step.to_dict() for step in self.teardown_steps],
            'stop_on_failure': self.stop_on_failure,
            'max_execution_time': self.max_execution_time,
            'repeat_count': self.repeat_count,
            'parallel_execution': self.parallel_execution,
            'tags': self.tags,
            'category': self.category,
            'priority': self.priority,
            'requirements': self.requirements,
            'custom_properties': self.custom_properties,
            'status': self.status.value,
            'current_step_index': self.current_step_index,
            'start_time': self.start_time,
            'end_time': self.end_time,
            'total_execution_time': self.total_execution_time,
            'current_iteration': self.current_iteration,
            'passed_steps': self.passed_steps,
            'failed_steps': self.failed_steps,
            'skipped_steps': self.skipped_steps,
            'error_message': self.error_message,
            'execution_log': self.execution_log
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'TestPlan':
        """Create test plan from dictionary"""
        plan = cls(
            id=data.get('id', str(uuid.uuid4())),
            name=data.get('name', ''),
            description=data.get('description', ''),
            version=data.get('version', '1.0.0'),
            author=data.get('author', ''),
            created_date=data.get('created_date', datetime.now().isoformat()),
            modified_date=data.get('modified_date', datetime.now().isoformat()),
            steps=[TestStep.from_dict(step_data) for step_data in data.get('steps', [])],
            setup_steps=[TestStep.from_dict(step_data) for step_data in data.get('setup_steps', [])],
            teardown_steps=[TestStep.from_dict(step_data) for step_data in data.get('teardown_steps', [])],
            stop_on_failure=data.get('stop_on_failure', True),
            max_execution_time=data.get('max_execution_time'),
            repeat_count=data.get('repeat_count', 1),
            parallel_execution=data.get('parallel_execution', False),
            tags=data.get('tags', []),
            category=data.get('category', ''),
            priority=data.get('priority', 'Normal'),
            requirements=data.get('requirements', []),
            custom_properties=data.get('custom_properties', {}),
            status=PlanStatus(data.get('status', PlanStatus.IDLE.value)),
            current_step_index=data.get('current_step_index', 0),
            start_time=data.get('start_time'),
            end_time=data.get('end_time'),
            total_execution_time=data.get('total_execution_time'),
            current_iteration=data.get('current_iteration', 0),
            passed_steps=data.get('passed_steps', 0),
            failed_steps=data.get('failed_steps', 0),
            skipped_steps=data.get('skipped_steps', 0),
            error_message=data.get('error_message'),
            execution_log=data.get('execution_log', [])
        )
        return plan