Detección de Tests Flaky con Machine Learning: Combatiendo Tests Inestables

execution_features = {
    'pass_rate': 0.73,  # Pasa 73% del tiempo
    'consecutive_failures': 2,
    'failure_variability': 0.45,  # Alta varianza en resultados
    'avg_execution_time': 12.3,
    'execution_time_stddev': 4.2,  # Alta varianza de tiempo
    'last_10_outcomes': [1,1,0,1,1,0,1,1,1,0],  # 1=pasa, 0=falla
}

import ast

def extract_code_features(test_code):
    tree = ast.parse(test_code)

    return {
        'uses_sleep': 'time.sleep' in test_code,
        'uses_random': 'random' in test_code,
        'async_count': test_code.count('async '),
        'network_calls': test_code.count('requests.') + test_code.count('httpx.'),
        'database_queries': test_code.count('execute('),
        'wait_statements': test_code.count('WebDriverWait'),
        'thread_usage': test_code.count('Thread('),
        'external_deps': test_code.count('mock.') == 0,  # Sin mocking
        'assertion_count': test_code.count('assert'),
        'test_length': len(test_code.split('\n')),
    }

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import pandas as pd

# Cargar dataset etiquetado
data = pd.read_csv('test_history.csv')

# Separar características y etiquetas
X = data.drop(['test_id', 'is_flaky'], axis=1)
y = data['is_flaky']

# Dividir datos
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# Entrenar modelo
model = RandomForestClassifier(
    n_estimators=100,
    max_depth=10,
    random_state=42
)
model.fit(X_train, y_train)

# Evaluar
from sklearn.metrics import classification_report

y_pred = model.predict(X_test)
print(classification_report(y_test, y_pred))

# Importancia de características
feature_importance = pd.DataFrame({
    'feature': X.columns,
    'importance': model.feature_importances_
}).sort_values('importance', ascending=False)

print("\nPrincipales indicadores de test flaky:")
print(feature_importance.head(10))

from sklearn.ensemble import IsolationForest
import numpy as np

class FlakyTestAnomalyDetector:
    def __init__(self, contamination=0.1):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42
        )

    def fit(self, test_execution_history):
        features = self._extract_time_series_features(test_execution_history)
        self.model.fit(features)
        return self

    def _extract_time_series_features(self, history):
        features = []
        for test_id in history['test_id'].unique():
            test_data = history[history['test_id'] == test_id]
            outcomes = test_data['outcome'].values
            times = test_data['execution_time'].values

            features.append({
                'pass_rate': np.mean(outcomes),
                'pass_rate_variance': np.var(outcomes),
                'execution_time_mean': np.mean(times),
                'execution_time_variance': np.var(times),
            })

        return pd.DataFrame(features)

    def predict_flaky(self, test_execution_history):
        features = self._extract_time_series_features(test_execution_history)
        predictions = self.model.predict(features)
        anomaly_scores = self.model.score_samples(features)

        results = pd.DataFrame({
            'test_id': test_execution_history['test_id'].unique(),
            'anomaly_score': anomaly_scores,
            'is_flaky': predictions == -1
        })

        return results.sort_values('anomaly_score')

import tensorflow as tf
from tensorflow import keras

class FlakyTestPredictor:
    def __init__(self, sequence_length=20):
        self.sequence_length = sequence_length
        self.model = self._build_model()

    def _build_model(self):
        model = keras.Sequential([
            keras.layers.LSTM(64, input_shape=(self.sequence_length, 1)),
            keras.layers.Dropout(0.2),
            keras.layers.Dense(32, activation='relu'),
            keras.layers.Dense(1, activation='sigmoid')
        ])

        model.compile(
            optimizer='adam',
            loss='binary_crossentropy',
            metrics=['accuracy']
        )

        return model

    def predict_stability(self, recent_outcomes):
        if len(recent_outcomes) < self.sequence_length:
            return None

        X = np.array(recent_outcomes[-self.sequence_length:])
        X = X.reshape(1, self.sequence_length, 1)

        prediction = self.model.predict(X, verbose=0)[0][0]

        # Alta varianza en predicciones indica flakiness
        predictions = []
        for _ in range(10):
            pred = self.model.predict(X, verbose=0)[0][0]
            predictions.append(pred)

        prediction_variance = np.var(predictions)

        return {
            'next_pass_probability': prediction,
            'prediction_variance': prediction_variance,
            'likely_flaky': prediction_variance > 0.05 or (0.3 < prediction < 0.7)
        }

class FlakyTestDetectionPipeline:
    def __init__(self):
        self.classifier = RandomForestClassifier()
        self.anomaly_detector = FlakyTestAnomalyDetector()

    def detect_flaky_tests(self, execution_data):
        # 1. Análisis estadístico
        stats = self._calculate_test_statistics(execution_data)

        # 2. Clasificación ML
 (como se discute en [AI Test Metrics Analytics: Intelligent Analysis of QA Metrics](/es/blog/ai-test-metrics))        ml_predictions = self._ml_classification(stats)

        # 3. Detección de anomalías
        anomalies = self.anomaly_detector.predict_flaky(execution_data)

        # Combinar resultados
        flaky_tests = self._combine_predictions(stats, ml_predictions, anomalies)

        return flaky_tests

    def _calculate_test_statistics(self, data):
        stats = []
        for test_id in data['test_id'].unique():
            test_data = data[data['test_id'] == test_id]
            outcomes = test_data['outcome'].values

            stats.append({
                'test_id': test_id,
                'pass_rate': np.mean(outcomes),
                'total_runs': len(outcomes),
                'failures': np.sum(outcomes == 0),
                'variance': np.var(outcomes),
            })

        return pd.DataFrame(stats)

    def _combine_predictions(self, stats, ml_preds, anomalies):
        results = stats.copy()

        # Votación simple: test es flaky si 2+ métodos están de acuerdo
        results['statistical_flaky'] = (results['pass_rate'] < 0.95) & (results['pass_rate'] > 0.05)
        results['ml_flaky'] = ml_preds if ml_preds is not None else False
        results['anomaly_flaky'] = anomalies['is_flaky'].values

        results['votes'] = (
            results['statistical_flaky'].astype(int) +
            results['ml_flaky'].astype(int) +
            results['anomaly_flaky'].astype(int)
        )

        results['is_flaky'] = results['votes'] >= 2

        return results[results['is_flaky']].sort_values('pass_rate')

# .github/workflows/flaky-detection.yml
name: Detección de Tests Flaky

on:
  schedule:

    - cron: '0 2 * * *'  # Diario a las 2 AM
  workflow_dispatch:

jobs:
  detect-flaky:
    runs-on: ubuntu-latest
    steps:

      - uses: actions/checkout@v3

      - name: Ejecutar tests con detección flaky
        run: |
          pip install flaky-detector pytest pytest-json-report
          pytest --json-report --json-report-file=report.json

      - name: Analizar tests flaky
        run: |
          python scripts/detect_flaky_tests.py --report report.json

      - name: Crear issue para tests flaky
        if: env.FLAKY_TESTS_FOUND == 'true'
        uses: actions/github-script@v6
        with:
          script: |
            const fs = require('fs');
            const flaky = JSON.parse(fs.readFileSync('flaky_tests.json'));

            let body = '## 🚨 Tests Flaky Detectados\n\n';
            flaky.forEach(test => {
              body += `### ${test.test_id}\n`;
              body += `- Tasa de paso: ${(test.pass_rate * 100).toFixed(1)}%\n`;
              body += `- Causa probable: ${test.root_cause}\n\n`;
            });

            github.rest.issues.create({
              owner: context.repo.owner,
              repo: context.repo.repo,
              title: 'Tests Flaky Detectados',
              body: body,
              labels: ['flaky-test', 'quality']
            });

def suggest_fixes(test_code, failure_patterns):
    suggestions = []

    if 'time.sleep' in test_code:
        suggestions.append({
            'issue': 'Usa time.sleep (no determinista)',
            'fix': 'Reemplazar con WebDriverWait explícito',
            'example': 'WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.ID, "submit")))'
        })

    if 'click()' in test_code and 'wait' not in test_code.lower():
        suggestions.append({
            'issue': 'Click sin espera explícita',
            'fix': 'Agregar espera antes de hacer click',
            'example': 'wait.until(EC.element_to_be_clickable(element)).click()'
        })

    if 'requests.get' in test_code and 'mock' not in test_code:
        suggestions.append({
            'issue': 'Dependencia de API externa',
            'fix': 'Mockear llamadas de API externas',
            'example': '@mock.patch("requests.get")\ndef test_api(mock_get): ...'
        })

    return suggestions