FlakyDetector — мой путь к созданию инструмента для ловли «нестабильных» тестов в Python

Привет! Меня зовут Артём Алимпиев, я Python‑разработчик.
Недавно я столкнулся с тем, что даже идеально написанные тесты порой ведут себя… странно.
Один день они проходят, другой — падают, хотя код не менялся.

Если вы когда‑нибудь ловили такие «призраки» в CI/CD, то знаете, насколько это раздражает.
Так начался мой эксперимент — сделать инструмент, который умеет находить и объяснять, почему тесты становятся нестабильными.
Так родился проект FlakyDetector.

🤔 Почему вообще тесты «флейкают»

Flaky-тесты — это как капризный будильник: иногда срабатывает, иногда — нет.
Основные причины:

• зависимость от порядка запуска тестов;

• асинхронные вызовы без ожидания;

• глобальные переменные, влияющие на соседние тесты;

• сетевые операции без моков.

Исследование “An Empirical Study of Flaky Tests in Python” ссылка на статью https://arxiv.org/abs/2101.09077 показало, что около 59% флейков появляются именно из-за порядка выполнения тестов.

Это вдохновило меня сделать инструмент, который сможет анализировать тесты автоматически, находить закономерности и помогать командам их устранять.

🎯 Что я хотел получить

Идея была простой:
создать инструмент, который:

1. собирает данные о тестах из CI;

2. анализирует причины нестабильности;

3. классифицирует их по типу ошибок;

4. показывает отчёты и графики;

5. интегрируется в CI/CD без боли.

⚙️ Как это работает

Архитектура проекта выглядит так:

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   Сбор данных   │────│   Анализ кода    │────│  Классификация  │
└─────────────────┘    └──────────────────┘    └─────────────────┘
         │                       │                       │
         └───────────────────────┼───────────────────────┘
                                 │
                     ┌───────────┴───────────┐
                     │    Визуализация и     │
                     │      отчетность       │
                     └───────────────────────┘

📘 Визуальная схема проекта:

🧩 Основные модули

1. Dataset Collector

Собирает данные из CI/CD (например, GitHub Actions):

• парсит логи тестов;

• обходит лимиты API;

• собирает статистику стабильности.

while True:
    response = self._make_request(page)
    if not response:
        break
    runs.extend(response)
    page += 1
    time.sleep(1)  # соблюдаем лимиты API

2. Log Analyzer

Анализирует текст ошибок и определяет характерные паттерны:

import re
from typing import List, Dict, Any
from collections import defaultdict
import statistics


class LogAnalyzer:
    def __init__(self):
        self.patterns = {
            'async_issue': r'(RuntimeWarning: coroutine.*was never awaited|'
                           r'was never awaited|coroutine.*never awaited)',
            'timing_issue': r'(timeout|timed out|slow response|took too long)',
            'network_issue': r'(ConnectionError|TimeoutError|NetworkError|'
                             r'502 Bad Gateway|503 Service Unavailable)',
            'concurrency_issue': r'(race condition|deadlock|lock timeout|'
                                 r'database lock|concurrent modification)',
            'order_dependency': r'(test_order|depends_on|setUpClass|'
                                r'tearDownClass)'
        }

    def analyze_test_logs(self, test_runs: List[Dict]) -> Dict[str, Any]:
        """Анализирует историю запусков теста"""
        if not test_runs:
            return {}

        flaky_metrics = {
            'total_runs': len(test_runs),
            'pass_count': sum(1 for run in test_runs if run.get('status') == 'PASS'),
            'fail_count': sum(1 for run in test_runs if run.get('status') == 'FAIL'),
            'flaky_rate': 0.0,
            'timing_std': 0.0,
            'error_patterns': defaultdict(int),
            'suspicious_patterns': []
        }

        # Расчет flaky rate
        if flaky_metrics['total_runs'] > 0:
            flaky_metrics['flaky_rate'] = (
                    flaky_metrics['fail_count'] / flaky_metrics['total_runs']
            )

        # Анализ времени выполнения
        durations = [run.get('duration', 0) for run in test_runs if run.get('duration')]
        if len(durations) > 1:
            flaky_metrics['timing_std'] = statistics.stdev(durations)

        # Анализ паттернов ошибок
        for run in test_runs:
            if run.get('status') == 'FAIL' and run.get('error_message'):
                error_msg = run['error_message']
                for pattern_type, pattern in self.patterns.items():
                    if re.search(pattern, error_msg, re.IGNORECASE):
                        flaky_metrics['error_patterns'][pattern_type] += 1

        return flaky_metrics

3. CatBoost Classifier

Модуль машинного обучения, который классифицирует flaky-тесты по типу:

• ASYNC_ISSUE

• TIMING_ISSUE

• NETWORK_ISSUE

• ORDER_DEPENDENCY

• GLOBAL_STATE

Точность модели: ≈87%
Поддерживает >10 000 тестов, анализирует один файл <100 мс.

import pandas as pd
import numpy as np
from catboost import CatBoostClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, accuracy_score
import joblib


class FlakyTestClassifier:
    def __init__(self):
        self.model = None
        self.feature_columns = [
            'flaky_rate', 'timing_std', 'has_async_calls', 'has_global_vars',
            'has_time_sleep', 'network_calls_count', 'db_operations_count',
            'test_duration_avg', 'error_pattern_count', 'concurrency_indicators'
        ]
        self.target_classes = [
            'ASYNC_ISSUE', 'TIMING_ISSUE', 'NETWORK_ISSUE',
            'CONCURRENCY_ISSUE', 'ORDER_DEPENDENCY', 'GLOBAL_STATE', 'NON_FLAKY'
        ]

    def prepare_features(self, raw_data: List[Dict]) -> pd.DataFrame:
        """Подготавливает признаки для модели"""
        features = []

        for item in raw_data:
            feature_vector = [
                item.get('flaky_rate', 0),
                item.get('timing_std', 0),
                int(item.get('has_async_issues', False)),
                int(item.get('has_global_variables', False)),
                int(item.get('has_time_sleep', False)),
                item.get('network_calls', 0),
                item.get('db_operations', 0),
                item.get('avg_duration', 0),
                item.get('error_patterns_count', 0),
                item.get('concurrency_indicators', 0)
            ]
            features.append(feature_vector)

        return pd.DataFrame(features, columns=self.feature_columns)

    def train(self, X: pd.DataFrame, y: pd.Series):
        """Обучает модель классификатора"""
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42, stratify=y
        )

        self.model = CatBoostClassifier(
            iterations=1000,
            learning_rate=0.1,
            depth=6,
            random_seed=42,
            verbose=False
        )

        self.model.fit(
            X_train, y_train,
            eval_set=(X_test, y_test),
            early_stopping_rounds=50,
            verbose=100
        )

        # Валидация модели
        y_pred = self.model.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        print(f"Model accuracy: {accuracy:.4f}")
        print(classification_report(y_test, y_pred, target_names=self.target_classes))

    def predict(self, features: pd.DataFrame) -> tuple:
        """Предсказывает класс и вероятность"""
        if self.model is None:
            raise ValueError("Model not trained yet")

        predictions = self.model.predict(features)
        probabilities = self.model.predict_proba(features)

        return predictions, probabilities

    def save_model(self, filepath: str):
        """Сохраняет обученную модель"""
        if self.model:
            joblib.dump(self.model, filepath)

    def load_model(self, filepath: str):
        """Загружает обученную модель"""
        self.model = joblib.load(filepath)

4. Визуализация

Фронтенд на React + Recharts:

• графики flaky rate;

• причины сбоев;

• рекомендации по исправлению.

📈 Результаты

🔮 В планах

• улучшение документации;

• поддержка Django, Flask, Tornado;

• интеграция с другими CI/CD системами;

• предсказание flaky-тестов с помощью LLM;

• автоматическая генерация патчей.

🧠 Идея проекта

FlakyDetector — это попытка соединить статический анализ, машинное обучение и визуализацию в одном инструменте.

Он помогает разработчикам понимать причины нестабильности тестов и делает тестирование умнее и надёжнее.

Проект всё ещё развивается, документация в процессе доработки, но уже сейчас FlakyDetector показывает отличные результаты в реальных проектах.

👨‍💻 Автор: Артём Алимпиев