Análisis de Impacto de Tests con IA: Selección Inteligente Tras Cambios de Código

import ast
import difflib

class CodeChangeAnalyzer:
    def __init__(self):
        self.changed_functions = set()
        self.changed_classes = set()
        self.changed_imports = set()

    def analyze_changes(self, old_code, new_code):
        """Analizar cambios de código usando análisis AST"""
        old_tree = ast.parse(old_code)
        new_tree = ast.parse(new_code)

        old_functions = self._extract_functions(old_tree)
        new_functions = self._extract_functions(new_tree)

        # Detectar funciones modificadas
        for func_name in old_functions.keys():
            if func_name in new_functions:
                if old_functions[func_name] != new_functions[func_name]:
                    self.changed_functions.add(func_name)

        # Detectar nuevas funciones
        for func_name in new_functions.keys():
            if func_name not in old_functions:
                self.changed_functions.add(func_name)

        return self.get_impact_summary()

    def _extract_functions(self, tree):
        """Extraer definiciones de funciones del AST"""
        functions = {}
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                functions[node.name] = ast.unparse(node)
        return functions

    def get_impact_summary(self):
        return {
            'functions': list(self.changed_functions),
            'classes': list(self.changed_classes),
            'imports': list(self.changed_imports)
        }

# Ejemplo de uso
analyzer = CodeChangeAnalyzer()
old_code = """
def calculate_total(items):
    return sum(item.price for item in items)
"""

new_code = """
def calculate_total(items, discount=0):
    subtotal = sum(item.price for item in items)
    return subtotal * (1 - discount)
"""

impact = analyzer.analyze_changes(old_code, new_code)
print(f"Funciones modificadas: {impact['functions']}")
# Salida: Funciones modificadas: ['calculate_total']

from transformers import AutoTokenizer, AutoModel
import torch

class SemanticChangeDetector:
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained('microsoft/codebert-base')
 (como se discute en [AI-Assisted Bug Triaging: Intelligent Defect Prioritization at Scale](/blog/ai-bug-triaging))        self.model = AutoModel.from_pretrained('microsoft/codebert-base')

 (como se discute en [AI Code Smell Detection: Finding Problems in Test Automation with ML](/blog/ai-code-smell-detection))    def get_embedding(self, code):
        """Generar embedding semántico para fragmento de código"""
        inputs = self.tokenizer(code, return_tensors='pt',
                               truncation=True, max_length=512)
        with torch.no_grad():
            outputs = self.model(**inputs)
        return outputs.last_hidden_state.mean(dim=1)

    def calculate_similarity(self, old_code, new_code):
        """Calcular similitud semántica entre versiones de código"""
        old_embedding = self.get_embedding(old_code)
        new_embedding = self.get_embedding(new_code)

        similarity = torch.cosine_similarity(old_embedding, new_embedding)
        return similarity.item()

    def is_significant_change(self, old_code, new_code, threshold=0.85):
        """Determinar si el cambio es semánticamente significativo"""
        similarity = self.calculate_similarity(old_code, new_code)
        return similarity < threshold

# Ejemplo: Detectar refactorización vs cambios de lógica
detector = SemanticChangeDetector()

# Refactorización (alta similitud)
old_v1 = "def add(a, b): return a + b"
new_v1 = "def add(x, y): return x + y"
print(f"Similitud de refactorización: {detector.calculate_similarity(old_v1, new_v1):.3f}")

# Cambio de lógica (baja similitud)
old_v2 = "def process(data): return data.sort()"
new_v2 = "def process(data): return data.filter(lambda x: x > 0).sort()"
print(f"Similitud de cambio de lógica: {detector.calculate_similarity(old_v2, new_v2):.3f}")

import networkx as nx
from typing import Set, Dict, List

class DependencyGraphBuilder:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.file_dependencies = {}

    def add_module(self, module_name: str, dependencies: List[str]):
        """Agregar módulo y sus dependencias al grafo"""
        self.graph.add_node(module_name)
        for dep in dependencies:
            self.graph.add_edge(module_name, dep)

    def find_affected_modules(self, changed_modules: Set[str]) -> Set[str]:
        """Encontrar todos los módulos afectados por cambios usando dependencias inversas"""
        affected = set(changed_modules)

        for module in changed_modules:
            # Encontrar todos los módulos que dependen de este módulo modificado
            if module in self.graph:
                ancestors = nx.ancestors(self.graph, module)
                affected.update(ancestors)

        return affected

    def get_test_coverage_map(self) -> Dict[str, Set[str]]:
        """Mapear archivos fuente a archivos de prueba que los cubren"""
        coverage_map = {}

        for node in self.graph.nodes():
            if node.endswith('_test.py'):
                # Encontrar todos los archivos fuente que cubre esta prueba
                descendants = nx.descendants(self.graph, node)
                for source_file in descendants:
                    if not source_file.endswith('_test.py'):
                        if source_file not in coverage_map:
                            coverage_map[source_file] = set()
                        coverage_map[source_file].add(node)

        return coverage_map

# Ejemplo de uso
builder = DependencyGraphBuilder()

# Construir grafo de dependencias
builder.add_module('src/auth.py', ['src/database.py', 'src/utils.py'])
builder.add_module('src/api.py', ['src/auth.py', 'src/models.py'])
builder.add_module('tests/test_auth.py', ['src/auth.py'])
builder.add_module('tests/test_api.py', ['src/api.py'])

# Encontrar módulos afectados
changed_files = {'src/database.py'}
affected = builder.find_affected_modules(changed_files)
print(f"Módulos afectados: {affected}")
# Salida: Módulos afectados: {'src/database.py', 'src/auth.py', 'src/api.py'}

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
import numpy as np

class TestRiskPredictor:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.scaler = StandardScaler()
        self.is_trained (como se discute en [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation)) = False

    def extract_features(self, change_data):
        """Extraer características de datos de cambio de código"""
        features = {
            'lines_added': change_data.get('additions', 0),
            'lines_deleted': change_data.get('deletions', 0),
            'files_changed': change_data.get('changed_files', 1),
            'cyclomatic_complexity': change_data.get('complexity', 1),
            'author_experience': change_data.get('author_commits', 0),
            'time_since_last_change': change_data.get('hours_since_change', 0),
            'num_dependencies': change_data.get('dependency_count', 0),
            'historical_failure_rate': change_data.get('past_failures', 0.0)
        }
        return list(features.values())

    def train(self, historical_data: pd.DataFrame):
        """Entrenar modelo con resultados históricos de pruebas"""
        X = np.array([self.extract_features(row)
                     for _, row in historical_data.iterrows()])
        y = historical_data['test_failed'].values

        X_scaled = self.scaler.fit_transform(X)
        self.model.fit(X_scaled, y)
        self.is_trained = True

    def predict_risk(self, change_data) -> float:
        """Predecir probabilidad de fallo de prueba"""
        if not self.is_trained:
            raise ValueError("El modelo debe estar entrenado primero")

        features = np.array([self.extract_features(change_data)])
        features_scaled = self.scaler.transform(features)

        # Devolver probabilidad de fallo (clase 1)
        return self.model.predict_proba(features_scaled)[0][1]

# Ejemplo de uso
predictor = TestRiskPredictor()

# Datos de entrenamiento (cambios históricos y resultados de pruebas)
training_data = pd.DataFrame([
    {'additions': 10, 'deletions': 5, 'changed_files': 2, 'complexity': 3,
     'author_commits': 50, 'hours_since_change': 2, 'dependency_count': 4,
     'past_failures': 0.1, 'test_failed': 0},
    {'additions': 150, 'deletions': 80, 'changed_files': 8, 'complexity': 12,
     'author_commits': 5, 'hours_since_change': 48, 'dependency_count': 15,
     'past_failures': 0.3, 'test_failed': 1},
    # ... más datos históricos
])

predictor.train(training_data)

# Predecir riesgo para nuevo cambio
new_change = {
    'additions': 75, 'deletions': 30, 'changed_files': 4,
    'complexity': 8, 'author_commits': 20, 'hours_since_change': 12,
    'dependency_count': 8, 'past_failures': 0.15
}

risk_score = predictor.predict_risk(new_change)
print(f"Riesgo de fallo de prueba: {risk_score:.2%}")

from typing import List, Set, Tuple
from dataclasses import dataclass

@dataclass
class TestCase:
    name: str
    file_path: str
    execution_time: float
    last_failure_date: str = None
    failure_rate: float = 0.0

class IntelligentTestSelector:
    def __init__(self, dependency_graph, risk_predictor):
        self.dependency_graph = dependency_graph
        self.risk_predictor = risk_predictor
        self.test_cases = []

    def select_tests(self, changed_files: Set[str],
                    time_budget: float = None,
                    min_confidence: float = 0.7) -> List[TestCase]:
        """
        Seleccionar pruebas usando toma de decisiones multicriterio
        """
        # Paso 1: Encontrar pruebas directamente afectadas
        affected_modules = self.dependency_graph.find_affected_modules(changed_files)
        candidate_tests = self._get_tests_for_modules(affected_modules)

        # Paso 2: Calcular puntuaciones de riesgo
        scored_tests = []
        for test in candidate_tests:
            risk_score = self._calculate_test_priority(test, changed_files)
            scored_tests.append((test, risk_score))

        # Paso 3: Ordenar por riesgo (descendente)
        scored_tests.sort(key=lambda x: x[1], reverse=True)

        # Paso 4: Aplicar restricción de presupuesto de tiempo
        selected_tests = []
        total_time = 0.0

        for test, score in scored_tests:
            if time_budget and total_time + test.execution_time > time_budget:
                if score >= min_confidence:
                    # Prueba de alto riesgo excede presupuesto - advertir al usuario
                    print(f"Advertencia: Prueba de alto riesgo {test.name} excluida por presupuesto de tiempo")
                continue

            selected_tests.append(test)
            total_time += test.execution_time

        return selected_tests

    def _calculate_test_priority(self, test: TestCase,
                                 changed_files: Set[str]) -> float:
        """
        Calcular puntuación de prioridad combinando múltiples factores
        """
        # Factor 1: Tasa de fallo histórica (0-1)
        failure_weight = test.failure_rate

        # Factor 2: Distancia de dependencia (más cerca = mayor prioridad)
        distance = self._calculate_dependency_distance(test, changed_files)
        distance_weight = 1.0 / (1.0 + distance)

        # Factor 3: Predicción de riesgo basada en ML
        risk_weight = self._get_ml_risk_score(test, changed_files)

        # Factor 4: Tiempo de ejecución (pruebas más rápidas = ligero aumento de prioridad)
        time_weight = 0.1 / (test.execution_time + 0.1)

        # Combinación ponderada
        priority = (
            0.35 * failure_weight +
            0.30 * distance_weight +
            0.30 * risk_weight +
            0.05 * time_weight
        )

        return priority

    def _calculate_dependency_distance(self, test: TestCase,
                                       changed_files: Set[str]) -> int:
        """Calcular longitud mínima de ruta de dependencia"""
        min_distance = float('inf')
        for changed_file in changed_files:
            try:
                distance = nx.shortest_path_length(
                    self.dependency_graph.graph,
                    source=test.file_path,
                    target=changed_file
                )
                min_distance = min(min_distance, distance)
            except nx.NetworkXNoPath:
                continue

        return min_distance if min_distance != float('inf') else 10

    def _get_ml_risk_score(self, test: TestCase,
                          changed_files: Set[str]) -> float:
        """Obtener predicción de riesgo basada en ML"""
        # Preparar características para predicción de riesgo
        change_data = {
            'changed_files': len(changed_files),
            'complexity': 5,  # Se calcularía del código real
            'dependency_count': len(self.dependency_graph.graph.neighbors(test.file_path))
        }

        return self.risk_predictor.predict_risk(change_data)

    def _get_tests_for_modules(self, modules: Set[str]) -> List[TestCase]:
        """Obtener todas las pruebas que cubren los módulos especificados"""
        return [t for t in self.test_cases
                if any(m in t.file_path for m in modules)]

name: Selección Inteligente de Pruebas

on:
  pull_request:
    branches: [ main, develop ]

jobs:
  smart-test-selection:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3
      with:
        fetch-depth: 0  # Obtener historial completo para análisis de cambios

    - name: Configurar Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'

    - name: Instalar dependencias
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov

    - name: Analizar cambios de código
      id: analyze
      run: |
        python scripts/analyze_changes.py \
          --base-ref ${{ github.event.pull_request.base.sha }} \
          --head-ref ${{ github.event.pull_request.head.sha }} \
          --output changes.json

    - name: Seleccionar pruebas con IA
      id: select
      run: |
        python scripts/select_tests.py \
          --changes changes.json \
          --time-budget 600 \
          --output selected_tests.txt

    - name: Ejecutar pruebas seleccionadas
      run: |
        pytest $(cat selected_tests.txt) \
          --cov=src \
          --cov-report=xml \
          --junit-xml=test-results.xml

    - name: Subir cobertura
      uses: codecov/codecov-action@v3
      with:
        files: ./coverage.xml

    - name: Comentar PR con resultados
      uses: actions/github-script@v6
      with:
        script: |
          const fs = require('fs');
          const selectedTests = fs.readFileSync('selected_tests.txt', 'utf8');
          const testCount = selectedTests.split('\n').filter(Boolean).length;

          github.rest.issues.createComment({
            issue_number: context.issue.number,
            owner: context.repo.owner,
            repo: context.repo.repo,
            body: `## Resultados de Selección Inteligente de Pruebas\n\n` +
                  `Seleccionadas ${testCount} pruebas basándose en análisis de IA.\n\n` +
                  `Tiempo ahorrado: ~${Math.round((1000 - testCount) / 1000 * 100)}%`
          });

pipeline {
    agent any

    stages {
        stage('Analizar Cambios') {
            steps {
                script {
                    def changes = sh(
                        script: 'python scripts/analyze_changes.py --base-ref origin/main --head-ref HEAD',
                        returnStdout: true
                    ).trim()

                    env.CHANGED_FILES = changes
                }
            }
        }

        stage('Seleccionar Pruebas') {
            steps {
                script {
                    def selectedTests = sh(
                        script: """
                            python scripts/select_tests.py \
                                --changes '${env.CHANGED_FILES}' \
                                --confidence-threshold 0.8
                        """,
                        returnStdout: true
                    ).trim()

                    env.SELECTED_TESTS = selectedTests
                }
            }
        }

        stage('Ejecutar Pruebas') {
            steps {
                sh "pytest ${env.SELECTED_TESTS} --junit-xml=results.xml"
            }
        }

        stage('Respaldo - Ejecutar Todas las Pruebas') {
            when {
                expression { currentBuild.result == 'FAILURE' }
            }
            steps {
                echo "Pruebas seleccionadas fallaron. Ejecutando suite completa..."
                sh "pytest tests/ --junit-xml=full-results.xml"
            }
        }
    }

    post {
        always {
            junit 'results.xml'
        }
    }
}

from dataclasses import dataclass
from typing import List
import time

@dataclass
class TIAMetrics:
    total_tests: int
    selected_tests: int
    execution_time_full: float
    execution_time_selected: float
    true_positives: int  # Pruebas seleccionadas que realmente fallaron
    false_negatives: int  # Pruebas perdidas que habrían fallado
    false_positives: int  # Pruebas seleccionadas que pasaron

    @property
    def selection_rate(self) -> float:
        """Porcentaje de pruebas seleccionadas"""
        return (self.selected_tests / self.total_tests) * 100

    @property
    def time_savings(self) -> float:
        """Porcentaje de tiempo ahorrado"""
        return ((self.execution_time_full - self.execution_time_selected) /
                self.execution_time_full) * 100

    @property
    def precision(self) -> float:
        """Precisión: VP / (VP + FP)"""
        return self.true_positives / (self.true_positives + self.false_positives)

    @property
    def recall(self) -> float:
        """Recall: VP / (VP + FN)"""
        return self.true_positives / (self.true_positives + self.false_negatives)

    @property
    def f1_score(self) -> float:
        """Puntuación F1: Media armónica de precisión y recall"""
        p = self.precision
        r = self.recall
        return 2 * (p * r) / (p + r)

    def print_report(self):
        print("="*50)
        print("Análisis de Impacto de Pruebas - Informe de Rendimiento")
        print("="*50)
        print(f"Total de pruebas: {self.total_tests}")
        print(f"Pruebas seleccionadas: {self.selected_tests} ({self.selection_rate:.1f}%)")
        print(f"Tiempo ahorrado: {self.time_savings:.1f}%")
        print(f"Precisión: {self.precision:.2%}")
        print(f"Recall: {self.recall:.2%}")
        print(f"Puntuación F1: {self.f1_score:.3f}")
        print("="*50)

# Ejemplo de métricas de despliegue en producción
metrics = TIAMetrics(
    total_tests=5000,
    selected_tests=850,
    execution_time_full=7200,  # 2 horas
    execution_time_selected=1080,  # 18 minutos
    true_positives=45,  # Pruebas que fallaron y fueron seleccionadas
    false_negatives=3,  # Pruebas que fallaron pero no fueron seleccionadas
    false_positives=802  # Pruebas que pasaron pero fueron seleccionadas
)

metrics.print_report()