From 75fa652ccb77dbe3978aae92316612c0ba9ffb56 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Mon, 23 Mar 2026 02:04:06 +0800 Subject: [PATCH] Add performance benchmark tests with scoring - Comprehensive performance benchmarks for CommandFilter operations - Memory usage benchmarks with tracemalloc tracking - High-throughput benchmarks with ops/sec metrics - Scoring system (0-100) for performance tracking - Overall performance score summary Benchmarks include: - CommandFilter.get_complete_command_names - Boolean and integer parameter validation - Memory footprint of filter creation - High-throughput validation throughput --- tests/benchmarks/__init__.py | 0 tests/benchmarks/test_performance.py | 338 +++++++++++++++++++++++++++ 2 files changed, 338 insertions(+) create mode 100644 tests/benchmarks/__init__.py create mode 100644 tests/benchmarks/test_performance.py diff --git a/tests/benchmarks/__init__.py b/tests/benchmarks/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/benchmarks/test_performance.py b/tests/benchmarks/test_performance.py new file mode 100644 index 000000000..53900169a --- /dev/null +++ b/tests/benchmarks/test_performance.py @@ -0,0 +1,338 @@ +"""Comprehensive performance benchmark tests for AstrBot core modules. + +This module provides performance benchmarks with scoring to track +performance regressions and improvements over time. +""" + +import asyncio +import gc +import tracemalloc +from typing import Callable, Any +from dataclasses import dataclass + +import pytest + +from astrbot.core.star.filter.command import CommandFilter, GreedyStr + + +@dataclass +class BenchmarkResult: + """Result of a benchmark test.""" + name: str + operation_count: int + total_time_ms: float + avg_time_ms: float + ops_per_second: float + memory_delta_kb: float + score: int # 0-100, 100 is best + + def __str__(self) -> str: + return ( + f"[{self.name}]\n" + f" Ops/sec: {self.ops_per_second:,.0f} | " + f"Avg: {self.avg_time_ms:.4f}ms | " + f"Memory: +{self.memory_delta_kb:.1f}KB | " + f"Score: {self.score}/100" + ) + + +class PerformanceBenchmark: + """Helper class to run benchmarks with memory tracking.""" + + def __init__(self, name: str, operations: int = 1000): + self.name = name + self.operations = operations + self.tracemalloc = tracemalloc + + def run(self, func: Callable, *args, **kwargs) -> BenchmarkResult: + """Run a function multiple times and measure performance.""" + gc.collect() + self.tracemalloc.start() + snapshot_before = self.tracemalloc.take_snapshot() + + start = asyncio.get_event_loop().time() + for _ in range(self.operations): + func(*args, **kwargs) + end = asyncio.get_event_loop().time() + + snapshot_after = self.tracemalloc.take_snapshot() + self.tracemalloc.stop() + + total_time = (end - start) * 1000 # ms + avg_time = total_time / self.operations + ops_per_sec = self.operations / ((end - start) if (end - start) > 0 else 0.001) + + # Calculate memory delta + top_stats = snapshot_after.compare_to(snapshot_before, 'lineno') + memory_delta_kb = sum(stat.size_diff for stat in top_stats) / 1024 + + # Calculate score (0-100) + # Higher ops/sec = better, lower memory = better + score = self._calculate_score(ops_per_sec, memory_delta_kb) + + return BenchmarkResult( + name=self.name, + operation_count=self.operations, + total_time_ms=total_time, + avg_time_ms=avg_time, + ops_per_second=ops_per_sec, + memory_delta_kb=memory_delta_kb, + score=score, + ) + + def _calculate_score(self, ops_per_sec: float, memory_kb: float) -> int: + """Calculate a score from 0-100 based on performance metrics.""" + # Score based on operations per second (log scale) + # 10k ops/sec = 80 points, 100k = 95 points, 1M = 100 points + if ops_per_sec >= 1_000_000: + ops_score = 100 + elif ops_per_sec >= 100_000: + ops_score = 95 + elif ops_per_sec >= 10_000: + ops_score = 80 + elif ops_per_sec >= 1_000: + ops_score = 60 + elif ops_per_sec >= 100: + ops_score = 40 + else: + ops_score = 20 + + # Memory penalty (lower is better) + # < 1KB per op = no penalty, > 100KB = max penalty + memory_per_op = memory_kb / self.operations + if memory_per_op < 0.001: + mem_score = 0 + elif memory_per_op < 0.1: + mem_score = 5 + elif memory_per_op < 1: + mem_score = 10 + else: + mem_score = min(15, int(memory_per_op / 10)) + + return max(0, min(100, ops_score - mem_score)) + + +class TestCommandFilterBenchmarks: + """Performance benchmarks for CommandFilter operations.""" + + def test_complete_command_names_performance(self): + """Benchmark get_complete_command_names with caching.""" + bench = PerformanceBenchmark("CommandFilter.get_complete_command_names", operations=10000) + + # Setup: create 100 filters + filters: list[CommandFilter] = [] + for i in range(100): + cf = CommandFilter(command_name=f"test_cmd_{i}") + cf.alias = {f"t{i}", f"alias{i}"} + cf.parent_command_names = [f"parent{i}"] + filters.append(cf) + + result = bench.run(lambda: [cf.get_complete_command_names() for cf in filters]) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Operations: {result.operation_count:,}") + print(f" Total time: {result.total_time_ms:.2f}ms") + print(f" Avg per call: {result.avg_time_ms:.6f}ms") + print(f" Ops/sec: {result.ops_per_second:,.0f}") + print(f" Memory delta: +{result.memory_delta_kb:.2f}KB") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + assert result.score >= 60, f"Performance score {result.score} is below threshold 60" + + def test_validate_bool_params_performance(self): + """Benchmark boolean parameter validation.""" + bench = PerformanceBenchmark("CommandFilter.validate_bool", operations=50000) + + cf = CommandFilter(command_name="test") + cf.handler_params = {"enabled": bool} + + result = bench.run( + lambda: cf.validate_and_convert_params(["true"], cf.handler_params) + ) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Operations: {result.operation_count:,}") + print(f" Total time: {result.total_time_ms:.2f}ms") + print(f" Avg per call: {result.avg_time_ms:.6f}ms") + print(f" Ops/sec: {result.ops_per_second:,.0f}") + print(f" Memory delta: +{result.memory_delta_kb:.2f}KB") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + assert result.score >= 70, f"Performance score {result.score} is below threshold 70" + + def test_validate_int_params_performance(self): + """Benchmark integer parameter validation.""" + bench = PerformanceBenchmark("CommandFilter.validate_int", operations=50000) + + cf = CommandFilter(command_name="test") + cf.handler_params = {"count": int} + + result = bench.run( + lambda: cf.validate_and_convert_params(["42"], cf.handler_params) + ) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Operations: {result.operation_count:,}") + print(f" Total time: {result.total_time_ms:.2f}ms") + print(f" Avg per call: {result.avg_time_ms:.6f}ms") + print(f" Ops/sec: {result.ops_per_second:,.0f}") + print(f" Memory delta: +{result.memory_delta_kb:.2f}KB") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + assert result.score >= 70, f"Performance score {result.score} is below threshold 70" + + +class TestMemoryBenchmarks: + """Memory usage benchmarks.""" + + def test_filter_creation_memory(self): + """Benchmark memory usage when creating many filters.""" + bench = PerformanceBenchmark("Filter creation (1000x)", operations=10) + + def create_filters(): + filters = [] + for i in range(1000): + cf = CommandFilter(command_name=f"cmd_{i}") + cf.alias = {f"a{i}", f"b{i}"} + filters.append(cf) + return filters + + result = bench.run(create_filters) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Creating 1000 filters") + print(f" Total memory: +{result.memory_delta_kb:.2f}KB") + print(f" Per filter: {result.memory_delta_kb:.4f}KB") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + # Each filter should use < 10KB of memory + per_filter_kb = result.memory_delta_kb / 1000 + assert per_filter_kb < 10, f"Filter memory usage {per_filter_kb:.2f}KB is too high" + + def test_greedy_str_memory(self): + """Benchmark GreedyStr memory usage.""" + bench = PerformanceBenchmark("GreedyStr creation", operations=10000) + + def create_greedy(): + return GreedyStr(" ".join([f"arg{i}" for i in range(100)])) + + result = bench.run(create_greedy) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Operations: {result.operation_count:,}") + print(f" Memory delta: +{result.memory_delta_kb:.2f}KB") + print(f" Per operation: {result.memory_delta_kb / result.operation_count:.4f}KB") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + assert result.score >= 50, f"Memory score {result.score} is below threshold 50" + + +class TestThroughputBenchmarks: + """High-throughput benchmarks.""" + + @pytest.mark.asyncio + async def test_high_throughput_bool_validation(self): + """Test boolean validation at high throughput.""" + bench = PerformanceBenchmark("High-throughput bool validation", operations=100000) + + cf = CommandFilter(command_name="test") + cf.handler_params = {"enabled": bool} + values = ["true", "false", "yes", "no", "1", "0"] + + def validate_many(): + for v in values: + cf.validate_and_convert_params([v], cf.handler_params) + + # Run 100k validations across 6 values + result = bench.run(validate_many) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Total operations: {result.operation_count * 6:,}") + print(f" Effective ops/sec: {result.ops_per_second * 6:,.0f}") + print(f" Total time: {result.total_time_ms:.2f}ms") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + # Should handle > 100k validations per second + effective_ops = result.ops_per_second * 6 + assert effective_ops > 100_000, f"Throughput {effective_ops:,.0f} is too low" + + @pytest.mark.asyncio + async def test_command_name_resolution_throughput(self): + """Test command name resolution at high throughput.""" + bench = PerformanceBenchmark("Command name resolution", operations=50000) + + filters = [] + for i in range(50): + cf = CommandFilter(command_name=f"cmd_{i}") + cf.alias = {f"c{i}", f"d{i}"} + filters.append(cf) + + def resolve_all(): + for cf in filters: + cf.get_complete_command_names() + + result = bench.run(resolve_all) + + print(f"\n{'='*60}") + print(f"Benchmark: {result.name}") + print(f" Operations: {result.operation_count:,}") + print(f" Ops/sec: {result.ops_per_second:,.0f}") + print(f" SCORE: {result.score}/100") + print(f"{'='*60}\n") + + assert result.ops_per_second > 100_000, f"Throughput {result.ops_per_second:,.0f} is too low" + + +class TestScoringSummary: + """Summary test that reports overall score.""" + + @pytest.mark.asyncio + async def test_overall_performance_score(self): + """Calculate overall performance score across all benchmarks.""" + scores = [] + + # Test 1: CommandFilter operations + bench1 = PerformanceBenchmark("CommandFilter ops", operations=10000) + filters = [CommandFilter(command_name=f"c{i}") for i in range(50)] + result1 = bench1.run(lambda: [f.get_complete_command_names() for f in filters]) + scores.append(result1.score) + + # Test 2: Bool validation + bench2 = PerformanceBenchmark("Bool validation", operations=50000) + cf = CommandFilter(command_name="test") + cf.handler_params = {"enabled": bool} + result2 = bench2.run(lambda: cf.validate_and_convert_params(["true"], cf.handler_params)) + scores.append(result2.score) + + # Test 3: Int validation + bench3 = PerformanceBenchmark("Int validation", operations=50000) + cf2 = CommandFilter(command_name="test2") + cf2.handler_params = {"count": int} + result3 = bench3.run(lambda: cf2.validate_and_convert_params(["42"], cf2.handler_params)) + scores.append(result3.score) + + overall_score = sum(scores) // len(scores) + + print(f"\n{'='*60}") + print(f"PERFORMANCE BENCHMARK SUMMARY") + print(f"{'='*60}") + print(f" CommandFilter operations: {scores[0]}/100") + print(f" Bool validation: {scores[1]}/100") + print(f" Int validation: {scores[2]}/100") + print(f" {'-'*40}") + print(f" OVERALL SCORE: {overall_score}/100") + print(f"{'='*60}\n") + + assert overall_score >= 60, f"Overall score {overall_score} is below threshold 60"