tests/test_core_services.py - master - devsec/anothertap

tests/test_core_services.py
345 строк12 KB
devsec
init project
23 авг 2025, 21:01
23 авг 2025, 21:015b0e5af
"""
Unit tests for core services of the Test Automation Platform
"""

import pytest
import tempfile
import json
from pathlib import Path
from unittest.mock import Mock, patch

# Import the modules to test
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from models.device import Device, ConnectionType, DeviceStatus
from models.test_step import TestStep, StepType, StepStatus
from models.test_plan import TestPlan, PlanStatus
from services.device_manager import DeviceManager
from services.test_plan_manager import TestPlanManager
from services.config_manager import ConfigManager


class TestDeviceModel:
    """Test the Device model"""
    
    def test_device_creation(self):
        """Test basic device creation"""
        device = Device(
            name="Test Device",
            manufacturer="Test Corp",
            model="TD-1000",
            connection_type=ConnectionType.TCPIP,
            connection_string="TCPIP::192.168.1.100::INSTR"
        )
        
        assert device.name == "Test Device"
        assert device.manufacturer == "Test Corp"
        assert device.model == "TD-1000"
        assert device.connection_type == ConnectionType.TCPIP
        assert device.status == DeviceStatus.DISCONNECTED
    
    def test_device_display_name(self):
        """Test device display name property"""
        # Test with name
        device1 = Device(name="My Device")
        assert device1.display_name == "My Device"
        
        # Test with manufacturer and model
        device2 = Device(manufacturer="Keysight", model="34461A")
        assert device2.display_name == "Keysight 34461A"
        
        # Test with ID fallback
        device3 = Device()
        assert device3.display_name.startswith("Device")
    
    def test_device_is_connected(self):
        """Test device connection status"""
        device = Device()
        assert not device.is_connected()
        
        device.status = DeviceStatus.CONNECTED
        assert device.is_connected()
    
    def test_device_serialization(self):
        """Test device JSON serialization"""
        device = Device(
            name="Test Device",
            manufacturer="Test Corp",
            connection_type=ConnectionType.USB,
            connection_string="USB0::0x1234::0x5678::SN123::INSTR"
        )
        
        # Test to_dict
        device_dict = device.to_dict()
        assert device_dict["name"] == "Test Device"
        assert device_dict["manufacturer"] == "Test Corp"
        assert device_dict["connection_type"] == "USB"
        
        # Test from_dict
        restored_device = Device.from_dict(device_dict)
        assert restored_device.name == device.name
        assert restored_device.manufacturer == device.manufacturer
        assert restored_device.connection_type == device.connection_type


class TestTestStepModel:
    """Test the TestStep model"""
    
    def test_test_step_creation(self):
        """Test basic test step creation"""
        step = TestStep(
            name="Measure Voltage",
            device_id="device123",
            step_type=StepType.SCPI_QUERY,
            scpi_command="READ?"
        )
        
        assert step.name == "Measure Voltage"
        assert step.device_id == "device123"
        assert step.step_type == StepType.SCPI_QUERY
        assert step.scpi_command == "READ?"
        assert step.status == StepStatus.PENDING
    
    def test_step_is_query(self):
        """Test step query detection"""
        query_step1 = TestStep(step_type=StepType.SCPI_QUERY)
        assert query_step1.is_query()
        
        query_step2 = TestStep(scpi_command="*IDN?")
        assert query_step2.is_query()
        
        command_step = TestStep(scpi_command="*RST")
        assert not command_step.is_query()
    
    def test_step_reset_execution_state(self):
        """Test resetting step execution state"""
        step = TestStep()
        step.status = StepStatus.PASSED
        step.execution_time = 1.5
        step.response = "OK"
        step.error_message = "Error"
        
        step.reset_execution_state()
        
        assert step.status == StepStatus.PENDING
        assert step.execution_time is None
        assert step.response is None
        assert step.error_message is None


class TestTestPlanModel:
    """Test the TestPlan model"""
    
    def test_test_plan_creation(self):
        """Test basic test plan creation"""
        plan = TestPlan(
            name="Voltage Test",
            description="Test voltage measurement"
        )
        
        assert plan.name == "Voltage Test"
        assert plan.description == "Test voltage measurement"
        assert plan.status == PlanStatus.IDLE
        assert len(plan.steps) == 0
    
    def test_plan_progress_calculation(self):
        """Test progress percentage calculation"""
        plan = TestPlan()
        
        # Add some steps
        step1 = TestStep(name="Step 1")
        step2 = TestStep(name="Step 2")
        step3 = TestStep(name="Step 3")
        plan.steps = [step1, step2, step3]
        
        # No steps completed
        assert plan.progress_percentage == 0.0
        
        # One step completed
        step1.status = StepStatus.PASSED
        assert plan.progress_percentage == 33.33333333333333
        
        # All steps completed
        step2.status = StepStatus.PASSED
        step3.status = StepStatus.FAILED
        assert plan.progress_percentage == 100.0
    
    def test_plan_validation(self):
        """Test test plan validation"""
        plan = TestPlan()
        
        # Empty plan should have issues
        issues = plan.validate_plan()
        assert "Test plan name is required" in issues
        assert "Test plan must contain at least one test step" in issues
        
        # Add name and steps
        plan.name = "Test Plan"
        step = TestStep(name="Test Step", scpi_command="*IDN?")
        plan.steps = [step]
        
        # Should still have device issue
        issues = plan.validate_plan()
        assert any("no device assigned" in issue.lower() for issue in issues)
        
        # Fix device assignment
        step.device_id = "device123"
        issues = plan.validate_plan()
        assert len(issues) == 0


class TestDeviceManager:
    """Test the DeviceManager service"""
    
    @pytest.fixture
    def temp_dir(self):
        """Create temporary directory for testing"""
        with tempfile.TemporaryDirectory() as temp_dir:
            yield temp_dir
    
    @pytest.fixture
    def device_manager(self, temp_dir):
        """Create DeviceManager instance for testing"""
        return DeviceManager(data_directory=temp_dir)
    
    def test_device_creation(self, device_manager):
        """Test device creation"""
        device = Device(name="Test Device")
        
        assert device_manager.create_device(device)
        assert len(device_manager.get_all_devices()) == 1
        
        retrieved = device_manager.get_device(device.id)
        assert retrieved is not None
        assert retrieved.name == "Test Device"
    
    def test_device_update(self, device_manager):
        """Test device update"""
        device = Device(name="Original Name")
        device_manager.create_device(device)
        
        device.name = "Updated Name"
        assert device_manager.update_device(device)
        
        retrieved = device_manager.get_device(device.id)
        assert retrieved.name == "Updated Name"
    
    def test_device_deletion(self, device_manager):
        """Test device deletion"""
        device = Device(name="Test Device")
        device_manager.create_device(device)
        
        assert len(device_manager.get_all_devices()) == 1
        assert device_manager.delete_device(device.id)
        assert len(device_manager.get_all_devices()) == 0
    
    def test_device_search(self, device_manager):
        """Test device search"""
        device1 = Device(name="Keysight Multimeter", manufacturer="Keysight")
        device2 = Device(name="Rigol Generator", manufacturer="Rigol")
        
        device_manager.create_device(device1)
        device_manager.create_device(device2)
        
        # Search by manufacturer
        results = device_manager.find_devices(manufacturer="Keysight")
        assert len(results) == 1
        assert results[0].name == "Keysight Multimeter"


class TestTestPlanManager:
    """Test the TestPlanManager service"""
    
    @pytest.fixture
    def temp_dir(self):
        """Create temporary directory for testing"""
        with tempfile.TemporaryDirectory() as temp_dir:
            yield temp_dir
    
    @pytest.fixture
    def plan_manager(self, temp_dir):
        """Create TestPlanManager instance for testing"""
        return TestPlanManager(data_directory=temp_dir)
    
    def test_plan_creation(self, plan_manager):
        """Test test plan creation"""
        plan = TestPlan(name="Test Plan")
        
        assert plan_manager.create_test_plan(plan)
        assert len(plan_manager.get_all_test_plans()) == 1
        
        retrieved = plan_manager.get_test_plan(plan.id)
        assert retrieved is not None
        assert retrieved.name == "Test Plan"
    
    def test_plan_duplication(self, plan_manager):
        """Test test plan duplication"""
        original = TestPlan(name="Original Plan")
        step = TestStep(name="Test Step")
        original.steps = [step]
        
        plan_manager.create_test_plan(original)
        
        duplicate = plan_manager.duplicate_test_plan(original.id, "Duplicated Plan")
        assert duplicate is not None
        assert duplicate.name == "Duplicated Plan"
        assert duplicate.id != original.id
        assert len(duplicate.steps) == 1
        assert duplicate.steps[0].id != step.id  # Step should have new ID


class TestConfigManager:
    """Test the ConfigManager service"""
    
    @pytest.fixture
    def temp_config_file(self):
        """Create temporary config file for testing"""
        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
            config_file = f.name
        yield config_file
        Path(config_file).unlink(missing_ok=True)
    
    def test_config_creation(self, temp_config_file):
        """Test configuration manager creation"""
        config_manager = ConfigManager(temp_config_file)
        assert config_manager.config is not None
        assert config_manager.config.default_timeout == 5.0
    
    def test_config_save_and_load(self, temp_config_file):
        """Test configuration save and load"""
        # Create and modify config
        config_manager = ConfigManager(temp_config_file)
        config_manager.set_setting("default_timeout", 10.0)
        config_manager.set_setting("log_level", "DEBUG")
        
        # Save config
        assert config_manager.save_config()
        
        # Create new manager and load
        new_manager = ConfigManager(temp_config_file)
        assert new_manager.get_setting("default_timeout") == 10.0
        assert new_manager.get_setting("log_level") == "DEBUG"
    
    def test_recent_files(self, temp_config_file):
        """Test recent files management"""
        config_manager = ConfigManager(temp_config_file)
        
        # Add recent files
        config_manager.add_recent_file("/path/to/plan1.json", "test_plan")
        config_manager.add_recent_file("/path/to/plan2.json", "test_plan")
        
        recent = config_manager.get_recent_files("test_plan")
        assert len(recent) == 2
        assert recent[0] == "/path/to/plan2.json"  # Most recent first
        
        # Test duplicate handling
        config_manager.add_recent_file("/path/to/plan1.json", "test_plan")
        recent = config_manager.get_recent_files("test_plan")
        assert len(recent) == 2
        assert recent[0] == "/path/to/plan1.json"


# Run tests when executed directly
if __name__ == "__main__":
    pytest.main([__file__, "-v"])