Mutation Testing con IA: Generación Inteligente de Mutantes para Mejor Calidad de Pruebas

# Código original
def calculate_discount(price, percentage):
    if percentage > 0:
        return price * (1 - percentage / 100)
    return price

# Mutación 1: Reemplazo de operador relacional (ROR)
def calculate_discount(price, percentage):
    if percentage >= 0:  # Cambiado > a >=
        return price * (1 - percentage / 100)
    return price

# Mutación 2: Reemplazo de operador aritmético (AOR)
def calculate_discount(price, percentage):
    if percentage > 0:
        return price * (1 + percentage / 100)  # Cambiado - a +
    return price

# Mutación 3: Reemplazo de constante (CRR)
def calculate_discount(price, percentage):
    if percentage > 1:  # Cambiado 0 a 1
        return price * (1 - percentage / 100)
    return price

import ast
from sklearn.ensemble import RandomForestClassifier
import numpy as np

class IntelligentMutantGenerator:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100)
        self.trained = False

    def extract_features(self, code_snippet, mutation_operator):
        """Extrae características para priorización de mutantes basada en ML"""
 (como se discute en [AI Code Smell Detection: Finding Problems in Test Automation with ML](/blog/ai-code-smell-detection))        tree = ast.parse(code_snippet)

        features = {
            'cyclomatic_complexity': self._calculate_complexity(tree),
            'nesting_depth': self._max_nesting_depth(tree),
            'num_branches': len([n for n in ast.walk(tree) if isinstance(n, ast.If)]),
            'num_loops': len([n for n in ast.walk(tree) if isinstance(n, (ast.For, ast.While))]),
            'code_length': len(code_snippet.split('\n')),
            'mutation_type': self._encode_mutation_type(mutation_operator),
            'variable_count': len([n for n in ast.walk(tree) if isinstance(n, ast.Name)]),
            'test_coverage': self._get_line_coverage(code_snippet)
        }

        return np.array(list(features.values())).reshape(1, -1)

    def predict_mutant_quality(self, code_snippet, mutation_operator):
        """Predice la probabilidad de que el mutante revele debilidades en las pruebas"""
        if not self.trained (como se discute en [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation)):
            raise ValueError("El modelo debe ser entrenado primero")

        features = self.extract_features(code_snippet, mutation_operator)
        # Retorna probabilidad de que el mutante sea valioso (no equivalente, probablemente sobreviva)
        return self.model.predict_proba(features)[0][1]

    def generate_prioritized_mutants(self, source_code, max_mutants=50):
        """Genera mutantes priorizados por valor predicho"""
        candidate_mutants = self._generate_all_mutants(source_code)

        # Puntúa cada mutante
        scored_mutants = []
        for mutant in candidate_mutants:
            score = self.predict_mutant_quality(mutant.code, mutant.operator)
            scored_mutants.append((score, mutant))

        # Retorna los N mejores mutantes por calidad predicha
        scored_mutants.sort(reverse=True, key=lambda x: x[0])
        return [mutant for score, mutant in scored_mutants[:max_mutants]]

class MutantTrainingData:
    def __init__(self):
        self.training_examples = []

    def collect_from_run(self, mutation_result):
        """Recopila datos de entrenamiento de la ejecución de mutation testing"""
        for mutant in mutation_result.mutants:
            features = self.extract_features(mutant.code, mutant.operator)

            # Etiqueta: 1 si el mutante sobrevivió y no era equivalente (valioso)
            #           0 si el mutante fue eliminado o era equivalente (menos valioso)
            label = 1 if (mutant.survived and not mutant.is_equivalent) else 0

            self.training_examples.append({
                'features': features,
                'label': label,
                'survival_time': mutant.detection_time,
                'test_count': len(mutant.killing_tests)
            })

    def train_model(self, generator):
        """Entrena el predictor de calidad de mutantes"""
        X = np.array([ex['features'] for ex in self.training_examples])
        y = np.array([ex['label'] for ex in self.training_examples])

        generator.model.fit(X, y)
        generator.trained = True

        return generator.model.score(X, y)

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

class SemanticMutationEngine:
    def __init__(self):
        # Usar CodeBERT o modelo similar para comprensión de código
        self.tokenizer = AutoTokenizer.from_pretrained("microsoft/codebert-base")
        self.model = AutoModelForSequenceClassification.from_pretrained(
            "microsoft/codebert-base"
        )

    def is_semantically_equivalent(self, original_code, mutant_code):
        """Usa IA para detectar mutantes equivalentes antes de la ejecución"""

        # Codifica ambas versiones
        original_tokens = self.tokenizer(original_code, return_tensors="pt",
                                        truncation=True, max_length=512)
        mutant_tokens = self.tokenizer(mutant_code, return_tensors="pt",
                                      truncation=True, max_length=512)

        with torch.no_grad():
            original_embedding = self.model(**original_tokens).logits
            mutant_embedding = self.model(**mutant_tokens).logits

        # Calcula similitud semántica
        similarity = torch.cosine_similarity(original_embedding, mutant_embedding)

        # Si similitud > umbral, probablemente equivalente
        return similarity.item() > 0.95

    def suggest_mutation_location(self, source_code):
        """Usa mecanismos de atención para identificar puntos críticos de mutación"""
        tokens = self.tokenizer(source_code, return_tensors="pt")

        with torch.no_grad():
            outputs = self.model(**tokens, output_attentions=True)
            attention = outputs.attentions[-1]  # Atención de la última capa

        # Agrega puntuaciones de atención
        attention_scores = attention.mean(dim=1).squeeze()

        # Identifica tokens con alta atención (importantes para el comportamiento)
        important_indices = torch.topk(attention_scores.mean(dim=0), k=10).indices

        return important_indices.tolist()

from sklearn.cluster import DBSCAN
from sklearn.decomposition import PCA

class MutantAnalyzer:
    def analyze_test_weaknesses(self, mutation_results):
        """Agrupa mutantes sobrevivientes para identificar brechas sistemáticas en pruebas"""

        # Extrae características de mutantes sobrevivientes
        survivors = [m for m in mutation_results.mutants if m.survived]

        features = []
        for mutant in survivors:
            features.append([
                mutant.location.line_number,
                mutant.location.column,
                self._encode_operator(mutant.operator),
                mutant.complexity_score,
                mutant.branch_id,
                self._encode_context(mutant.context)
            ])

        # Agrupa para encontrar patrones
        clustering = DBSCAN(eps=0.5, min_samples=3).fit(features)

        # Analiza clusters
        weakness_patterns = {}
        for cluster_id in set(clustering.labels_):
            if cluster_id == -1:  # Omite ruido
                continue

            cluster_mutants = [survivors[i] for i, label in enumerate(clustering.labels_)
                             if label == cluster_id]

            pattern = self._extract_pattern(cluster_mutants)
            weakness_patterns[cluster_id] = {
                'pattern': pattern,
                'severity': len(cluster_mutants),
                'suggested_tests': self._suggest_tests(pattern)
            }

        return weakness_patterns

    def _suggest_tests(self, pattern):
        """Usa análisis de patrones para sugerir nuevos casos de prueba"""
        suggestions = []

        if pattern['type'] == 'boundary_condition':
            suggestions.append({
                'test_type': 'Pruebas de casos límite',
                'focus': f"Valores límite alrededor de {pattern['value']}",
                'example': self._generate_test_template(pattern)
            })

        elif pattern['type'] == 'conditional_logic':
            suggestions.append({
                'test_type': 'Cobertura de ramas',
                'focus': f"Condiciones complementarias en {pattern['location']}",
                'example': self._generate_branch_tests(pattern)
            })

        return suggestions

class TestEffectivenessAnalyzer:
    def calculate_mtes(self, test_suite, mutation_results):
        """
        Puntuación de Efectividad de Pruebas Basada en Mutación (MTES)
        Combina múltiples dimensiones de calidad de pruebas
        """

        metrics = {
            'mutation_score': self._basic_mutation_score(mutation_results),
            'detection_speed': self._average_detection_time(mutation_results),
            'coverage_diversity': self._mutation_coverage_diversity(mutation_results),
            'false_negative_rate': self._equivalent_mutant_ratio(mutation_results),
            'critical_path_coverage': self._critical_mutation_coverage(mutation_results)
        }

        # Combinación ponderada
        weights = {
            'mutation_score': 0.35,
            'detection_speed': 0.15,
            'coverage_diversity': 0.25,
            'false_negative_rate': -0.10,  # Penalización
            'critical_path_coverage': 0.35
        }

        mtes = sum(metrics[k] * weights[k] for k in weights.keys())

        return {
            'overall_score': mtes,
            'breakdown': metrics,
            'grade': self._assign_grade(mtes),
            'recommendations': self._generate_recommendations(metrics)
        }

    def _critical_mutation_coverage(self, results):
        """Mide cobertura de mutaciones críticas identificadas por IA"""
        critical_mutants = [m for m in results.mutants if m.criticality_score > 0.8]
        killed_critical = [m for m in critical_mutants if not m.survived]

        return len(killed_critical) / len(critical_mutants) if critical_mutants else 0

# Integración con MutPy para proyectos Python
from mutpy import controller
from mutpy.commandline import build_parser

class AIMutPyController(controller.MutationController):
    def __init__(self, ai_engine):
        super().__init__()
        self.ai_engine = ai_engine

    def run(self):
        """Sobrescribe para agregar priorización de mutantes basada en IA"""
        # Genera mutantes como siempre
        all_mutants = super().generate_mutants()

        # Usa IA para priorizar
        prioritized = self.ai_engine.prioritize_mutants(all_mutants)

        # Ejecuta solo el N% superior de mutantes
        results = self.execute_mutants(prioritized[:self.config.mutant_limit])

        # Usa IA para detectar mutantes equivalentes
        filtered_results = self.filter_equivalent_mutants(results)

        return self.generate_report(filtered_results)

    def filter_equivalent_mutants(self, results):
        """Usa análisis semántico para identificar mutantes equivalentes"""
        filtered = []

        for result in results:
            if result.survived:
                is_equivalent = self.ai_engine.is_semantically_equivalent(
                    result.original_code,
                    result.mutant_code
                )

                result.is_equivalent = is_equivalent
                result.confidence = self.ai_engine.equivalence_confidence

            filtered.append(result)

        return filtered

# Uso
ai_engine = IntelligentMutantGenerator()
controller = AIMutPyController(ai_engine)
controller.config.mutant_limit = 100
results = controller.run()

<!-- Configuración mejorada de PIT con mutaciones guiadas por IA -->
<plugin>
    <groupId>org.pitest</groupId>
    <artifactId>pitest-maven</artifactId>
    <version>1.15.0</version>
    <configuration>
        <targetClasses>
            <param>com.example.critical.*</param>
        </targetClasses>
        <targetTests>
            <param>com.example.tests.*</param>
        </targetTests>

        <!-- Selección personalizada de mutadores basada en recomendaciones de IA -->
        <mutators>
            <mutator>CONDITIONALS_BOUNDARY</mutator>
            <mutator>INCREMENTS</mutator>
            <mutator>MATH</mutator>
            <mutator>NEGATE_CONDITIONALS</mutator>
        </mutators>

        <!-- Filtrado de mutantes basado en IA -->
        <features>
            <feature>+AI_MUTANT_FILTER</feature>
            <feature>+SEMANTIC_EQUIVALENCE_DETECTION</feature>
        </features>

        <threads>4</threads>
        <timeoutConstant>10000</timeoutConstant>
    </configuration>
</plugin>

from concurrent.futures import ProcessPoolExecutor, as_completed
import multiprocessing as mp

class OptimizedMutationRunner:
    def __init__(self, num_workers=None):
        self.num_workers = num_workers or mp.cpu_count()
        self.ai_scheduler = IntelligentScheduler()

    def run_mutations_parallel(self, mutants, test_suite):
        """Ejecuta mutaciones con programación basada en IA"""

        # Agrupa mutantes por tiempo de ejecución predicho
        mutant_groups = self.ai_scheduler.group_by_execution_time(mutants)

        results = []
        with ProcessPoolExecutor(max_workers=self.num_workers) as executor:
            # Envía trabajos en orden de tiempo de ejecución descendente
            futures = {}
            for group in sorted(mutant_groups, key=lambda g: g.estimated_time, reverse=True):
                future = executor.submit(self._execute_mutant_group, group, test_suite)
                futures[future] = group

            # Recopila resultados a medida que se completan
            for future in as_completed(futures):
                group = futures[future]
                try:
                    group_results = future.result()
                    results.extend(group_results)

                    # Actualiza estimaciones de tiempo de ejecución para aprendizaje
                    self.ai_scheduler.update_estimates(group, group_results)
                except Exception as e:
                    print(f"El grupo {group.id} falló: {e}")

        return results

    def _execute_mutant_group(self, group, test_suite):
        """Ejecuta un grupo de mutantes similares"""
        results = []

        for mutant in group.mutants:
            # Terminación temprana si la primera prueba elimina el mutante
            result = self._execute_with_early_stop(mutant, test_suite)
            results.append(result)

        return results

    def _execute_with_early_stop(self, mutant, test_suite):
        """Detiene la ejecución tan pronto como el mutante es eliminado"""
        # IA predice qué pruebas tienen más probabilidades de eliminar este mutante
        ordered_tests = self.ai_scheduler.order_tests_for_mutant(
            mutant, test_suite
        )

        for test in ordered_tests:
            result = self._run_single_test(mutant, test)
            if result.killed:
                return result  # Terminación temprana

        return MutantResult(mutant, survived=True)

class IncrementalMutationTester:
    def __init__(self, cache_dir='.mutation_cache'):
        self.cache_dir = cache_dir
        self.git_analyzer = GitChangeAnalyzer()

    def run_incremental(self, baseline_commit, current_commit):
        """Solo ejecuta mutaciones para código modificado"""

        # Identifica líneas cambiadas
        changed_regions = self.git_analyzer.get_changed_regions(
            baseline_commit, current_commit
        )

        # Carga resultados en caché
        cached_results = self._load_cache(baseline_commit)

        # Genera mutantes solo para regiones modificadas
        new_mutants = []
        for region in changed_regions:
            mutants = self.generate_mutants_for_region(region)
            new_mutants.extend(mutants)

        # Ejecuta solo mutantes nuevos
        new_results = self.execute_mutants(new_mutants)

        # Combina con resultados en caché
        combined_results = self._merge_results(cached_results, new_results)

        # Actualiza caché
        self._save_cache(current_commit, combined_results)

        return combined_results

# Función de validación de pagos original
class PaymentProcessor:
    def validate_payment(self, amount, currency, card_number):
        if amount <= 0:
            raise ValueError("El monto debe ser positivo")

        if currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError("Moneda no soportada")

        if not self._validate_card(card_number):
            raise ValueError("Número de tarjeta inválido")

        return True

    def _validate_card(self, card_number):
        # Algoritmo de Luhn
        digits = [int(d) for d in str(card_number)]
        checksum = sum(digits[-1::-2] + [sum(divmod(d*2, 10)) for d in digits[-2::-2]])
        return checksum % 10 == 0

# La IA identificó estos mutantes sobrevivientes críticos:

# Mutante 1: Condición límite (ALTA PRIORIDAD)
if amount < 0:  # Cambiado <= a <
    raise ValueError("El monto debe ser positivo")
# Impacto: Pagos con monto cero no rechazados

# Mutante 2: Validación incompleta (ALTA PRIORIDAD)
if currency not in ['USD', 'EUR']:  # Eliminó 'GBP'
    raise ValueError("Moneda no soportada")
# Impacto: Manejo de GBP no probado

# Mutante 3: Error de lógica (CRÍTICO)
return checksum % 10 == 1  # Cambiado 0 a 1
# Impacto: Lógica de validación de tarjeta no verificada