Selección Predictiva de Tests: Optimización de Pruebas con IA

class MapeadorCodigoTest:
    def __init__(self):
        self.mapa_codigo_test = defaultdict(set)
        self.cobertura_test = {}

    def analizar_cobertura(self, datos_ejecucion_test):
        """Construir mapeo desde datos de cobertura"""
        for nombre_test, datos_cobertura in datos_ejecucion_test.items():
            archivos_cubiertos = datos_cobertura['archivos']

            for ruta_archivo in archivos_cubiertos:
                self.mapa_codigo_test[ruta_archivo].add(nombre_test)

    def obtener_tests_afectados(self, archivos_cambiados):
        """Obtener tests afectados por cambios de código"""
        afectados = set()

        for ruta_archivo in archivos_cambiados:
            afectados.update(self.mapa_codigo_test.get(ruta_archivo, set()))

        return list(afectados)

from sklearn.ensemble import RandomForestClassifier

class PredictorFallosTest:
    def __init__(self):
        self.modelo = RandomForestClassifier(n_estimators=100)

    def extraer_caracteristicas(self, commit, test):
        """Extraer características para predicción"""
        return {
            'archivos_cambiados': len(commit['archivos']),
            'lineas_agregadas': commit['adiciones'],
            'lineas_eliminadas': commit['eliminaciones'],
            'tiempo_ejecucion_test_ms': test['duracion_promedio'],
            'puntaje_inestabilidad_test': test['inestabilidad'],
            'tasa_fallo_30d': test['fallos_ultimos_30_dias'] / test['ejecuciones_ultimos_30_dias']
        }

    def predecir_probabilidad_fallo(self, commit, test):
        """Predecir probabilidad de que test falle"""
        caracteristicas = self.extraer_caracteristicas(commit, test)
        vector_caracteristicas = [list(caracteristicas.values())]

        probabilidad = self.modelo.predict_proba(vector_caracteristicas)[0][1]

        return {
            'test': test['nombre'],
            'probabilidad_fallo': probabilidad
        }

class PriorizadorTests:
    def calcular_valor_test(self, test, commit):
        """Calcular puntuación de valor para test"""
        prob_fallo = self.predictor.predecir_probabilidad_fallo(commit, test)['probabilidad_fallo']
        cobertura_codigo = test['cobertura_lineas'] / lineas_totales
        historial_deteccion_bugs = test['bugs_atrapados_ultimo_año']
        costo_ejecucion = test['duracion_promedio_ms'] / 1000

        # Valor = (Riesgo Fallo × Cobertura × Historial Bugs) / Costo
        puntuacion_valor = (prob_fallo * cobertura_codigo * historial_deteccion_bugs) / max(costo_ejecucion, 1)

        return puntuacion_valor

    def priorizar(self, commit, presupuesto_tiempo_segundos):
        """Seleccionar tests para maximizar valor dentro de presupuesto de tiempo"""
        todos_tests = self.mapeador.obtener_catalogo_tests()

        # Calcular valor para cada test
        puntuaciones_tests = [
            {
                'test': test,
                'valor': self.calcular_valor_test(test, commit),
                'duracion': test['duracion_promedio_ms'] / 1000
            }
            for test in todos_tests
        ]

        # Ordenar por valor (descendente)
        puntuaciones_tests.sort(key=lambda x: x['valor'], reverse=True)

        # Selección greedy dentro de presupuesto
        tests_seleccionados = []
        tiempo_total = 0

        for item in puntuaciones_tests:
            if tiempo_total + item['duracion'] <= presupuesto_tiempo_segundos:
                tests_seleccionados.append(item['test'])
                tiempo_total += item['duracion']

        return {
            'tests_seleccionados': tests_seleccionados,
            'duracion_estimada': tiempo_total,
            'cobertura': len(tests_seleccionados) / len(todos_tests)
        }

name: Selección Inteligente de Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:

    - uses: actions/checkout@v3
      with:
        fetch-depth: 0

    - name: Analizar Cambios de Código
      id: changes
      run: |
        CHANGED_FILES=$(git diff --name-only ${{ github.event.before }} ${{ github.sha }})
        echo "files=$CHANGED_FILES" >> $GITHUB_OUTPUT

    - name: Predecir Selección de Tests
      id: selection
      run: |
        python predecir_tests.py \
          --archivos-cambiados "$CHANGED_FILES" \
          --presupuesto-tiempo 600

    - name: Ejecutar Tests Seleccionados
      run: |
        pytest $(cat tests_seleccionados.json | jq -r '.tests[]')

class AnalizadorImpactoTests:
    def __init__(self):
        self.grafo_impacto = nx.DiGraph()

    def construir_grafo_impacto(self, base_codigo):
        """Construir grafo de dependencias"""
        for archivo in base_codigo.archivos:
            self.grafo_impacto.add_node(archivo.ruta, tipo='codigo')

        for test in base_codigo.tests:
            self.grafo_impacto.add_node(test.nombre, tipo='test')
            for archivo_cubierto in test.cobertura:
                self.grafo_impacto.add_edge(archivo_cubierto, test.nombre)

    def obtener_tests_impactados(self, archivos_cambiados):
        """Encontrar todos los tests impactados transitivamente"""
        impactados = set()

        for archivo_cambiado in archivos_cambiados:
            alcanzables = nx.descendants(self.grafo_impacto, archivo_cambiado)

            for nodo in alcanzables:
                if self.grafo_impacto.nodes[nodo]['tipo'] == 'test':
                    impactados.add(nodo)

        return list(impactados)

class MetricasSeleccion:
    def registrar_seleccion(self, commit, seleccionados, omitidos, resultados):
        """Registrar efectividad de selección"""
        fallos_seleccionados = [t for t in seleccionados if resultados[t] == 'fallido']
        fallos_omitidos = [t for t in omitidos if resultados[t] == 'fallido']

        self.metricas.append({
            'commit': commit,
            'tests_seleccionados': len(seleccionados),
            'tests_omitidos': len(omitidos),
            'tiempo_ahorrado_porcentaje': len(omitidos) / (len(seleccionados) + len(omitidos)),
            'fallos_capturados': len(fallos_seleccionados),
            'fallos_perdidos': len(fallos_omitidos),
            'precision': len(fallos_seleccionados) / len(seleccionados) if seleccionados else 0,
            'recall': len(fallos_seleccionados) / (len(fallos_seleccionados) + len(fallos_omitidos)) if (fallos_seleccionados or fallos_omitidos) else 1.0
        })