Анализ Влияния Тестов с ИИ: Умный Выбор Тестов После Изменений Кода

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):
        """Анализ изменений кода с использованием парсинга AST"""
        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)

        # Обнаружить модифицированные функции
        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)

        # Обнаружить новые функции
        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):
        """Извлечь определения функций из 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)
        }

# Пример использования
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"Измененные функции: {impact['functions']}")
# Вывод: Измененные функции: ['calculate_total']

from transformers import AutoTokenizer, AutoModel
import torch

class SemanticChangeDetector:
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained('microsoft/codebert-base')
 (как обсуждается в [AI-Assisted Bug Triaging: Intelligent Defect Prioritization at Scale](/blog/ai-bug-triaging))        self.model = AutoModel.from_pretrained('microsoft/codebert-base')

 (как обсуждается в [AI Code Smell Detection: Finding Problems in Test Automation with ML](/blog/ai-code-smell-detection))    def get_embedding(self, code):
        """Генерировать семантическое представление для фрагмента кода"""
        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):
        """Вычислить семантическое сходство между версиями кода"""
        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):
        """Определить, является ли изменение семантически значимым"""
        similarity = self.calculate_similarity(old_code, new_code)
        return similarity < threshold

# Пример: Обнаружение рефакторинга vs изменений логики
detector = SemanticChangeDetector()

# Рефакторинг (высокая схожесть)
old_v1 = "def add(a, b): return a + b"
new_v1 = "def add(x, y): return x + y"
print(f"Схожесть рефакторинга: {detector.calculate_similarity(old_v1, new_v1):.3f}")

# Изменение логики (низкая схожесть)
old_v2 = "def process(data): return data.sort()"
new_v2 = "def process(data): return data.filter(lambda x: x > 0).sort()"
print(f"Схожесть изменения логики: {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]):
        """Добавить модуль и его зависимости в граф"""
        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]:
        """Найти все модули, затронутые изменениями, используя обратные зависимости"""
        affected = set(changed_modules)

        for module in changed_modules:
            # Найти все модули, которые зависят от этого измененного модуля
            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]]:
        """Сопоставить исходные файлы с тестовыми файлами, которые их покрывают"""
        coverage_map = {}

        for node in self.graph.nodes():
            if node.endswith('_test.py'):
                # Найти все исходные файлы, которые покрывает этот тест
                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

# Пример использования
builder = DependencyGraphBuilder()

# Построить граф зависимостей
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'])

# Найти затронутые модули
changed_files = {'src/database.py'}
affected = builder.find_affected_modules(changed_files)
print(f"Затронутые модули: {affected}")
# Вывод: Затронутые модули: {'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 (как обсуждается в [AI-powered Test Generation: The Future Is Already Here](/blog/ai-powered-test-generation)) = False

    def extract_features(self, 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):
        """Обучить модель на исторических результатах тестов"""
        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:
        """Предсказать вероятность сбоя теста"""
        if not self.is_trained:
            raise ValueError("Модель должна быть сначала обучена")

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

        # Вернуть вероятность сбоя (класс 1)
        return self.model.predict_proba(features_scaled)[0][1]

# Пример использования
predictor = TestRiskPredictor()

# Обучающие данные (исторические изменения и результаты тестов)
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},
    # ... больше исторических данных
])

predictor.train(training_data)

# Предсказать риск для нового изменения
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"Риск сбоя теста: {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]:
        """
        Выбрать тесты используя многокритериальное принятие решений
        """
        # Шаг 1: Найти непосредственно затронутые тесты
        affected_modules = self.dependency_graph.find_affected_modules(changed_files)
        candidate_tests = self._get_tests_for_modules(affected_modules)

        # Шаг 2: Вычислить оценки рисков
        scored_tests = []
        for test in candidate_tests:
            risk_score = self._calculate_test_priority(test, changed_files)
            scored_tests.append((test, risk_score))

        # Шаг 3: Отсортировать по риску (по убыванию)
        scored_tests.sort(key=lambda x: x[1], reverse=True)

        # Шаг 4: Применить ограничение временного бюджета
        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:
                    # Тест с высоким риском превышает бюджет - предупредить пользователя
                    print(f"Предупреждение: Тест с высоким риском {test.name} исключен из-за временного бюджета")
                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:
        """
        Вычислить оценку приоритета, комбинируя несколько факторов
        """
        # Фактор 1: Историческая частота сбоев (0-1)
        failure_weight = test.failure_rate

        # Фактор 2: Расстояние зависимости (ближе = выше приоритет)
        distance = self._calculate_dependency_distance(test, changed_files)
        distance_weight = 1.0 / (1.0 + distance)

        # Фактор 3: Предсказание риска на основе ML
        risk_weight = self._get_ml_risk_score(test, changed_files)

        # Фактор 4: Время выполнения (быстрые тесты = небольшое повышение приоритета)
        time_weight = 0.1 / (test.execution_time + 0.1)

        # Взвешенная комбинация
        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:
        """Вычислить минимальную длину пути зависимости"""
        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:
        """Получить предсказание риска на основе ML"""
        # Подготовить признаки для предсказания риска
        change_data = {
            'changed_files': len(changed_files),
            'complexity': 5,  # Будет вычислено из реального кода
            '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]:
        """Получить все тесты, покрывающие указанные модули"""
        return [t for t in self.test_cases
                if any(m in t.file_path for m in modules)]

name: Умный Выбор Тестов

on:
  pull_request:
    branches: [ main, develop ]

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

    steps:
    - uses: actions/checkout@v3
      with:
        fetch-depth: 0  # Получить полную историю для анализа изменений

    - name: Настроить Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.11'

    - name: Установить зависимости
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov

    - name: Проанализировать изменения кода
      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: Выбрать тесты с ИИ
      id: select
      run: |
        python scripts/select_tests.py \
          --changes changes.json \
          --time-budget 600 \
          --output selected_tests.txt

    - name: Запустить выбранные тесты
      run: |
        pytest $(cat selected_tests.txt) \
          --cov=src \
          --cov-report=xml \
          --junit-xml=test-results.xml

    - name: Загрузить покрытие
      uses: codecov/codecov-action@v3
      with:
        files: ./coverage.xml

    - name: Прокомментировать PR с результатами
      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: `## Результаты Умного Выбора Тестов\n\n` +
                  `Выбрано ${testCount} тестов на основе анализа ИИ.\n\n` +
                  `Сэкономлено времени: ~${Math.round((1000 - testCount) / 1000 * 100)}%`
          });

pipeline {
    agent any

    stages {
        stage('Анализ Изменений') {
            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('Выбор Тестов') {
            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('Выполнение Тестов') {
            steps {
                sh "pytest ${env.SELECTED_TESTS} --junit-xml=results.xml"
            }
        }

        stage('Запасной Вариант - Запуск Всех Тестов') {
            when {
                expression { currentBuild.result == 'FAILURE' }
            }
            steps {
                echo "Выбранные тесты провалились. Запуск полного набора..."
                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  # Выбранные тесты, которые действительно провалились
    false_negatives: int  # Пропущенные тесты, которые провалились бы
    false_positives: int  # Выбранные тесты, которые прошли

    @property
    def selection_rate(self) -> float:
        """Процент выбранных тестов"""
        return (self.selected_tests / self.total_tests) * 100

    @property
    def time_savings(self) -> float:
        """Процент сэкономленного времени"""
        return ((self.execution_time_full - self.execution_time_selected) /
                self.execution_time_full) * 100

    @property
    def precision(self) -> float:
        """Точность: ИП / (ИП + ЛП)"""
        return self.true_positives / (self.true_positives + self.false_positives)

    @property
    def recall(self) -> float:
        """Полнота: ИП / (ИП + ЛО)"""
        return self.true_positives / (self.true_positives + self.false_negatives)

    @property
    def f1_score(self) -> float:
        """F1 Оценка: Гармоническое среднее точности и полноты"""
        p = self.precision
        r = self.recall
        return 2 * (p * r) / (p + r)

    def print_report(self):
        print("="*50)
        print("Анализ Влияния Тестов - Отчет о Производительности")
        print("="*50)
        print(f"Всего тестов: {self.total_tests}")
        print(f"Выбрано тестов: {self.selected_tests} ({self.selection_rate:.1f}%)")
        print(f"Сэкономлено времени: {self.time_savings:.1f}%")
        print(f"Точность: {self.precision:.2%}")
        print(f"Полнота: {self.recall:.2%}")
        print(f"F1 Оценка: {self.f1_score:.3f}")
        print("="*50)

# Пример метрик из промышленного развертывания
metrics = TIAMetrics(
    total_tests=5000,
    selected_tests=850,
    execution_time_full=7200,  # 2 часа
    execution_time_selected=1080,  # 18 минут
    true_positives=45,  # Тесты, которые провалились и были выбраны
    false_negatives=3,  # Тесты, которые провалились, но не были выбраны
    false_positives=802  # Тесты, которые прошли, но были выбраны
)

metrics.print_report()