mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-15 17:30:13 +08:00
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
This commit is contained in:
LIghtJUNction
0
tests/benchmarks/__init__.py
Normal file
0
tests/benchmarks/__init__.py
Normal file
338
tests/benchmarks/test_performance.py
Normal file
338
tests/benchmarks/test_performance.py
Normal file
@@ -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"
|
||||
Reference in New Issue
Block a user