Mutation Testing с ИИ: Интеллектуальная Генерация Мутантов для Лучшего Качества Тестов

# Исходный код
def calculate_discount(price, percentage):
    if percentage > 0:
        return price * (1 - percentage / 100)
    return price

# Мутация 1: Замена реляционного оператора (ROR)
def calculate_discount(price, percentage):
    if percentage >= 0:  # Изменено > на >=
        return price * (1 - percentage / 100)
    return price

# Мутация 2: Замена арифметического оператора (AOR)
def calculate_discount(price, percentage):
    if percentage > 0:
        return price * (1 + percentage / 100)  # Изменено - на +
    return price

# Мутация 3: Замена константы (CRR)
def calculate_discount(price, percentage):
    if percentage > 1:  # Изменено 0 на 1
        return price * (1 - percentage / 100)
    return price

import ast
from sklearn.ensemble import RandomForestClassifier
import numpy as np

class IntelligentMutantGenerator:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100)
        self.trained = False

    def extract_features(self, code_snippet, mutation_operator):
        """Извлекает признаки для приоритизации мутантов на основе ML"""
 (как обсуждается в [AI Code Smell Detection: Finding Problems in Test Automation with ML](/blog/ai-code-smell-detection))        tree = ast.parse(code_snippet)

        features = {
            'cyclomatic_complexity': self._calculate_complexity(tree),
            'nesting_depth': self._max_nesting_depth(tree),
            'num_branches': len([n for n in ast.walk(tree) if isinstance(n, ast.If)]),
            'num_loops': len([n for n in ast.walk(tree) if isinstance(n, (ast.For, ast.While))]),
            'code_length': len(code_snippet.split('\n')),
            'mutation_type': self._encode_mutation_type(mutation_operator),
            'variable_count': len([n for n in ast.walk(tree) if isinstance(n, ast.Name)]),
            'test_coverage': self._get_line_coverage(code_snippet)
        }

        return np.array(list(features.values())).reshape(1, -1)

    def predict_mutant_quality(self, code_snippet, mutation_operator):
        """Предсказывает вероятность того, что мутант выявит слабости тестов"""
        if not self.trained (как обсуждается в [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation)):
            raise ValueError("Модель должна быть обучена сначала")

        features = self.extract_features(code_snippet, mutation_operator)
        # Возвращает вероятность того, что мутант ценен (не эквивалентен, вероятно выживет)
        return self.model.predict_proba(features)[0][1]

    def generate_prioritized_mutants(self, source_code, max_mutants=50):
        """Генерирует мутантов, приоритизированных по предсказанной ценности"""
        candidate_mutants = self._generate_all_mutants(source_code)

        # Оценивает каждого мутанта
        scored_mutants = []
        for mutant in candidate_mutants:
            score = self.predict_mutant_quality(mutant.code, mutant.operator)
            scored_mutants.append((score, mutant))

        # Возвращает топ N мутантов по предсказанному качеству
        scored_mutants.sort(reverse=True, key=lambda x: x[0])
        return [mutant for score, mutant in scored_mutants[:max_mutants]]

class MutantTrainingData:
    def __init__(self):
        self.training_examples = []

    def collect_from_run(self, mutation_result):
        """Собирает обучающие данные из выполнения mutation testing"""
        for mutant in mutation_result.mutants:
            features = self.extract_features(mutant.code, mutant.operator)

            # Метка: 1 если мутант выжил и не был эквивалентным (ценный)
            #        0 если мутант был убит или эквивалентен (менее ценный)
            label = 1 if (mutant.survived and not mutant.is_equivalent) else 0

            self.training_examples.append({
                'features': features,
                'label': label,
                'survival_time': mutant.detection_time,
                'test_count': len(mutant.killing_tests)
            })

    def train_model(self, generator):
        """Обучает предсказатель качества мутантов"""
        X = np.array([ex['features'] for ex in self.training_examples])
        y = np.array([ex['label'] for ex in self.training_examples])

        generator.model.fit(X, y)
        generator.trained = True

        return generator.model.score(X, y)

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

class SemanticMutationEngine:
    def __init__(self):
        # Использовать CodeBERT или похожую модель для понимания кода
        self.tokenizer = AutoTokenizer.from_pretrained("microsoft/codebert-base")
        self.model = AutoModelForSequenceClassification.from_pretrained(
            "microsoft/codebert-base"
        )

    def is_semantically_equivalent(self, original_code, mutant_code):
        """Использует ИИ для обнаружения эквивалентных мутантов перед выполнением"""

        # Кодирует обе версии
        original_tokens = self.tokenizer(original_code, return_tensors="pt",
                                        truncation=True, max_length=512)
        mutant_tokens = self.tokenizer(mutant_code, return_tensors="pt",
                                      truncation=True, max_length=512)

        with torch.no_grad():
            original_embedding = self.model(**original_tokens).logits
            mutant_embedding = self.model(**mutant_tokens).logits

        # Вычисляет семантическое сходство
        similarity = torch.cosine_similarity(original_embedding, mutant_embedding)

        # Если сходство > порога, вероятно эквивалентны
        return similarity.item() > 0.95

    def suggest_mutation_location(self, source_code):
        """Использует механизмы внимания для идентификации критических точек мутации"""
        tokens = self.tokenizer(source_code, return_tensors="pt")

        with torch.no_grad():
            outputs = self.model(**tokens, output_attentions=True)
            attention = outputs.attentions[-1]  # Внимание последнего слоя

        # Агрегирует оценки внимания
        attention_scores = attention.mean(dim=1).squeeze()

        # Идентифицирует токены с высоким вниманием (важные для поведения)
        important_indices = torch.topk(attention_scores.mean(dim=0), k=10).indices

        return important_indices.tolist()

from sklearn.cluster import DBSCAN
from sklearn.decomposition import PCA

class MutantAnalyzer:
    def analyze_test_weaknesses(self, mutation_results):
        """Кластеризует выживших мутантов для выявления систематических пробелов в тестах"""

        # Извлекает признаки из выживших мутантов
        survivors = [m for m in mutation_results.mutants if m.survived]

        features = []
        for mutant in survivors:
            features.append([
                mutant.location.line_number,
                mutant.location.column,
                self._encode_operator(mutant.operator),
                mutant.complexity_score,
                mutant.branch_id,
                self._encode_context(mutant.context)
            ])

        # Кластеризует для поиска паттернов
        clustering = DBSCAN(eps=0.5, min_samples=3).fit(features)

        # Анализирует кластеры
        weakness_patterns = {}
        for cluster_id in set(clustering.labels_):
            if cluster_id == -1:  # Пропускает шум
                continue

            cluster_mutants = [survivors[i] for i, label in enumerate(clustering.labels_)
                             if label == cluster_id]

            pattern = self._extract_pattern(cluster_mutants)
            weakness_patterns[cluster_id] = {
                'pattern': pattern,
                'severity': len(cluster_mutants),
                'suggested_tests': self._suggest_tests(pattern)
            }

        return weakness_patterns

    def _suggest_tests(self, pattern):
        """Использует анализ паттернов для предложения новых тест-кейсов"""
        suggestions = []

        if pattern['type'] == 'boundary_condition':
            suggestions.append({
                'test_type': 'Тестирование граничных случаев',
                'focus': f"Граничные значения вокруг {pattern['value']}",
                'example': self._generate_test_template(pattern)
            })

        elif pattern['type'] == 'conditional_logic':
            suggestions.append({
                'test_type': 'Покрытие ветвей',
                'focus': f"Дополнительные условия в {pattern['location']}",
                'example': self._generate_branch_tests(pattern)
            })

        return suggestions

class TestEffectivenessAnalyzer:
    def calculate_mtes(self, test_suite, mutation_results):
        """
        Оценка Эффективности Тестов на Основе Мутаций (MTES)
        Комбинирует несколько измерений качества тестов
        """

        metrics = {
            'mutation_score': self._basic_mutation_score(mutation_results),
            'detection_speed': self._average_detection_time(mutation_results),
            'coverage_diversity': self._mutation_coverage_diversity(mutation_results),
            'false_negative_rate': self._equivalent_mutant_ratio(mutation_results),
            'critical_path_coverage': self._critical_mutation_coverage(mutation_results)
        }

        # Взвешенная комбинация
        weights = {
            'mutation_score': 0.35,
            'detection_speed': 0.15,
            'coverage_diversity': 0.25,
            'false_negative_rate': -0.10,  # Штраф
            'critical_path_coverage': 0.35
        }

        mtes = sum(metrics[k] * weights[k] for k in weights.keys())

        return {
            'overall_score': mtes,
            'breakdown': metrics,
            'grade': self._assign_grade(mtes),
            'recommendations': self._generate_recommendations(metrics)
        }

    def _critical_mutation_coverage(self, results):
        """Измеряет покрытие критических мутаций, выявленных ИИ"""
        critical_mutants = [m for m in results.mutants if m.criticality_score > 0.8]
        killed_critical = [m for m in critical_mutants if not m.survived]

        return len(killed_critical) / len(critical_mutants) if critical_mutants else 0

# Интеграция с MutPy для Python-проектов
from mutpy import controller
from mutpy.commandline import build_parser

class AIMutPyController(controller.MutationController):
    def __init__(self, ai_engine):
        super().__init__()
        self.ai_engine = ai_engine

    def run(self):
        """Переопределяет для добавления приоритизации мутантов на основе ИИ"""
        # Генерирует мутантов как обычно
        all_mutants = super().generate_mutants()

        # Использует ИИ для приоритизации
        prioritized = self.ai_engine.prioritize_mutants(all_mutants)

        # Выполняет только топ N% мутантов
        results = self.execute_mutants(prioritized[:self.config.mutant_limit])

        # Использует ИИ для обнаружения эквивалентных мутантов
        filtered_results = self.filter_equivalent_mutants(results)

        return self.generate_report(filtered_results)

    def filter_equivalent_mutants(self, results):
        """Использует семантический анализ для идентификации эквивалентных мутантов"""
        filtered = []

        for result in results:
            if result.survived:
                is_equivalent = self.ai_engine.is_semantically_equivalent(
                    result.original_code,
                    result.mutant_code
                )

                result.is_equivalent = is_equivalent
                result.confidence = self.ai_engine.equivalence_confidence

            filtered.append(result)

        return filtered

# Использование
ai_engine = IntelligentMutantGenerator()
controller = AIMutPyController(ai_engine)
controller.config.mutant_limit = 100
results = controller.run()

<!-- Улучшенная конфигурация PIT с мутациями, управляемыми ИИ -->
<plugin>
    <groupId>org.pitest</groupId>
    <artifactId>pitest-maven</artifactId>
    <version>1.15.0</version>
    <configuration>
        <targetClasses>
            <param>com.example.critical.*</param>
        </targetClasses>
        <targetTests>
            <param>com.example.tests.*</param>
        </targetTests>

        <!-- Пользовательский выбор мутаторов на основе рекомендаций ИИ -->
        <mutators>
            <mutator>CONDITIONALS_BOUNDARY</mutator>
            <mutator>INCREMENTS</mutator>
            <mutator>MATH</mutator>
            <mutator>NEGATE_CONDITIONALS</mutator>
        </mutators>

        <!-- Фильтрация мутантов на основе ИИ -->
        <features>
            <feature>+AI_MUTANT_FILTER</feature>
            <feature>+SEMANTIC_EQUIVALENCE_DETECTION</feature>
        </features>

        <threads>4</threads>
        <timeoutConstant>10000</timeoutConstant>
    </configuration>
</plugin>

from concurrent.futures import ProcessPoolExecutor, as_completed
import multiprocessing as mp

class OptimizedMutationRunner:
    def __init__(self, num_workers=None):
        self.num_workers = num_workers or mp.cpu_count()
        self.ai_scheduler = IntelligentScheduler()

    def run_mutations_parallel(self, mutants, test_suite):
        """Выполняет мутации с планированием на основе ИИ"""

        # Группирует мутантов по предсказанному времени выполнения
        mutant_groups = self.ai_scheduler.group_by_execution_time(mutants)

        results = []
        with ProcessPoolExecutor(max_workers=self.num_workers) as executor:
            # Отправляет задания в порядке убывания времени выполнения
            futures = {}
            for group in sorted(mutant_groups, key=lambda g: g.estimated_time, reverse=True):
                future = executor.submit(self._execute_mutant_group, group, test_suite)
                futures[future] = group

            # Собирает результаты по мере их завершения
            for future in as_completed(futures):
                group = futures[future]
                try:
                    group_results = future.result()
                    results.extend(group_results)

                    # Обновляет оценки времени выполнения для обучения
                    self.ai_scheduler.update_estimates(group, group_results)
                except Exception as e:
                    print(f"Группа {group.id} провалилась: {e}")

        return results

    def _execute_mutant_group(self, group, test_suite):
        """Выполняет группу похожих мутантов"""
        results = []

        for mutant in group.mutants:
            # Ранняя остановка если первый тест убивает мутанта
            result = self._execute_with_early_stop(mutant, test_suite)
            results.append(result)

        return results

    def _execute_with_early_stop(self, mutant, test_suite):
        """Останавливает выполнение как только мутант убит"""
        # ИИ предсказывает какие тесты с наибольшей вероятностью убьют этого мутанта
        ordered_tests = self.ai_scheduler.order_tests_for_mutant(
            mutant, test_suite
        )

        for test in ordered_tests:
            result = self._run_single_test(mutant, test)
            if result.killed:
                return result  # Ранняя остановка

        return MutantResult(mutant, survived=True)

class IncrementalMutationTester:
    def __init__(self, cache_dir='.mutation_cache'):
        self.cache_dir = cache_dir
        self.git_analyzer = GitChangeAnalyzer()

    def run_incremental(self, baseline_commit, current_commit):
        """Выполняет мутации только для измененного кода"""

        # Идентифицирует измененные строки
        changed_regions = self.git_analyzer.get_changed_regions(
            baseline_commit, current_commit
        )

        # Загружает кэшированные результаты
        cached_results = self._load_cache(baseline_commit)

        # Генерирует мутантов только для измененных регионов
        new_mutants = []
        for region in changed_regions:
            mutants = self.generate_mutants_for_region(region)
            new_mutants.extend(mutants)

        # Выполняет только новых мутантов
        new_results = self.execute_mutants(new_mutants)

        # Объединяет с кэшированными результатами
        combined_results = self._merge_results(cached_results, new_results)

        # Обновляет кэш
        self._save_cache(current_commit, combined_results)

        return combined_results

# Исходная функция валидации платежей
class PaymentProcessor:
    def validate_payment(self, amount, currency, card_number):
        if amount <= 0:
            raise ValueError("Сумма должна быть положительной")

        if currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError("Неподдерживаемая валюта")

        if not self._validate_card(card_number):
            raise ValueError("Неверный номер карты")

        return True

    def _validate_card(self, card_number):
        # Алгоритм Луна
        digits = [int(d) for d in str(card_number)]
        checksum = sum(digits[-1::-2] + [sum(divmod(d*2, 10)) for d in digits[-2::-2]])
        return checksum % 10 == 0

# ИИ идентифицировал эти критические выжившие мутанты:

# Мутант 1: Граничное условие (ВЫСОКИЙ ПРИОРИТЕТ)
if amount < 0:  # Изменено <= на <
    raise ValueError("Сумма должна быть положительной")
# Влияние: Платежи с нулевой суммой не отклоняются

# Мутант 2: Неполная валидация (ВЫСОКИЙ ПРИОРИТЕТ)
if currency not in ['USD', 'EUR']:  # Удалено 'GBP'
    raise ValueError("Неподдерживаемая валюта")
# Влияние: Обработка GBP не протестирована

# Мутант 3: Логическая ошибка (КРИТИЧНО)
return checksum % 10 == 1  # Изменено 0 на 1
# Влияние: Логика валидации карты не проверена