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

src/models/test_step.py
160 строк6 KB
devsec
init project
23 авг 2025, 21:01
23 авг 2025, 21:015b0e5af
"""
Test step model for individual SCPI commands and operations
"""

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


class StepType(Enum):
    """Types of test steps"""
    SCPI_COMMAND = "SCPI_COMMAND"
    SCPI_QUERY = "SCPI_QUERY"
    DELAY = "DELAY"
    CONDITION = "CONDITION"
    LOOP = "LOOP"


class StepStatus(Enum):
    """Execution status of a test step"""
    PENDING = "PENDING"
    RUNNING = "RUNNING"
    PASSED = "PASSED"
    FAILED = "FAILED"
    SKIPPED = "SKIPPED"


@dataclass_json
@dataclass
class TestStep:
    """
    Represents a single test step with SCPI command or other operation
    """
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    description: str = ""
    device_id: str = ""  # Reference to the device this step targets
    step_type: StepType = StepType.SCPI_COMMAND
    scpi_command: str = ""  # The actual SCPI command to execute
    expected_response: Optional[str] = None  # For queries, expected response pattern
    timeout: float = 5.0  # Timeout in seconds
    retry_count: int = 0  # Number of retries on failure
    delay_before: float = 0.0  # Delay before executing this step (seconds)
    delay_after: float = 0.0  # Delay after executing this step (seconds)
    enabled: bool = True
    parameters: Dict[str, Any] = field(default_factory=dict)  # Step-specific parameters
    validation_rules: List[Dict[str, Any]] = field(default_factory=list)  # Response validation
    
    # Execution results
    status: StepStatus = StepStatus.PENDING
    execution_time: Optional[float] = None  # Actual execution time
    response: Optional[str] = None  # Actual response from device
    error_message: Optional[str] = None
    timestamp: Optional[str] = None
    
    def __post_init__(self):
        """Post-initialization processing"""
        if isinstance(self.step_type, str):
            self.step_type = StepType(self.step_type)
        if isinstance(self.status, str):
            self.status = StepStatus(self.status)
    
    @property
    def display_name(self) -> str:
        """Get a user-friendly display name for the step"""
        if self.name:
            return self.name
        elif self.scpi_command:
            return f"SCPI: {self.scpi_command[:30]}..."
        else:
            return f"Step {self.id[:8]}"
    
    def is_query(self) -> bool:
        """Check if this step is a query (expects response)"""
        return self.step_type == StepType.SCPI_QUERY or self.scpi_command.endswith("?")
    
    def reset_execution_state(self):
        """Reset execution state for re-running"""
        self.status = StepStatus.PENDING
        self.execution_time = None
        self.response = None
        self.error_message = None
        self.timestamp = None
    
    def validate_response(self, response: str) -> bool:
        """Validate response against expected patterns and rules"""
        if self.expected_response and response != self.expected_response:
            return False
        
        # Apply validation rules
        for rule in self.validation_rules:
            rule_type = rule.get("type", "")
            if rule_type == "contains":
                if rule["value"] not in response:
                    return False
            elif rule_type == "regex":
                import re
                if not re.match(rule["pattern"], response):
                    return False
            elif rule_type == "numeric_range":
                try:
                    value = float(response.strip())
                    if not (rule["min"] <= value <= rule["max"]):
                        return False
                except ValueError:
                    return False
        
        return True
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert test step to dictionary for JSON serialization"""
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'device_id': self.device_id,
            'step_type': self.step_type.value,
            'scpi_command': self.scpi_command,
            'expected_response': self.expected_response,
            'timeout': self.timeout,
            'retry_count': self.retry_count,
            'delay_before': self.delay_before,
            'delay_after': self.delay_after,
            'enabled': self.enabled,
            'parameters': self.parameters,
            'validation_rules': self.validation_rules,
            'status': self.status.value,
            'execution_time': self.execution_time,
            'response': self.response,
            'error_message': self.error_message,
            'timestamp': self.timestamp
        }
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'TestStep':
        """Create test step from dictionary"""
        step = cls(
            id=data.get('id', str(uuid.uuid4())),
            name=data.get('name', ''),
            description=data.get('description', ''),
            device_id=data.get('device_id', ''),
            step_type=StepType(data.get('step_type', StepType.SCPI_COMMAND.value)),
            scpi_command=data.get('scpi_command', ''),
            expected_response=data.get('expected_response'),
            timeout=data.get('timeout', 5.0),
            retry_count=data.get('retry_count', 0),
            delay_before=data.get('delay_before', 0.0),
            delay_after=data.get('delay_after', 0.0),
            enabled=data.get('enabled', True),
            parameters=data.get('parameters', {}),
            validation_rules=data.get('validation_rules', []),
            status=StepStatus(data.get('status', StepStatus.PENDING.value)),
            execution_time=data.get('execution_time'),
            response=data.get('response'),
            error_message=data.get('error_message'),
            timestamp=data.get('timestamp')
        )
        return step