Test Impact Analysis with AI: Smart Test Selection After Code Changes

import ast
import difflib

class CodeChangeAnalyzer:
    def __init__(self):
        self.changed_functions = set()
        self.changed_classes = set()
        self.changed_imports = set()

    def analyze_changes(self, old_code, new_code):
        """Analyze code changes using AST parsing"""
        old_tree = ast.parse(old_code)
        new_tree = ast.parse(new_code)

        old_functions = self._extract_functions(old_tree)
        new_functions = self._extract_functions(new_tree)

        # Detect modified functions
        for func_name in old_functions.keys():
            if func_name in new_functions:
                if old_functions[func_name] != new_functions[func_name]:
                    self.changed_functions.add(func_name)

        # Detect new functions
        for func_name in new_functions.keys():
            if func_name not in old_functions:
                self.changed_functions.add(func_name)

        return self.get_impact_summary()

    def _extract_functions(self, tree):
        """Extract function definitions from AST"""
        functions = {}
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                functions[node.name] = ast.unparse(node)
        return functions

    def get_impact_summary(self):
        return {
            'functions': list(self.changed_functions),
            'classes': list(self.changed_classes),
            'imports': list(self.changed_imports)
        }

# Usage example
analyzer = CodeChangeAnalyzer()
old_code = """
def calculate_total(items):
    return sum(item.price for item in items)
"""

new_code = """
def calculate_total(items, discount=0):
    subtotal = sum(item.price for item in items)
    return subtotal * (1 - discount)
"""

impact = analyzer.analyze_changes(old_code, new_code)
print(f"Changed functions: {impact['functions']}")
# Output: Changed functions: ['calculate_total']

from transformers import AutoTokenizer, AutoModel
import torch

class SemanticChangeDetector:
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained('microsoft/codebert-base')
 (as discussed in [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation))        self.model = AutoModel.from_pretrained('microsoft/codebert-base')

    def get_embedding(self, code):
        """Generate semantic embedding for code snippet"""
        inputs = self.tokenizer(code, return_tensors='pt',
                               truncation=True, max_length=512)
        with torch.no_grad():
            outputs = self.model(**inputs)
        return outputs.last_hidden_state.mean(dim=1)

    def calculate_similarity(self, old_code, new_code):
        """Calculate semantic similarity between code versions"""
        old_embedding = self.get_embedding(old_code)
        new_embedding = self.get_embedding(new_code)

        similarity = torch.cosine_similarity(old_embedding, new_embedding)
        return similarity.item()

    def is_significant_change(self, old_code, new_code, threshold=0.85):
        """Determine if change is semantically significant"""
        similarity = self.calculate_similarity(old_code, new_code)
        return similarity < threshold

# Example: Detect refactoring vs logic changes
detector = SemanticChangeDetector()

# Refactoring (high similarity)
old_v1 = "def add(a, b): return a + b"
new_v1 = "def add(x, y): return x + y"
print(f"Refactoring similarity: {detector.calculate_similarity(old_v1, new_v1):.3f}")

# Logic change (low similarity)
old_v2 = "def process(data): return data.sort()"
new_v2 = "def process(data): return data.filter(lambda x: x > 0).sort()"
print(f"Logic change similarity: {detector.calculate_similarity(old_v2, new_v2):.3f}")

import networkx as nx
from typing import Set, Dict, List

class DependencyGraphBuilder:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.file_dependencies = {}

    def add_module(self, module_name: str, dependencies: List[str]):
        """Add module and its dependencies to graph"""
        self.graph.add_node(module_name)
        for dep in dependencies:
            self.graph.add_edge(module_name, dep)

    def find_affected_modules(self, changed_modules: Set[str]) -> Set[str]:
        """Find all modules affected by changes using reverse dependencies"""
        affected = set(changed_modules)

        for module in changed_modules:
            # Find all modules that depend on this changed module
            if module in self.graph:
                ancestors = nx.ancestors(self.graph, module)
                affected.update(ancestors)

        return affected

    def get_test_coverage_map(self) -> Dict[str, Set[str]]:
        """Map source files to test files that cover them"""
        coverage_map = {}

        for node in self.graph.nodes():
            if node.endswith('_test.py'):
                # Find all source files this test covers
                descendants = nx.descendants(self.graph, node)
                for source_file in descendants:
                    if not source_file.endswith('_test.py'):
                        if source_file not in coverage_map:
                            coverage_map[source_file] = set()
                        coverage_map[source_file].add(node)

        return coverage_map

# Example usage
builder = DependencyGraphBuilder()

# Build dependency graph
builder.add_module('src/auth.py', ['src/database.py', 'src/utils.py'])
builder.add_module('src/api.py', ['src/auth.py', 'src/models.py'])
builder.add_module('tests/test_auth.py', ['src/auth.py'])
builder.add_module('tests/test_api.py', ['src/api.py'])

# Find affected modules
changed_files = {'src/database.py'}
affected = builder.find_affected_modules(changed_files)
print(f"Affected modules: {affected}")
# Output: Affected modules: {'src/database.py', 'src/auth.py', 'src/api.py'}

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
import numpy as np

class TestRiskPredictor:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.scaler = StandardScaler()
        self.is_trained = False

    def extract_features(self, change_data):
        """Extract features from code change data"""
        features = {
            'lines_added': change_data.get('additions', 0),
            'lines_deleted': change_data.get('deletions', 0),
            'files_changed': change_data.get('changed_files', 1),
            'cyclomatic_complexity': change_data.get('complexity', 1),
            'author_experience': change_data.get('author_commits', 0),
            'time_since_last_change': change_data.get('hours_since_change', 0),
            'num_dependencies': change_data.get('dependency_count', 0),
            'historical_failure_rate': change_data.get('past_failures', 0.0)
        }
        return list(features.values())

    def train(self, historical_data: pd.DataFrame):
        """Train model on historical test outcomes"""
        X = np.array([self.extract_features(row)
                     for _, row in historical_data.iterrows()])
        y = historical_data['test_failed'].values

        X_scaled = self.scaler.fit_transform(X)
        self.model.fit(X_scaled, y)
        self.is_trained = True

    def predict_risk(self, change_data) -> float:
        """Predict probability of test failure"""
        if not self.is_trained:
            raise ValueError("Model must be trained first")

        features = np.array([self.extract_features(change_data)])
        features_scaled = self.scaler.transform(features)

        # Return probability of failure (class 1)
        return self.model.predict_proba(features_scaled)[0][1]

# Example usage
predictor = TestRiskPredictor()

# Training data (historical changes and test outcomes)
training_data = pd.DataFrame([
    {'additions': 10, 'deletions': 5, 'changed_files': 2, 'complexity': 3,
     'author_commits': 50, 'hours_since_change': 2, 'dependency_count': 4,
     'past_failures': 0.1, 'test_failed': 0},
    {'additions': 150, 'deletions': 80, 'changed_files': 8, 'complexity': 12,
     'author_commits': 5, 'hours_since_change': 48, 'dependency_count': 15,
     'past_failures': 0.3, 'test_failed': 1},
    # ... more historical data
])

predictor.train(training_data)

# Predict risk for new change
new_change = {
    'additions': 75, 'deletions': 30, 'changed_files': 4,
    'complexity': 8, 'author_commits': 20, 'hours_since_change': 12,
    'dependency_count': 8, 'past_failures': 0.15
}

risk_score = predictor.predict_risk(new_change)
print(f"Test failure risk: {risk_score:.2%}")

from typing import List, Set, Tuple
from dataclasses import dataclass

@dataclass
class TestCase:
    name: str
    file_path: str
    execution_time: float
    last_failure_date: str = None
    failure_rate: float = 0.0

class IntelligentTestSelector:
    def __init__(self, dependency_graph, risk_predictor):
        self.dependency_graph = dependency_graph
        self.risk_predictor = risk_predictor
        self.test_cases = []

    def select_tests(self, changed_files: Set[str],
                    time_budget: float = None,
                    min_confidence: float = 0.7) -> List[TestCase]:
        """
        Select tests using multi-criteria decision making
        """
        # Step 1: Find directly affected tests
        affected_modules = self.dependency_graph.find_affected_modules(changed_files)
        candidate_tests = self._get_tests_for_modules(affected_modules)

        # Step 2: Calculate risk scores
        scored_tests = []
        for test in candidate_tests:
            risk_score = self._calculate_test_priority(test, changed_files)
            scored_tests.append((test, risk_score))

        # Step 3: Sort by risk (descending)
        scored_tests.sort(key=lambda x: x[1], reverse=True)

        # Step 4: Apply time budget constraint
        selected_tests = []
        total_time = 0.0

        for test, score in scored_tests:
            if time_budget and total_time + test.execution_time > time_budget:
                if score >= min_confidence:
                    # High-risk test exceeds budget - warn user
                    print(f"Warning: High-risk test {test.name} excluded due to time budget")
                continue

            selected_tests.append(test)
            total_time += test.execution_time

        return selected_tests

    def _calculate_test_priority(self, test: TestCase,
                                 changed_files: Set[str]) -> float:
        """
        Calculate priority score combining multiple factors
        """
        # Factor 1: Historical failure rate (0-1)
        failure_weight = test.failure_rate

        # Factor 2: Dependency distance (closer = higher priority)
        distance = self._calculate_dependency_distance(test, changed_files)
        distance_weight = 1.0 / (1.0 + distance)

        # Factor 3: ML-based risk prediction
        risk_weight = self._get_ml_risk_score(test, changed_files)

        # Factor 4: Execution time (faster tests = slight priority boost)
        time_weight = 0.1 / (test.execution_time + 0.1)

        # Weighted combination
        priority = (
            0.35 * failure_weight +
            0.30 * distance_weight +
            0.30 * risk_weight +
            0.05 * time_weight
        )

        return priority

    def _calculate_dependency_distance(self, test: TestCase,
                                       changed_files: Set[str]) -> int:
        """Calculate minimum dependency path length"""
        min_distance = float('inf')
        for changed_file in changed_files:
            try:
                distance = nx.shortest_path_length(
                    self.dependency_graph.graph,
                    source=test.file_path,
                    target=changed_file
                )
                min_distance = min(min_distance, distance)
            except nx.NetworkXNoPath:
                continue

        return min_distance if min_distance != float('inf') else 10

    def _get_ml_risk_score(self, test: TestCase,
                          changed_files: Set[str]) -> float:
        """Get ML-based risk prediction"""
        # Prepare features for risk prediction
        change_data = {
            'changed_files': len(changed_files),
            'complexity': 5,  # Would be calculated from actual code
            'dependency_count': len(self.dependency_graph.graph.neighbors(test.file_path))
        }

        return self.risk_predictor.predict_risk(change_data)

    def _get_tests_for_modules(self, modules: Set[str]) -> List[TestCase]:
        """Get all tests covering specified modules"""
        return [t for t in self.test_cases
                if any(m in t.file_path for m in modules)]

name: Smart Test Selection

on:
  pull_request:
    branches: [ main, develop ]

jobs:
  smart-test-selection:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3
      with:
        fetch-depth: 0  # Fetch full history for change analysis

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'

    - name: Install dependencies
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov

    - name: Analyze code changes
      id: analyze
      run: |
        python scripts/analyze_changes.py \
          --base-ref ${{ github.event.pull_request.base.sha }} \
          --head-ref ${{ github.event.pull_request.head.sha }} \
          --output changes.json

    - name: Select tests with AI
      id: select
      run: |
        python scripts/select_tests.py \
          --changes changes.json \
          --time-budget 600 \
          --output selected_tests.txt

    - name: Run selected tests
      run: |
        pytest $(cat selected_tests.txt) \
          --cov=src \
          --cov-report=xml \
          --junit-xml=test-results.xml

    - name: Upload coverage
      uses: codecov/codecov-action@v3
      with:
        files: ./coverage.xml

    - name: Comment PR with results
      uses: actions/github-script@v6
      with:
        script: |
          const fs = require('fs');
          const selectedTests = fs.readFileSync('selected_tests.txt', 'utf8');
          const testCount = selectedTests.split('\n').filter(Boolean).length;

          github.rest.issues.createComment({
            issue_number: context.issue.number,
            owner: context.repo.owner,
            repo: context.repo.repo,
            body: `## Smart Test Selection Results\n\n` +
                  `Selected ${testCount} tests based on AI analysis.\n\n` +
                  `Time saved: ~${Math.round((1000 - testCount) / 1000 * 100)}%`
          });

pipeline {
    agent any

    stages {
        stage('Analyze Changes') {
            steps {
                script {
                    def changes = sh(
                        script: 'python scripts/analyze_changes.py --base-ref origin/main --head-ref HEAD',
                        returnStdout: true
                    ).trim()

                    env.CHANGED_FILES = changes
                }
            }
        }

        stage('Select Tests') {
            steps {
                script {
                    def selectedTests = sh(
                        script: """
                            python scripts/select_tests.py \
                                --changes '${env.CHANGED_FILES}' \
                                --confidence-threshold 0.8
                        """,
                        returnStdout: true
                    ).trim()

                    env.SELECTED_TESTS = selectedTests
                }
            }
        }

        stage('Execute Tests') {
            steps {
                sh "pytest ${env.SELECTED_TESTS} --junit-xml=results.xml"
            }
        }

        stage('Fallback - Run All Tests') {
            when {
                expression { currentBuild.result == 'FAILURE' }
            }
            steps {
                echo "Selected tests failed. Running full suite..."
                sh "pytest tests/ --junit-xml=full-results.xml"
            }
        }
    }

    post {
        always {
            junit 'results.xml'
        }
    }
}

from dataclasses import dataclass
from typing import List
import time

@dataclass
class TIAMetrics:
    total_tests: int
    selected_tests: int
    execution_time_full: float
    execution_time_selected: float
    true_positives: int  # Selected tests that actually failed
    false_negatives: int  # Missed tests that would have failed
    false_positives: int  # Selected tests that passed

    @property
    def selection_rate(self) -> float:
        """Percentage of tests selected"""
        return (self.selected_tests / self.total_tests) * 100

    @property
    def time_savings(self) -> float:
        """Percentage of time saved"""
        return ((self.execution_time_full - self.execution_time_selected) /
                self.execution_time_full) * 100

    @property
    def precision(self) -> float:
        """Precision: TP / (TP + FP)"""
        return self.true_positives / (self.true_positives + self.false_positives)

    @property
    def recall(self) -> float:
        """Recall: TP / (TP + FN)"""
        return self.true_positives / (self.true_positives + self.false_negatives)

    @property
    def f1_score(self) -> float:
        """F1 Score: Harmonic mean of precision and recall"""
        p = self.precision
        r = self.recall
        return 2 * (p * r) / (p + r)

    def print_report(self):
        print("="*50)
        print("Test Impact Analysis - Performance Report")
        print("="*50)
        print(f"Total tests: {self.total_tests}")
        print(f"Selected tests: {self.selected_tests} ({self.selection_rate:.1f}%)")
        print(f"Time saved: {self.time_savings:.1f}%")
        print(f"Precision: {self.precision:.2%}")
        print(f"Recall: {self.recall:.2%}")
        print(f"F1 Score: {self.f1_score:.3f}")
        print("="*50)

# Example metrics from production deployment
metrics = TIAMetrics(
    total_tests=5000,
    selected_tests=850,
    execution_time_full=7200,  # 2 hours
    execution_time_selected=1080,  # 18 minutes
    true_positives=45,  # Tests that failed and were selected
    false_negatives=3,  # Tests that failed but were not selected
    false_positives=802  # Tests that passed but were selected
)

metrics.print_report()