breakshaft/tests/test_benchmarks.py

"""
Бенчмарки производительности для breakshaft.

Измеряет время построения графа преобразований для разного числа инжекторов.

Использование:
    uv run pytest tests/test_benchmarks.py -v -s
"""

import time
from dataclasses import dataclass

import pytest

from breakshaft import ConvRepo
from breakshaft.graph_walker import GraphWalker


@dataclass
class TypeN:
    """Базовый тип для бенчмарков."""
    n: int


# =============================================================================
# Бенчмарки: Базовая производительность
# =============================================================================

class TestBenchmarkBasic:
    """Базовые бенчмарки производительности."""

    @pytest.mark.benchmark
    def test_benchmark_chain_10(self):
        """Бенчмарк: цепочка 10 инжекторов."""
        repo = ConvRepo()
        
        # Создаём цепочку Type0 -> Type1 -> ... -> Type10
        for i in range(10):
            def make_injector(idx):
                def injector(value: TypeN) -> TypeN:
                    return TypeN(value.n + 1)
                injector.__name__ = f'type_{idx}_to_type_{idx+1}'
                injector.__qualname__ = injector.__name__
                return injector
            
            repo.add_injector(make_injector(i), priority=i)
        
        start = time.perf_counter()
        
        def consumer(value: TypeN) -> int:
            return value.n
        
        fn = repo.get_conversion((TypeN,), consumer, force_commutative=False)
        
        elapsed = time.perf_counter() - start
        print(f"\nChain 10: {elapsed*1000:.2f}ms")
        assert elapsed < 1.0

    @pytest.mark.benchmark
    def test_benchmark_chain_20(self):
        """Бенчмарк: цепочка 20 инжекторов."""
        repo = ConvRepo()
        
        for i in range(20):
            def make_injector(idx):
                def injector(value: TypeN) -> TypeN:
                    return TypeN(value.n + 1)
                injector.__name__ = f'type_{idx}_to_type_{idx+1}'
                return injector
            
            repo.add_injector(make_injector(i))
        
        start = time.perf_counter()
        
        def consumer(value: TypeN) -> int:
            return value.n
        
        fn = repo.get_conversion((TypeN,), consumer, force_commutative=False)
        
        elapsed = time.perf_counter() - start
        print(f"\nChain 20: {elapsed*1000:.2f}ms")
        assert elapsed < 5.0

    @pytest.mark.benchmark
    def test_benchmark_chain_50(self):
        """Бенчмарк: цепочка 50 инжекторов."""
        repo = ConvRepo()
        
        for i in range(50):
            def make_injector(idx):
                def injector(value: TypeN) -> TypeN:
                    return TypeN(value.n + 1)
                injector.__name__ = f'type_{idx}_to_type_{idx+1}'
                return injector
            
            repo.add_injector(make_injector(i))
        
        start = time.perf_counter()
        
        def consumer(value: TypeN) -> int:
            return value.n
        
        fn = repo.get_conversion((TypeN,), consumer, force_commutative=False)
        
        elapsed = time.perf_counter() - start
        print(f"\nChain 50: {elapsed*1000:.2f}ms")

    @pytest.mark.benchmark
    def test_benchmark_fan_10(self):
        """Бенчмарк: веер 10 инжекторов (int -> TypeN)."""
        repo = ConvRepo()
        
        for i in range(10):
            def make_injector(idx):
                def injector(value: int) -> TypeN:
                    return TypeN(idx)
                injector.__name__ = f'int_to_type_{idx}'
                return injector
            
            repo.add_injector(make_injector(i))
        
        start = time.perf_counter()
        
        def consumer(value: TypeN) -> int:
            return value.n
        
        fn = repo.get_conversion((int,), consumer, force_commutative=False)
        
        elapsed = time.perf_counter() - start
        print(f"\nFan 10: {elapsed*1000:.2f}ms")

    @pytest.mark.benchmark
    def test_benchmark_fan_20(self):
        """Бенчмарк: веер 20 инжекторов."""
        repo = ConvRepo()
        
        for i in range(20):
            def make_injector(idx):
                def injector(value: int) -> TypeN:
                    return TypeN(idx)
                injector.__name__ = f'int_to_type_{idx}'
                return injector
            
            repo.add_injector(make_injector(i))
        
        start = time.perf_counter()
        
        def consumer(value: TypeN) -> int:
            return value.n
        
        fn = repo.get_conversion((int,), consumer, force_commutative=False)
        
        elapsed = time.perf_counter() - start
        print(f"\nFan 20: {elapsed*1000:.2f}ms")


# =============================================================================
# Бенчмарки: explode_callgraph_branches
# =============================================================================

class TestBenchmarkExplode:
    """Бенчмарки для explode_callgraph_branches."""

    @pytest.mark.benchmark
    def test_benchmark_explode_simple(self):
        """Бенчмарк: explode на простом графе."""
        repo = ConvRepo()
        
        @repo.mark_injector()
        def int_to_a(i: int) -> TypeN:
            return TypeN(i)
        
        @repo.mark_injector()
        def a_to_b(a: TypeN) -> TypeN:
            return TypeN(a.n + 1)
        
        walker = GraphWalker()
        
        def consumer(dep: TypeN) -> int:
            return dep.n
        
        start = time.perf_counter()
        
        cg = walker.generate_callgraph(repo.convertor_set, frozenset({int}), consumer)
        exploded = walker.explode_callgraph_branches(cg, frozenset({int}))
        
        elapsed = time.perf_counter() - start
        print(f"\nexplode (simple): {elapsed*1000:.2f}ms, variants: {len(exploded)}")
        assert len(exploded) > 0

    @pytest.mark.benchmark
    def test_benchmark_explode_fan(self):
        """Бенчмарк: explode на веерном графе."""
        repo = ConvRepo()
        
        for i in range(10):
            def make_injector(idx):
                def injector(value: int) -> TypeN:
                    return TypeN(idx)
                injector.__name__ = f'int_to_type_{idx}'
                return injector
            
            repo.add_injector(make_injector(i))
        
        walker = GraphWalker()
        
        def consumer(dep: TypeN) -> int:
            return dep.n
        
        start = time.perf_counter()
        
        cg = walker.generate_callgraph(repo.convertor_set, frozenset({int}), consumer)
        if cg:
            exploded = walker.explode_callgraph_branches(cg, frozenset({int}))
            elapsed = time.perf_counter() - start
            print(f"\nexplode (fan 10): {elapsed*1000:.2f}ms, variants: {len(exploded)}")
        else:
            elapsed = time.perf_counter() - start
            print(f"\nexplode (fan 10): no graph in {elapsed*1000:.2f}ms")


# =============================================================================
# Бенчмарки: Сравнение сценариев
# =============================================================================

class TestBenchmarkScenarios:
    """Бенчмарки различных сценариев использования."""

    @pytest.mark.benchmark
    def test_benchmark_repeated_calls(self):
        """Бенчмарк: Повторные вызовы get_conversion."""
        repo = ConvRepo()
        
        @repo.mark_injector()
        def int_to_a(i: int) -> TypeN:
            return TypeN(i)
        
        @repo.mark_injector()
        def a_to_b(a: TypeN) -> TypeN:
            return TypeN(a.n + 1)
        
        def consumer(dep: TypeN) -> int:
            return dep.n
        
        # Первый вызов
        start = time.perf_counter()
        fn1 = repo.get_conversion((int,), consumer, force_commutative=False)
        elapsed1 = time.perf_counter() - start
        
        # Второй вызов
        start = time.perf_counter()
        fn2 = repo.get_conversion((int,), consumer, force_commutative=False)
        elapsed2 = time.perf_counter() - start
        
        print(f"\nRepeated calls: {elapsed1*1000:.2f}ms -> {elapsed2*1000:.2f}ms")
        if elapsed2 > 0:
            print(f"Speedup: {elapsed1/elapsed2:.2f}x")

    @pytest.mark.benchmark
    def test_benchmark_pipeline(self):
        """Бенчмарк: create_pipeline."""
        repo = ConvRepo()
        
        @repo.mark_injector()
        def int_to_a(i: int) -> TypeN:
            return TypeN(i)
        
        @repo.mark_injector()
        def a_to_b(a: TypeN) -> TypeN:
            return TypeN(a.n + 1)
        
        def consumer1(dep: TypeN) -> TypeN:
            return dep
        
        def consumer2(dep: TypeN) -> int:
            return dep.n
        
        start = time.perf_counter()
        
        pipeline = repo.create_pipeline(
            (int,),
            [consumer1, consumer2],
            force_commutative=False
        )
        
        elapsed = time.perf_counter() - start
        print(f"\nPipeline: {elapsed*1000:.2f}ms")


# =============================================================================
# Утилиты
# =============================================================================

def run_benchmark_suite():
    """Запустить полный набор бенчмарков."""
    import subprocess
    import sys
    
    result = subprocess.run([
        sys.executable, '-m', 'pytest',
        'tests/test_benchmarks.py',
        '-v', '--tb=short',
        '-s'
    ])
    
    return result.returncode


if __name__ == '__main__':
    exit(run_benchmark_suite())