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

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

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)