rpa_vision_v3/tests/unit/test_self_healing.py

"""Unit tests for self-healing workflows."""

import pytest
import tempfile
import shutil
from pathlib import Path
from datetime import datetime

from core.healing.healing_engine import SelfHealingEngine
from core.healing.learning_repository import LearningRepository
from core.healing.confidence_scorer import ConfidenceScorer
from core.healing.models import RecoveryContext, RecoveryResult, RecoveryPattern
from core.healing.strategies import (
    SemanticVariantStrategy,
    SpatialFallbackStrategy,
    TimingAdaptationStrategy,
    FormatTransformationStrategy
)


class TestConfidenceScorer:
    """Tests for confidence scorer."""
    
    def test_confidence_score_range(self):
        """Confidence scores should be between 0 and 1."""
        scorer = ConfidenceScorer()
        
        context = RecoveryContext(
            original_action='click',
            target_element='button',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test.png',
            workflow_id='test_wf',
            node_id='node1',
            attempt_count=1
        )
        
        confidence = scorer.calculate_recovery_confidence(
            'semantic_variant',
            context,
            0.5
        )
        
        assert 0.0 <= confidence <= 1.0
    
    def test_text_similarity(self):
        """Text similarity should work correctly."""
        scorer = ConfidenceScorer()
        
        # Exact match
        assert scorer._text_similarity('submit', 'submit') == 1.0
        
        # Similar
        similarity = scorer._text_similarity('submit', 'send')
        assert 0.0 < similarity < 1.0
        
        # Different
        similarity = scorer._text_similarity('submit', 'xyz')
        assert similarity < 0.5


class TestLearningRepository:
    """Tests for learning repository."""
    
    def setup_method(self):
        """Setup test repository."""
        self.temp_dir = tempfile.mkdtemp()
        self.repo = LearningRepository(Path(self.temp_dir))
    
    def teardown_method(self):
        """Cleanup test repository."""
        shutil.rmtree(self.temp_dir, ignore_errors=True)
    
    def test_store_and_retrieve_pattern(self):
        """Should store and retrieve patterns."""
        context = RecoveryContext(
            original_action='click',
            target_element='button',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test.png',
            workflow_id='test_wf',
            node_id='node1',
            attempt_count=1
        )
        
        result = RecoveryResult(
            success=True,
            strategy_used='semantic_variant',
            confidence_score=0.85
        )
        
        # Store pattern
        self.repo.store_pattern(context, result)
        
        # Retrieve patterns
        patterns = self.repo.get_all_patterns()
        assert len(patterns) == 1
        assert patterns[0].recovery_strategy == 'semantic_variant'
    
    def test_pattern_matching(self):
        """Should match patterns correctly."""
        context1 = RecoveryContext(
            original_action='click',
            target_element='button1',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test.png',
            workflow_id='test_wf',
            node_id='node1',
            attempt_count=1,
            metadata={'element_type': 'button'}
        )
        
        result = RecoveryResult(
            success=True,
            strategy_used='semantic_variant',
            confidence_score=0.85
        )
        
        self.repo.store_pattern(context1, result)
        
        # Similar context
        context2 = RecoveryContext(
            original_action='click',
            target_element='button2',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test2.png',
            workflow_id='test_wf',
            node_id='node2',
            attempt_count=1,
            metadata={'element_type': 'button'}
        )
        
        matching = self.repo.get_matching_patterns(context2)
        assert len(matching) > 0


class TestSemanticVariantStrategy:
    """Tests for semantic variant strategy."""
    
    def test_can_handle(self):
        """Should handle element_not_found failures."""
        strategy = SemanticVariantStrategy()
        
        context = RecoveryContext(
            original_action='click',
            target_element='button',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test.png',
            workflow_id='test_wf',
            node_id='node1',
            attempt_count=1
        )
        
        assert strategy.can_handle(context)
    
    def test_get_semantic_variants(self):
        """Should get semantic variants."""
        strategy = SemanticVariantStrategy()
        
        variants = strategy._get_semantic_variants('submit')
        assert 'send' in variants
        assert 'ok' in variants
        assert 'confirm' in variants


class TestSelfHealingEngine:
    """Tests for self-healing engine."""
    
    def setup_method(self):
        """Setup test engine."""
        self.temp_dir = tempfile.mkdtemp()
        self.engine = SelfHealingEngine(storage_path=Path(self.temp_dir))
    
    def teardown_method(self):
        """Cleanup test engine."""
        shutil.rmtree(self.temp_dir, ignore_errors=True)
    
    def test_initialization(self):
        """Engine should initialize correctly."""
        assert self.engine.learning_repo is not None
        assert self.engine.confidence_scorer is not None
        assert len(self.engine.recovery_strategies) > 0
    
    def test_max_attempts_exceeded(self):
        """Should fail when max attempts exceeded."""
        context = RecoveryContext(
            original_action='click',
            target_element='button',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test.png',
            workflow_id='test_wf',
            node_id='node1',
            attempt_count=5,
            max_attempts=3
        )
        
        result = self.engine.attempt_recovery(context)
        assert not result.success
        assert 'Max recovery attempts' in result.error_message
    
    def test_learn_from_success(self):
        """Should learn from successful recovery."""
        context = RecoveryContext(
            original_action='click',
            target_element='button',
            failure_reason='element_not_found',
            screenshot_path='/tmp/test.png',
            workflow_id='test_wf',
            node_id='node1',
            attempt_count=1
        )
        
        result = RecoveryResult(
            success=True,
            strategy_used='semantic_variant',
            confidence_score=0.85
        )
        
        self.engine.learn_from_success(context, result)
        
        patterns = self.engine.learning_repo.get_all_patterns()
        assert len(patterns) > 0


if __name__ == '__main__':
    pytest.main([__file__, '-v'])