Непрерывное Обучение в Автоматизации Тестирования: Создание Самосовершенствующихся Тестовых Систем

class TestFeedbackLoop:
    def __init__(self, model_store, metric_collector):
        self.model_store = model_store
        self.metric_collector = metric_collector
        self.learning_buffer = []

    def execute_test_cycle(self, test_suite):
        """Выполнить тесты и собрать обратную связь"""
        results = []

        for test in test_suite:
            # Выполнить тест
            outcome = test.run()

            # Собрать контекстные данные
            context = {
                'test_id': test.id,
                'execution_time': outcome.duration,
                'failure_reason': outcome.error_message,
                'environment': test.environment,
                'timestamp': outcome.timestamp,
                'flakiness_score': self._calculate_flakiness(test.id)
            }

            # Добавить в буфер обучения
            self.learning_buffer.append({
                'outcome': outcome.status,
                'context': context,
                'test_characteristics': test.get_features()
            })

            results.append(outcome)

        return results

    def learn_from_feedback(self):
        """Обработать обратную связь и обновить модели"""
        if len(self.learning_buffer) < 100:
            return  # Дождаться достаточных данных

        # Извлечь паттерны
        patterns = self._extract_patterns(self.learning_buffer)

        # Обновить прогнозные модели
        self.model_store.update_models({
            'failure_predictor': patterns['failure_patterns'],
            'flakiness_detector': patterns['flakiness_patterns'],
            'execution_time_estimator': patterns['timing_patterns']
        })

        # Очистить обработанную обратную связь
        self.learning_buffer.clear()

    def _extract_patterns(self, feedback_data):
        """Извлечь действенные паттерны из обратной связи"""
        from sklearn.cluster import DBSCAN
        import numpy as np

        # Извлечь признаки для кластеризации
        features = np.array([
            [
                item['context']['execution_time'],
                item['context']['flakiness_score'],
                hash(item['context']['environment']) % 1000
            ]
            for item in feedback_data
        ])

        # Идентифицировать кластеры сбоев
        clustering = DBSCAN(eps=0.3, min_samples=5).fit(features)

        return {
            'failure_patterns': clustering.labels_,
            'flakiness_patterns': self._detect_flakiness_patterns(feedback_data),
            'timing_patterns': self._analyze_timing_trends(feedback_data)
        }

from river import tree, metrics, ensemble
import datetime

class OnlineTestOptimizer:
    def __init__(self):
        # Адаптивный случайный лес для прогнозирования сбоев
        self.failure_model = ensemble.AdaptiveRandomForestClassifier(
            n_models=10,
            max_features='sqrt',
            lambda_value=6
        )

        # Дерево Хоффдинга для выбора тестов
        self.selection_model = tree.HoeffdingTreeClassifier()

        # Метрики производительности
        self.failure_metric = metrics.Accuracy()
        self.selection_metric = metrics.Precision()

    def predict_failure_probability(self, test_features):
        """Предсказать вероятность сбоя теста"""
        return self.failure_model.predict_proba_one(test_features)

    def update_from_execution(self, test_features, actual_outcome):
        """Обучиться на одном выполнении теста"""
        # Обновить модель прогнозирования сбоев
        self.failure_model.learn_one(test_features, actual_outcome['failed'])

        # Обновить метрики
        prediction = self.failure_model.predict_one(test_features)
        self.failure_metric.update(actual_outcome['failed'], prediction)

    def select_tests(self, available_tests, budget):
        """Адаптивно выбрать наиболее ценные тесты в рамках бюджета"""
        test_scores = []

        for test in available_tests:
            features = test.extract_features()

            # Предсказать вероятность сбоя
            fail_prob = self.predict_failure_probability(features).get(True, 0.0)

            # Вычислить оценку ценности
            value_score = self._calculate_test_value(
                fail_probability=fail_prob,
                execution_cost=test.estimated_duration,
                code_coverage=test.coverage_metrics,
                last_execution=test.last_run_timestamp
            )

            test_scores.append((test, value_score))

        # Выбрать тесты с наивысшей ценностью в пределах бюджета
        test_scores.sort(key=lambda x: x[1], reverse=True)

        selected = []
        total_cost = 0

        for test, score in test_scores:
            if total_cost + test.estimated_duration <= budget:
                selected.append(test)
                total_cost += test.estimated_duration

        return selected

    def _calculate_test_value(self, fail_probability, execution_cost,
                             code_coverage, last_execution):
        """Вычислить метрику ценности для приоритизации тестов"""
        # Фактор временного затухания
        hours_since_execution = (
            datetime.datetime.now() - last_execution
        ).total_seconds() / 3600
        recency_factor = 1 / (1 + hours_since_execution / 24)

        # Вычисление ценности
        value = (
            fail_probability * 0.4 +          # Вероятность сбоя
            code_coverage * 0.3 +              # Важность покрытия
            recency_factor * 0.2 +             # Бонус за свежесть
            (1 / execution_cost) * 0.1         # Фактор эффективности
        )

        return value

import re
from collections import defaultdict, Counter
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import AgglomerativeClustering

class FailurePatternLearner:
    def __init__(self):
        self.failure_history = []
        self.pattern_database = defaultdict(list)
        self.vectorizer = TfidfVectorizer(max_features=100)

    def analyze_failure(self, test_failure):
        """Извлечь и категоризировать паттерны сбоев"""
        # Проанализировать stack trace
        stack_trace = test_failure.stack_trace
        error_message = test_failure.error_message

        # Извлечь ключевые элементы
        failure_signature = {
            'exception_type': self._extract_exception_type(error_message),
            'failing_component': self._extract_component(stack_trace),
            'error_keywords': self._extract_keywords(error_message),
            'stack_depth': len(stack_trace.split('\n')),
            'timestamp': test_failure.timestamp,
            'environment': test_failure.environment
        }

        # Сохранить сбой
        self.failure_history.append({
            'signature': failure_signature,
            'full_context': test_failure
        })

        # Обновить базу данных паттернов
        pattern_key = f"{failure_signature['exception_type']}_{failure_signature['failing_component']}"
        self.pattern_database[pattern_key].append(failure_signature)

        # Проверить повторяющиеся паттерны
        if len(self.pattern_database[pattern_key]) >= 3:
            return self._generate_pattern_alert(pattern_key)

        return None

    def cluster_similar_failures(self, recent_failures=100):
        """Сгруппировать похожие сбои с помощью кластеризации"""
        if len(self.failure_history) < 10:
            return []

        # Получить недавние сбои
        recent = self.failure_history[-recent_failures:]

        # Создать текстовые представления
        failure_texts = [
            f"{f['signature']['exception_type']} {' '.join(f['signature']['error_keywords'])}"
            for f in recent
        ]

        # Векторизация
        vectors = self.vectorizer.fit_transform(failure_texts)

        # Кластеризация
        clustering = AgglomerativeClustering(
            n_clusters=None,
            distance_threshold=0.5,
            linkage='average'
        )
        labels = clustering.fit_predict(vectors.toarray())

        # Группировка по кластерам
        clusters = defaultdict(list)
        for idx, label in enumerate(labels):
            clusters[label].append(recent[idx])

        return clusters

    def suggest_fixes(self, failure_signature):
        """Предложить потенциальные исправления на основе исторических паттернов"""
        # Найти похожие прошлые сбои
        similar_failures = self._find_similar_failures(failure_signature)

        # Извлечь общие паттерны решения
        resolutions = []
        for similar in similar_failures:
            if similar.get('resolution'):
                resolutions.append(similar['resolution'])

        # Ранжировать по частоте
        resolution_counts = Counter(resolutions)

        suggestions = []
        for resolution, count in resolution_counts.most_common(3):
            confidence = count / len(similar_failures)
            suggestions.append({
                'action': resolution,
                'confidence': confidence,
                'evidence_count': count
            })

        return suggestions

    def _extract_exception_type(self, error_message):
        """Извлечь тип исключения из сообщения об ошибке"""
        match = re.search(r'(\w+Exception|\w+Error)', error_message)
        return match.group(1) if match else 'UnknownException'

    def _extract_component(self, stack_trace):
        """Идентифицировать неисправный компонент из stack trace"""
        lines = stack_trace.split('\n')
        for line in lines:
            if 'at ' in line and 'test' not in line.lower():
                match = re.search(r'at ([\w.]+)', line)
                if match:
                    return match.group(1).split('.')[0]
        return 'UnknownComponent'

    def _extract_keywords(self, error_message):
        """Извлечь значимые ключевые слова из сообщения об ошибке"""
        # Удалить общие слова
        stop_words = {'the', 'a', 'an', 'in', 'to', 'of', 'at', 'for'}
        words = re.findall(r'\b[a-z]{3,}\b', error_message.lower())
        return [w for w in words if w not in stop_words][:5]

from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException
import Levenshtein

class SelfHealingLocator:
    def __init__(self, driver, learning_rate=0.1):
        self.driver = driver
        self.locator_history = {}
        self.learning_rate = learning_rate

    def find_element(self, locator_strategies, element_context):
        """Попробовать множество стратегий и учиться на успехах"""
        element_id = element_context['id']

        # Сначала попробовать лучшую историческую стратегию
        if element_id in self.locator_history:
            best_strategy = self.locator_history[element_id]['best']
            try:
                element = self._try_locator(best_strategy)
                self._update_success(element_id, best_strategy)
                return element
            except NoSuchElementException:
                self._update_failure(element_id, best_strategy)

        # Попробовать все стратегии
        for strategy in locator_strategies:
            try:
                element = self._try_locator(strategy)

                # Учиться на успехе
                self._learn_successful_strategy(element_id, strategy, element)
                return element

            except NoSuchElementException:
                continue

        # Все стратегии не сработали, попытаться восстановление
        return self._attempt_healing(locator_strategies, element_context)

    def _try_locator(self, strategy):
        """Попытаться найти элемент с данной стратегией"""
        by_type, value = strategy
        return self.driver.find_element(by_type, value)

    def _attempt_healing(self, failed_strategies, element_context):
        """Попытаться восстановить локатор, найдя похожие элементы"""
        # Получить все элементы на странице
        all_elements = self.driver.find_elements(By.XPATH, '//*')

        # Оценить элементы по схожести с ожидаемым контекстом
        candidates = []
        for elem in all_elements:
            score = self._calculate_similarity(elem, element_context)
            if score > 0.7:  # Порог для рассмотрения
                candidates.append((elem, score))

        if not candidates:
            raise NoSuchElementException(f"Не удалось восстановить локатор для {element_context['id']}")

        # Вернуть лучшее совпадение
        candidates.sort(key=lambda x: x[1], reverse=True)
        healed_element = candidates[0][0]

        # Выучить новую стратегию локатора
        new_strategy = self._generate_locator_from_element(healed_element)
        self._learn_successful_strategy(
            element_context['id'],
            new_strategy,
            healed_element
        )

        return healed_element

    def _calculate_similarity(self, element, context):
        """Вычислить оценку схожести между элементом и ожидаемым контекстом"""
        score = 0.0

        # Схожесть текста
        if context.get('text'):
            elem_text = element.text.lower()
            expected_text = context['text'].lower()
            text_sim = 1 - (Levenshtein.distance(elem_text, expected_text) /
                           max(len(elem_text), len(expected_text)))
            score += text_sim * 0.4

        # Схожесть атрибутов
        if context.get('attributes'):
            for attr, value in context['attributes'].items():
                elem_value = element.get_attribute(attr)
                if elem_value:
                    attr_sim = 1 - (Levenshtein.distance(elem_value.lower(), value.lower()) /
                                   max(len(elem_value), len(value)))
                    score += attr_sim * 0.3

        # Схожесть позиции
        if context.get('position'):
            elem_location = element.location
            expected_location = context['position']
            position_diff = abs(elem_location['x'] - expected_location['x']) + \
                          abs(elem_location['y'] - expected_location['y'])
            position_sim = 1 / (1 + position_diff / 100)
            score += position_sim * 0.3

        return score

    def _learn_successful_strategy(self, element_id, strategy, element):
        """Обновить модель обучения успешной стратегией"""
        if element_id not in self.locator_history:
            self.locator_history[element_id] = {
                'strategies': {},
                'best': strategy
            }

        strategies = self.locator_history[element_id]['strategies']

        # Обновить оценку стратегии
        if strategy not in strategies:
            strategies[strategy] = {'successes': 0, 'failures': 0, 'score': 0.5}

        strategies[strategy]['successes'] += 1
        strategies[strategy]['score'] = (
            strategies[strategy]['successes'] /
            (strategies[strategy]['successes'] + strategies[strategy]['failures'])
        )

        # Обновить лучшую стратегию
        best_score = max(s['score'] for s in strategies.values())
        for strat, data in strategies.items():
            if data['score'] == best_score:
                self.locator_history[element_id]['best'] = strat
                break

import numpy as np
from collections import deque
import random

class RLTestSelector:
    def __init__(self, num_tests, learning_rate=0.1, discount_factor=0.95):
        self.num_tests = num_tests
        self.lr = learning_rate
        self.gamma = discount_factor
        self.epsilon = 1.0  # Коэффициент исследования
        self.epsilon_decay = 0.995
        self.epsilon_min = 0.01

        # Q-таблица: состояние -> действие -> значение
        self.q_table = {}

        # Воспроизведение опыта
        self.memory = deque(maxlen=2000)

    def get_state(self, test_suite):
        """Сгенерировать представление состояния"""
        state_features = []

        for test in test_suite:
            state_features.extend([
                test.recent_failure_rate,
                test.code_coverage_delta,
                test.execution_time_normalized,
                test.days_since_last_run
            ])

        return tuple(state_features)

    def select_action(self, state, available_tests):
        """Выбрать тесты с использованием epsilon-greedy политики"""
        # Исследование
        if random.random() < self.epsilon:
            return random.sample(
                range(len(available_tests)),
                k=min(10, len(available_tests))
            )

        # Эксплуатация
        if state not in self.q_table:
            self.q_table[state] = np.zeros(len(available_tests))

        q_values = self.q_table[state]
        return np.argsort(q_values)[-10:].tolist()  # Топ 10 тестов

    def calculate_reward(self, selected_tests, execution_results):
        """Вычислить награду на основе результатов выполнения"""
        reward = 0.0

        for test, result in zip(selected_tests, execution_results):
            if result.failed:
                # Высокая награда за обнаружение сбоев
                reward += 10.0

                # Бонус за раннее обнаружение
                if test.execution_order <= 5:
                    reward += 5.0
            else:
                # Небольшая награда за прошедшие тесты (валидация)
                reward += 0.1

            # Штраф за время выполнения
            reward -= result.execution_time / 60.0  # Нормализовать к минутам

        return reward

    def train(self, state, action, reward, next_state):
        """Обновить Q-значения с использованием Q-learning"""
        if state not in self.q_table:
            self.q_table[state] = np.zeros(self.num_tests)
        if next_state not in self.q_table:
            self.q_table[next_state] = np.zeros(self.num_tests)

        # Обновление Q-learning
        for test_idx in action:
            current_q = self.q_table[state][test_idx]
            max_next_q = np.max(self.q_table[next_state])

            new_q = current_q + self.lr * (reward + self.gamma * max_next_q - current_q)
            self.q_table[state][test_idx] = new_q

        # Затухание коэффициента исследования
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

    def remember(self, state, action, reward, next_state):
        """Сохранить опыт для воспроизведения"""
        self.memory.append((state, action, reward, next_state))

    def replay(self, batch_size=32):
        """Обучить на случайной партии из памяти"""
        if len(self.memory) < batch_size:
            return

        batch = random.sample(self.memory, batch_size)

        for state, action, reward, next_state in batch:
            self.train(state, action, reward, next_state)

from datetime import datetime, timedelta
import hashlib

class ModelRetrainingOrchestrator:
    def __init__(self, models):
        self.models = models
        self.performance_tracker = {}
        self.data_tracker = {}

    def should_retrain(self, model_name):
        """Определить, нужно ли переобучать модель"""
        triggers = {
            'time_based': self._check_time_trigger(model_name),
            'performance_degradation': self._check_performance_trigger(model_name),
            'data_drift': self._check_data_drift_trigger(model_name),
            'significant_events': self._check_event_trigger(model_name)
        }

        # Переобучить, если активен любой триггер
        return any(triggers.values()), triggers

    def _check_time_trigger(self, model_name):
        """Проверить, прошло ли достаточно времени с последнего обучения"""
        last_training = self.models[model_name].last_training_time
        time_threshold = timedelta(days=7)  # Переобучать еженедельно

        return datetime.now() - last_training > time_threshold

    def _check_performance_trigger(self, model_name):
        """Проверить на деградацию производительности"""
        if model_name not in self.performance_tracker:
            return False

        recent_accuracy = self.performance_tracker[model_name]['recent_accuracy']
        baseline_accuracy = self.performance_tracker[model_name]['baseline_accuracy']

        # Триггер, если производительность падает на 5%
        return recent_accuracy < baseline_accuracy * 0.95

    def _check_data_drift_trigger(self, model_name):
        """Обнаружить дрейф распределения данных"""
        if model_name not in self.data_tracker:
            return False

        current_distribution = self.data_tracker[model_name]['current']
        training_distribution = self.data_tracker[model_name]['baseline']

        # Вычислить дивергенцию KL или аналогичную метрику
        drift_score = self._calculate_drift(current_distribution, training_distribution)

        return drift_score > 0.1  # Порог для значительного дрейфа

    def _check_event_trigger(self, model_name):
        """Проверить значительные события, требующие переобучения"""
        events = self.models[model_name].recent_events

        significant_events = [
            'major_release',
            'architecture_change',
            'test_suite_expansion'
        ]

        return any(event['type'] in significant_events for event in events)

    def execute_retraining(self, model_name):
        """Выполнить инкрементальное или полное переобучение"""
        model = self.models[model_name]

        # Собрать данные для обучения
        training_data = self._prepare_training_data(model_name)

        # Выбрать стратегию переобучения
        if len(training_data) > 10000:
            # Инкрементальное переобучение для больших наборов данных
            self._incremental_retrain(model, training_data)
        else:
            # Полное переобучение для меньших наборов данных
            self._full_retrain(model, training_data)

        # Обновить метаданные
        model.last_training_time = datetime.now()
        model.training_data_hash = self._hash_data(training_data)

        # Валидировать новую модель
        validation_score = self._validate_model(model)

        if validation_score > self.performance_tracker[model_name]['baseline_accuracy']:
            # Развернуть новую модель
            self._deploy_model(model_name, model)
            print(f"Модель {model_name} переобучена и развернута. Новая точность: {validation_score:.3f}")
        else:
            # Откатиться к предыдущей модели
            print(f"Переобучение модели {model_name} не прошло валидацию. Сохранена предыдущая версия.")

    def _calculate_drift(self, current, baseline):
        """Вычислить дрейф распределения с использованием дивергенции KL"""
        import scipy.stats as stats
        return stats.entropy(current, baseline)

class ProductionTestingSystem:
    """Самосовершенствующаяся тестовая система корпоративного уровня"""

    def __init__(self):
        self.feedback_loop = TestFeedbackLoop(model_store, metric_collector)
        self.online_optimizer = OnlineTestOptimizer()
        self.pattern_learner = FailurePatternLearner()
        self.self_healing = SelfHealingLocator(driver)
        self.rl_selector = RLTestSelector(num_tests=500)
        self.retraining_orchestrator = ModelRetrainingOrchestrator(models)

    def run_intelligent_test_cycle(self, time_budget):
        """Выполнить оптимизированный цикл тестов с непрерывным обучением"""
        # Получить текущее состояние системы
        state = self._capture_system_state()

        # Выбрать тесты с использованием RL
        available_tests = self._get_available_tests()
        selected_indices = self.rl_selector.select_action(state, available_tests)
        selected_tests = [available_tests[i] for i in selected_indices]

        # Выполнить с самовосстановлением
        results = []
        for test in selected_tests:
            result = test.run_with_healing(self.self_healing)
            results.append(result)

            # Обновление онлайн-обучения
            self.online_optimizer.update_from_execution(
                test.extract_features(),
                result
            )

            # Анализировать сбои
            if result.failed:
                pattern_alert = self.pattern_learner.analyze_failure(result)
                if pattern_alert:
                    self._handle_pattern_alert(pattern_alert)

        # Вычислить награду и обучить RL-агента
        reward = self.rl_selector.calculate_reward(selected_tests, results)
        next_state = self._capture_system_state()
        self.rl_selector.train(state, selected_indices, reward, next_state)

        # Обработка цикла обратной связи
        self.feedback_loop.learn_from_feedback()

        # Проверить триггеры переобучения
        for model_name in self.retraining_orchestrator.models:
            should_retrain, triggers = self.retraining_orchestrator.should_retrain(model_name)
            if should_retrain:
                self.retraining_orchestrator.execute_retraining(model_name)

        return {
            'tests_executed': len(selected_tests),
            'failures_found': sum(1 for r in results if r.failed),
            'time_used': sum(r.execution_time for r in results),
            'efficiency_score': reward / time_budget
        }