Mutation Testing with AI: Intelligent Mutant Generation for Better Test Quality

# Original code
def calculate_discount(price, percentage):
    if percentage > 0:
        return price * (1 - percentage / 100)
    return price

# Mutation 1: Relational operator replacement (ROR)
def calculate_discount(price, percentage):
    if percentage >= 0:  # Changed > to >=
        return price * (1 - percentage / 100)
    return price

# Mutation 2: Arithmetic operator replacement (AOR)
def calculate_discount(price, percentage):
    if percentage > 0:
        return price * (1 + percentage / 100)  # Changed - to +
    return price

# Mutation 3: Constant replacement (CRR)
def calculate_discount(price, percentage):
    if percentage > 1:  # Changed 0 to 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):
        """Extract features for ML-based (as discussed in [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation)) mutant prioritization"""
        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):
        """Predict probability that mutant will reveal test weaknesses"""
        if not self.trained:
            raise ValueError("Model must be trained first")

        features = self.extract_features(code_snippet, mutation_operator)
        # Returns probability that mutant is valuable (not equivalent, likely to survive)
        return self.model.predict_proba(features)[0][1]

    def generate_prioritized_mutants(self, source_code, max_mutants=50):
        """Generate mutants prioritized by predicted value"""
        candidate_mutants = self._generate_all_mutants(source_code)

        # Score each mutant
        scored_mutants = []
        for mutant in candidate_mutants:
            score = self.predict_mutant_quality(mutant.code, mutant.operator)
            scored_mutants.append((score, mutant))

        # Return top N mutants by predicted quality
        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):
        """Collect training data from mutation testing execution"""
        for mutant in mutation_result.mutants:
            features = self.extract_features(mutant.code, mutant.operator)

            # Label: 1 if mutant survived and wasn't equivalent (valuable)
            #        0 if mutant was killed or equivalent (less valuable)
            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):
        """Train the mutant quality predictor"""
        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):
        # Use CodeBERT or similar model for code understanding
        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):
        """Use AI to detect equivalent mutants before execution"""

        # Encode both versions
        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

        # Calculate semantic similarity
        similarity = torch.cosine_similarity(original_embedding, mutant_embedding)

        # If similarity > threshold, likely equivalent
        return similarity.item() > 0.95

    def suggest_mutation_location(self, source_code):
        """Use attention mechanisms to identify critical mutation points"""
        tokens = self.tokenizer(source_code, return_tensors="pt")

        with torch.no_grad():
            outputs = self.model(**tokens, output_attentions=True)
            attention = outputs.attentions[-1]  # Last layer attention

        # Aggregate attention scores
        attention_scores = attention.mean(dim=1).squeeze()

        # Identify tokens with high attention (important for behavior)
        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):
        """Cluster surviving mutants to identify systematic test gaps"""

        # Extract features from surviving mutants
        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)
            ])

        # Cluster to find patterns
        clustering = DBSCAN(eps=0.5, min_samples=3).fit(features)

        # Analyze clusters
        weakness_patterns = {}
        for cluster_id in set(clustering.labels_):
            if cluster_id == -1:  # Skip noise
                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):
        """Use pattern analysis to suggest new test cases"""
        suggestions = []

        if pattern['type'] == 'boundary_condition':
            suggestions.append({
                'test_type': 'Edge case testing',
                'focus': f"Boundary values around {pattern['value']}",
                'example': self._generate_test_template(pattern)
            })

        elif pattern['type'] == 'conditional_logic':
            suggestions.append({
                'test_type': 'Branch coverage',
                'focus': f"Complement conditions in {pattern['location']}",
                'example': self._generate_branch_tests(pattern)
            })

        return suggestions

class TestEffectivenessAnalyzer:
    def calculate_mtes(self, test_suite, mutation_results):
        """
        Mutation-based Test Effectiveness Score (MTES)
        Combines multiple dimensions of test quality
        """

        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)
        }

        # Weighted combination
        weights = {
            'mutation_score': 0.35,
            'detection_speed': 0.15,
            'coverage_diversity': 0.25,
            'false_negative_rate': -0.10,  # Penalty
            '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):
        """Measure coverage of AI-identified critical mutations"""
        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

# Integration with MutPy for Python projects
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):
        """Override to add AI-based mutant prioritization"""
        # Generate mutants as usual
        all_mutants = super().generate_mutants()

        # Use AI to prioritize
        prioritized = self.ai_engine.prioritize_mutants(all_mutants)

        # Run only top N% of mutants
        results = self.execute_mutants(prioritized[:self.config.mutant_limit])

        # Use AI to detect equivalent mutants
        filtered_results = self.filter_equivalent_mutants(results)

        return self.generate_report(filtered_results)

    def filter_equivalent_mutants(self, results):
        """Use semantic analysis to identify equivalent mutants"""
        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

# Usage
ai_engine = IntelligentMutantGenerator()
controller = AIMutPyController(ai_engine)
controller.config.mutant_limit = 100
results = controller.run()

<!-- Enhanced PIT configuration with AI-guided mutations -->
<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>

        <!-- Custom mutator selection based on AI recommendations -->
        <mutators>
            <mutator>CONDITIONALS_BOUNDARY</mutator>
            <mutator>INCREMENTS</mutator>
            <mutator>MATH</mutator>
            <mutator>NEGATE_CONDITIONALS</mutator>
        </mutators>

        <!-- AI-based mutant filtering -->
        <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):
        """Execute mutations with AI-based scheduling"""

        # Group mutants by predicted execution time
        mutant_groups = self.ai_scheduler.group_by_execution_time(mutants)

        results = []
        with ProcessPoolExecutor(max_workers=self.num_workers) as executor:
            # Submit jobs in order of descending execution time
            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

            # Collect results as they complete
            for future in as_completed(futures):
                group = futures[future]
                try:
                    group_results = future.result()
                    results.extend(group_results)

                    # Update execution time estimates for learning
                    self.ai_scheduler.update_estimates(group, group_results)
                except Exception as e:
                    print(f"Group {group.id} failed: {e}")

        return results

    def _execute_mutant_group(self, group, test_suite):
        """Execute a group of similar mutants"""
        results = []

        for mutant in group.mutants:
            # Early termination if first test kills mutant
            result = self._execute_with_early_stop(mutant, test_suite)
            results.append(result)

        return results

    def _execute_with_early_stop(self, mutant, test_suite):
        """Stop execution as soon as mutant is killed"""
        # AI predicts which tests are most likely to kill this mutant
        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  # Early termination

        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):
        """Only run mutations for changed code"""

        # Identify changed lines
        changed_regions = self.git_analyzer.get_changed_regions(
            baseline_commit, current_commit
        )

        # Load cached results
        cached_results = self._load_cache(baseline_commit)

        # Generate mutants only for changed regions
        new_mutants = []
        for region in changed_regions:
            mutants = self.generate_mutants_for_region(region)
            new_mutants.extend(mutants)

        # Execute only new mutants
        new_results = self.execute_mutants(new_mutants)

        # Combine with cached results
        combined_results = self._merge_results(cached_results, new_results)

        # Update cache
        self._save_cache(current_commit, combined_results)

        return combined_results

# Original payment validation function
class PaymentProcessor:
    def validate_payment(self, amount, currency, card_number):
        if amount <= 0:
            raise ValueError("Amount must be positive")

        if currency not in ['USD', 'EUR', 'GBP']:
            raise ValueError("Unsupported currency")

        if not self._validate_card(card_number):
            raise ValueError("Invalid card number")

        return True

    def _validate_card(self, card_number):
        # Luhn algorithm
        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

# AI identified these critical surviving mutants:

# Mutant 1: Boundary condition (HIGH PRIORITY)
if amount < 0:  # Changed <= to <
    raise ValueError("Amount must be positive")
# Impact: Zero amount payments not rejected

# Mutant 2: Incomplete validation (HIGH PRIORITY)
if currency not in ['USD', 'EUR']:  # Removed 'GBP'
    raise ValueError("Unsupported currency")
# Impact: GBP handling not tested

# Mutant 3: Logic error (CRITICAL)
return checksum % 10 == 1  # Changed 0 to 1
# Impact: Card validation logic not verified