Aprendizaje Continuo en Automatización de Pruebas: Construyendo Sistemas de Test Auto-Mejorables

class TestFeedbackLoop:
    def __init__(self, model_store, metric_collector):
        self.model_store = model_store
        self.metric_collector = metric_collector
        self.learning_buffer = []

    def execute_test_cycle(self, test_suite):
        """Ejecutar tests y recopilar retroalimentación"""
        results = []

        for test in test_suite:
            # Ejecutar test
            outcome = test.run()

            # Recopilar datos contextuales
            context = {
                'test_id': test.id,
                'execution_time': outcome.duration,
                'failure_reason': outcome.error_message,
                'environment': test.environment,
                'timestamp': outcome.timestamp,
                'flakiness_score': self._calculate_flakiness(test.id)
            }

            # Añadir al buffer de aprendizaje
            self.learning_buffer.append({
                'outcome': outcome.status,
                'context': context,
                'test_characteristics': test.get_features()
            })

            results.append(outcome)

        return results

    def learn_from_feedback(self):
        """Procesar retroalimentación y actualizar modelos"""
        if len(self.learning_buffer) < 100:
            return  # Esperar datos suficientes

        # Extraer patrones
        patterns = self._extract_patterns(self.learning_buffer)

        # Actualizar modelos de predicción
        self.model_store.update_models({
            'failure_predictor': patterns['failure_patterns'],
            'flakiness_detector': patterns['flakiness_patterns'],
            'execution_time_estimator': patterns['timing_patterns']
        })

        # Limpiar retroalimentación procesada
        self.learning_buffer.clear()

    def _extract_patterns(self, feedback_data):
        """Extraer patrones accionables de la retroalimentación"""
        from sklearn.cluster import DBSCAN
        import numpy as np

        # Extraer características para clustering
        features = np.array([
            [
                item['context']['execution_time'],
                item['context']['flakiness_score'],
                hash(item['context']['environment']) % 1000
            ]
            for item in feedback_data
        ])

        # Identificar clusters de fallos
        clustering = DBSCAN(eps=0.3, min_samples=5).fit(features)

        return {
            'failure_patterns': clustering.labels_,
            'flakiness_patterns': self._detect_flakiness_patterns(feedback_data),
            'timing_patterns': self._analyze_timing_trends(feedback_data)
        }

from river import tree, metrics, ensemble
import datetime

class OnlineTestOptimizer:
    def __init__(self):
        # Bosque aleatorio adaptativo para predicción de fallos
        self.failure_model = ensemble.AdaptiveRandomForestClassifier(
            n_models=10,
            max_features='sqrt',
            lambda_value=6
        )

        # Árbol Hoeffding para selección de tests
        self.selection_model = tree.HoeffdingTreeClassifier()

        # Métricas de rendimiento
        self.failure_metric = metrics.Accuracy()
        self.selection_metric = metrics.Precision()

    def predict_failure_probability(self, test_features):
        """Predecir probabilidad de fallo del test"""
        return self.failure_model.predict_proba_one(test_features)

    def update_from_execution(self, test_features, actual_outcome):
        """Aprender de una sola ejecución de test"""
        # Actualizar modelo de predicción de fallos
        self.failure_model.learn_one(test_features, actual_outcome['failed'])

        # Actualizar métricas
        prediction = self.failure_model.predict_one(test_features)
        self.failure_metric.update(actual_outcome['failed'], prediction)

    def select_tests(self, available_tests, budget):
        """Seleccionar adaptativamente los tests más valiosos dentro del presupuesto"""
        test_scores = []

        for test in available_tests:
            features = test.extract_features()

            # Predecir probabilidad de fallo
            fail_prob = self.predict_failure_probability(features).get(True, 0.0)

            # Calcular puntuación de valor
            value_score = self._calculate_test_value(
                fail_probability=fail_prob,
                execution_cost=test.estimated_duration,
                code_coverage=test.coverage_metrics,
                last_execution=test.last_run_timestamp
            )

            test_scores.append((test, value_score))

        # Seleccionar tests de mayor valor dentro del presupuesto
        test_scores.sort(key=lambda x: x[1], reverse=True)

        selected = []
        total_cost = 0

        for test, score in test_scores:
            if total_cost + test.estimated_duration <= budget:
                selected.append(test)
                total_cost += test.estimated_duration

        return selected

    def _calculate_test_value(self, fail_probability, execution_cost,
                             code_coverage, last_execution):
        """Calcular métrica de valor para priorización de tests"""
        # Factor de decaimiento temporal
        hours_since_execution = (
            datetime.datetime.now() - last_execution
        ).total_seconds() / 3600
        recency_factor = 1 / (1 + hours_since_execution / 24)

        # Cálculo de valor
        value = (
            fail_probability * 0.4 +          # Probabilidad de fallo
            code_coverage * 0.3 +              # Importancia de cobertura
            recency_factor * 0.2 +             # Bonus de recencia
            (1 / execution_cost) * 0.1         # Factor de eficiencia
        )

        return value

import re
from collections import defaultdict, Counter
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import AgglomerativeClustering

class FailurePatternLearner:
    def __init__(self):
        self.failure_history = []
        self.pattern_database = defaultdict(list)
        self.vectorizer = TfidfVectorizer(max_features=100)

    def analyze_failure(self, test_failure):
        """Extraer y categorizar patrones de fallo"""
        # Analizar stack trace
        stack_trace = test_failure.stack_trace
        error_message = test_failure.error_message

        # Extraer elementos clave
        failure_signature = {
            'exception_type': self._extract_exception_type(error_message),
            'failing_component': self._extract_component(stack_trace),
            'error_keywords': self._extract_keywords(error_message),
            'stack_depth': len(stack_trace.split('\n')),
            'timestamp': test_failure.timestamp,
            'environment': test_failure.environment
        }

        # Almacenar fallo
        self.failure_history.append({
            'signature': failure_signature,
            'full_context': test_failure
        })

        # Actualizar base de datos de patrones
        pattern_key = f"{failure_signature['exception_type']}_{failure_signature['failing_component']}"
        self.pattern_database[pattern_key].append(failure_signature)

        # Verificar patrones recurrentes
        if len(self.pattern_database[pattern_key]) >= 3:
            return self._generate_pattern_alert(pattern_key)

        return None

    def cluster_similar_failures(self, recent_failures=100):
        """Agrupar fallos similares usando clustering"""
        if len(self.failure_history) < 10:
            return []

        # Obtener fallos recientes
        recent = self.failure_history[-recent_failures:]

        # Crear representaciones de texto
        failure_texts = [
            f"{f['signature']['exception_type']} {' '.join(f['signature']['error_keywords'])}"
            for f in recent
        ]

        # Vectorizar
        vectors = self.vectorizer.fit_transform(failure_texts)

        # Clustering
        clustering = AgglomerativeClustering(
            n_clusters=None,
            distance_threshold=0.5,
            linkage='average'
        )
        labels = clustering.fit_predict(vectors.toarray())

        # Agrupar por cluster
        clusters = defaultdict(list)
        for idx, label in enumerate(labels):
            clusters[label].append(recent[idx])

        return clusters

    def suggest_fixes(self, failure_signature):
        """Sugerir posibles soluciones basadas en patrones históricos"""
        # Encontrar fallos similares pasados
        similar_failures = self._find_similar_failures(failure_signature)

        # Extraer patrones comunes de resolución
        resolutions = []
        for similar in similar_failures:
            if similar.get('resolution'):
                resolutions.append(similar['resolution'])

        # Clasificar por frecuencia
        resolution_counts = Counter(resolutions)

        suggestions = []
        for resolution, count in resolution_counts.most_common(3):
            confidence = count / len(similar_failures)
            suggestions.append({
                'action': resolution,
                'confidence': confidence,
                'evidence_count': count
            })

        return suggestions

    def _extract_exception_type(self, error_message):
        """Extraer tipo de excepción del mensaje de error"""
        match = re.search(r'(\w+Exception|\w+Error)', error_message)
        return match.group(1) if match else 'UnknownException'

    def _extract_component(self, stack_trace):
        """Identificar componente que falla del stack trace"""
        lines = stack_trace.split('\n')
        for line in lines:
            if 'at ' in line and 'test' not in line.lower():
                match = re.search(r'at ([\w.]+)', line)
                if match:
                    return match.group(1).split('.')[0]
        return 'UnknownComponent'

    def _extract_keywords(self, error_message):
        """Extraer palabras clave significativas del mensaje de error"""
        # Eliminar palabras comunes
        stop_words = {'the', 'a', 'an', 'in', 'to', 'of', 'at', 'for', 'el', 'la', 'de', 'en'}
        words = re.findall(r'\b[a-z]{3,}\b', error_message.lower())
        return [w for w in words if w not in stop_words][:5]

from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException
import Levenshtein

class SelfHealingLocator:
    def __init__(self, driver, learning_rate=0.1):
        self.driver = driver
        self.locator_history = {}
        self.learning_rate = learning_rate

    def find_element(self, locator_strategies, element_context):
        """Probar múltiples estrategias y aprender de los éxitos"""
        element_id = element_context['id']

        # Probar primero la mejor estrategia histórica
        if element_id in self.locator_history:
            best_strategy = self.locator_history[element_id]['best']
            try:
                element = self._try_locator(best_strategy)
                self._update_success(element_id, best_strategy)
                return element
            except NoSuchElementException:
                self._update_failure(element_id, best_strategy)

        # Probar todas las estrategias
        for strategy in locator_strategies:
            try:
                element = self._try_locator(strategy)

                # Aprender del éxito
                self._learn_successful_strategy(element_id, strategy, element)
                return element

            except NoSuchElementException:
                continue

        # Todas las estrategias fallaron, intentar reparación
        return self._attempt_healing(locator_strategies, element_context)

    def _try_locator(self, strategy):
        """Intentar localizar elemento con estrategia dada"""
        by_type, value = strategy
        return self.driver.find_element(by_type, value)

    def _attempt_healing(self, failed_strategies, element_context):
        """Intentar reparar localizador encontrando elementos similares"""
        # Obtener todos los elementos en la página
        all_elements = self.driver.find_elements(By.XPATH, '//*')

        # Puntuar elementos por similitud con contexto esperado
        candidates = []
        for elem in all_elements:
            score = self._calculate_similarity(elem, element_context)
            if score > 0.7:  # Umbral para consideración
                candidates.append((elem, score))

        if not candidates:
            raise NoSuchElementException(f"No se pudo reparar localizador para {element_context['id']}")

        # Devolver mejor coincidencia
        candidates.sort(key=lambda x: x[1], reverse=True)
        healed_element = candidates[0][0]

        # Aprender nueva estrategia de localizador
        new_strategy = self._generate_locator_from_element(healed_element)
        self._learn_successful_strategy(
            element_context['id'],
            new_strategy,
            healed_element
        )

        return healed_element

    def _calculate_similarity(self, element, context):
        """Calcular puntuación de similitud entre elemento y contexto esperado"""
        score = 0.0

        # Similitud de texto
        if context.get('text'):
            elem_text = element.text.lower()
            expected_text = context['text'].lower()
            text_sim = 1 - (Levenshtein.distance(elem_text, expected_text) /
                           max(len(elem_text), len(expected_text)))
            score += text_sim * 0.4

        # Similitud de atributos
        if context.get('attributes'):
            for attr, value in context['attributes'].items():
                elem_value = element.get_attribute(attr)
                if elem_value:
                    attr_sim = 1 - (Levenshtein.distance(elem_value.lower(), value.lower()) /
                                   max(len(elem_value), len(value)))
                    score += attr_sim * 0.3

        # Similitud de posición
        if context.get('position'):
            elem_location = element.location
            expected_location = context['position']
            position_diff = abs(elem_location['x'] - expected_location['x']) + \
                          abs(elem_location['y'] - expected_location['y'])
            position_sim = 1 / (1 + position_diff / 100)
            score += position_sim * 0.3

        return score

    def _learn_successful_strategy(self, element_id, strategy, element):
        """Actualizar modelo de aprendizaje con estrategia exitosa"""
        if element_id not in self.locator_history:
            self.locator_history[element_id] = {
                'strategies': {},
                'best': strategy
            }

        strategies = self.locator_history[element_id]['strategies']

        # Actualizar puntuación de estrategia
        if strategy not in strategies:
            strategies[strategy] = {'successes': 0, 'failures': 0, 'score': 0.5}

        strategies[strategy]['successes'] += 1
        strategies[strategy]['score'] = (
            strategies[strategy]['successes'] /
            (strategies[strategy]['successes'] + strategies[strategy]['failures'])
        )

        # Actualizar mejor estrategia
        best_score = max(s['score'] for s in strategies.values())
        for strat, data in strategies.items():
            if data['score'] == best_score:
                self.locator_history[element_id]['best'] = strat
                break

import numpy as np
from collections import deque
import random

class RLTestSelector:
    def __init__(self, num_tests, learning_rate=0.1, discount_factor=0.95):
        self.num_tests = num_tests
        self.lr = learning_rate
        self.gamma = discount_factor
        self.epsilon = 1.0  # Tasa de exploración
        self.epsilon_decay = 0.995
        self.epsilon_min = 0.01

        # Tabla Q: estado -> acción -> valor
        self.q_table = {}

        # Replay de experiencia
        self.memory = deque(maxlen=2000)

    def get_state(self, test_suite):
        """Generar representación de estado"""
        state_features = []

        for test in test_suite:
            state_features.extend([
                test.recent_failure_rate,
                test.code_coverage_delta,
                test.execution_time_normalized,
                test.days_since_last_run
            ])

        return tuple(state_features)

    def select_action(self, state, available_tests):
        """Seleccionar tests usando política epsilon-greedy"""
        # Exploración
        if random.random() < self.epsilon:
            return random.sample(
                range(len(available_tests)),
                k=min(10, len(available_tests))
            )

        # Explotación
        if state not in self.q_table:
            self.q_table[state] = np.zeros(len(available_tests))

        q_values = self.q_table[state]
        return np.argsort(q_values)[-10:].tolist()  # Top 10 tests

    def calculate_reward(self, selected_tests, execution_results):
        """Calcular recompensa basada en resultados de ejecución"""
        reward = 0.0

        for test, result in zip(selected_tests, execution_results):
            if result.failed:
                # Alta recompensa por detectar fallos
                reward += 10.0

                # Bonus por detección temprana
                if test.execution_order <= 5:
                    reward += 5.0
            else:
                # Pequeña recompensa por tests que pasan (validación)
                reward += 0.1

            # Penalización por tiempo de ejecución
            reward -= result.execution_time / 60.0  # Normalizar a minutos

        return reward

    def train(self, state, action, reward, next_state):
        """Actualizar valores Q usando Q-learning"""
        if state not in self.q_table:
            self.q_table[state] = np.zeros(self.num_tests)
        if next_state not in self.q_table:
            self.q_table[next_state] = np.zeros(self.num_tests)

        # Actualización Q-learning
        for test_idx in action:
            current_q = self.q_table[state][test_idx]
            max_next_q = np.max(self.q_table[next_state])

            new_q = current_q + self.lr * (reward + self.gamma * max_next_q - current_q)
            self.q_table[state][test_idx] = new_q

        # Decaer tasa de exploración
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

    def remember(self, state, action, reward, next_state):
        """Almacenar experiencia para replay"""
        self.memory.append((state, action, reward, next_state))

    def replay(self, batch_size=32):
        """Entrenar con lote aleatorio de memoria"""
        if len(self.memory) < batch_size:
            return

        batch = random.sample(self.memory, batch_size)

        for state, action, reward, next_state in batch:
            self.train(state, action, reward, next_state)

from datetime import datetime, timedelta
import hashlib

class ModelRetrainingOrchestrator:
    def __init__(self, models):
        self.models = models
        self.performance_tracker = {}
        self.data_tracker = {}

    def should_retrain(self, model_name):
        """Determinar si el modelo necesita re-entrenamiento"""
        triggers = {
            'time_based': self._check_time_trigger(model_name),
            'performance_degradation': self._check_performance_trigger(model_name),
            'data_drift': self._check_data_drift_trigger(model_name),
            'significant_events': self._check_event_trigger(model_name)
        }

        # Re-entrenar si algún trigger está activo
        return any(triggers.values()), triggers

    def _check_time_trigger(self, model_name):
        """Verificar si ha pasado suficiente tiempo desde el último entrenamiento"""
        last_training = self.models[model_name].last_training_time
        time_threshold = timedelta(days=7)  # Re-entrenar semanalmente

        return datetime.now() - last_training > time_threshold

    def _check_performance_trigger(self, model_name):
        """Verificar degradación de rendimiento"""
        if model_name not in self.performance_tracker:
            return False

        recent_accuracy = self.performance_tracker[model_name]['recent_accuracy']
        baseline_accuracy = self.performance_tracker[model_name]['baseline_accuracy']

        # Trigger si el rendimiento cae un 5%
        return recent_accuracy < baseline_accuracy * 0.95

    def _check_data_drift_trigger(self, model_name):
        """Detectar drift en distribución de datos"""
        if model_name not in self.data_tracker:
            return False

        current_distribution = self.data_tracker[model_name]['current']
        training_distribution = self.data_tracker[model_name]['baseline']

        # Calcular divergencia KL o métrica similar
        drift_score = self._calculate_drift(current_distribution, training_distribution)

        return drift_score > 0.1  # Umbral para drift significativo

    def _check_event_trigger(self, model_name):
        """Verificar eventos significativos que requieran re-entrenamiento"""
        events = self.models[model_name].recent_events

        significant_events = [
            'major_release',
            'architecture_change',
            'test_suite_expansion'
        ]

        return any(event['type'] in significant_events for event in events)

    def execute_retraining(self, model_name):
        """Realizar re-entrenamiento incremental o completo"""
        model = self.models[model_name]

        # Recopilar datos de entrenamiento
        training_data = self._prepare_training_data(model_name)

        # Elegir estrategia de re-entrenamiento
        if len(training_data) > 10000:
            # Re-entrenamiento incremental para datasets grandes
            self._incremental_retrain(model, training_data)
        else:
            # Re-entrenamiento completo para datasets pequeños
            self._full_retrain(model, training_data)

        # Actualizar metadatos
        model.last_training_time = datetime.now()
        model.training_data_hash = self._hash_data(training_data)

        # Validar nuevo modelo
        validation_score = self._validate_model(model)

        if validation_score > self.performance_tracker[model_name]['baseline_accuracy']:
            # Desplegar nuevo modelo
            self._deploy_model(model_name, model)
            print(f"Modelo {model_name} re-entrenado y desplegado. Nueva precisión: {validation_score:.3f}")
        else:
            # Revertir a modelo anterior
            print(f"Re-entrenamiento del modelo {model_name} falló validación. Manteniendo versión anterior.")

    def _calculate_drift(self, current, baseline):
        """Calcular drift de distribución usando divergencia KL"""
        import scipy.stats as stats
        return stats.entropy(current, baseline)

class ProductionTestingSystem:
    """Sistema de testing auto-mejorable de nivel empresarial"""

    def __init__(self):
        self.feedback_loop = TestFeedbackLoop(model_store, metric_collector)
        self.online_optimizer = OnlineTestOptimizer()
        self.pattern_learner = FailurePatternLearner()
        self.self_healing = SelfHealingLocator(driver)
        self.rl_selector = RLTestSelector(num_tests=500)
        self.retraining_orchestrator = ModelRetrainingOrchestrator(models)

    def run_intelligent_test_cycle(self, time_budget):
        """Ejecutar ciclo de test optimizado con aprendizaje continuo"""
        # Obtener estado actual del sistema
        state = self._capture_system_state()

        # Seleccionar tests usando RL
        available_tests = self._get_available_tests()
        selected_indices = self.rl_selector.select_action(state, available_tests)
        selected_tests = [available_tests[i] for i in selected_indices]

        # Ejecutar con auto-reparación
        results = []
        for test in selected_tests:
            result = test.run_with_healing(self.self_healing)
            results.append(result)

            # Actualización de aprendizaje online
            self.online_optimizer.update_from_execution(
                test.extract_features(),
                result
            )

            # Analizar fallos
            if result.failed:
                pattern_alert = self.pattern_learner.analyze_failure(result)
                if pattern_alert:
                    self._handle_pattern_alert(pattern_alert)

        # Calcular recompensa y entrenar agente RL
        reward = self.rl_selector.calculate_reward(selected_tests, results)
        next_state = self._capture_system_state()
        self.rl_selector.train(state, selected_indices, reward, next_state)

        # Procesamiento del ciclo de retroalimentación
        self.feedback_loop.learn_from_feedback()

        # Verificar triggers de re-entrenamiento
        for model_name in self.retraining_orchestrator.models:
            should_retrain, triggers = self.retraining_orchestrator.should_retrain(model_name)
            if should_retrain:
                self.retraining_orchestrator.execute_retraining(model_name)

        return {
            'tests_executed': len(selected_tests),
            'failures_found': sum(1 for r in results if r.failed),
            'time_used': sum(r.execution_time for r in results),
            'efficiency_score': reward / time_budget
        }