Run a quick smoke test.
(verbose: bool = False)
| 721 | |
| 722 | |
| 723 | def run_quick_test(verbose: bool = False) -> list[BenchmarkResult]: |
| 724 | """Run a quick smoke test.""" |
| 725 | results = [] |
| 726 | |
| 727 | bench = IsolatedBenchmark(dirty_workers=1, dirty_threads=1, verbose=verbose) |
| 728 | |
| 729 | print("\nRunning quick smoke test...") |
| 730 | |
| 731 | try: |
| 732 | bench.start() |
| 733 | bench.warmup(5) |
| 734 | |
| 735 | # Simple test |
| 736 | start_time = time.perf_counter() |
| 737 | latencies, errors = bench.run_benchmark( |
| 738 | action="sleep_task", |
| 739 | args=(10,), |
| 740 | total_requests=100, |
| 741 | concurrency=10, |
| 742 | ) |
| 743 | duration = time.perf_counter() - start_time |
| 744 | |
| 745 | result = BenchmarkResult( |
| 746 | scenario="quick_test", |
| 747 | config={"dirty_workers": 1, "dirty_threads": 1}, |
| 748 | total_requests=100, |
| 749 | successful=len(latencies), |
| 750 | failed=len(errors), |
| 751 | errors=errors[:5] if errors else [], |
| 752 | duration_sec=round(duration, 2), |
| 753 | requests_per_sec=round(len(latencies) / duration, 1), |
| 754 | latency_ms=LatencyStats.from_samples(latencies), |
| 755 | ) |
| 756 | results.append(result) |
| 757 | |
| 758 | print(f" Requests/sec: {result.requests_per_sec:.1f}, " |
| 759 | f"p50: {result.latency_ms.p50:.1f}ms, " |
| 760 | f"failed: {result.failed}") |
| 761 | |
| 762 | if result.failed == 0: |
| 763 | print(" PASS: Quick test successful") |
| 764 | else: |
| 765 | print(f" WARN: {result.failed} requests failed") |
| 766 | |
| 767 | finally: |
| 768 | bench.stop() |
| 769 | |
| 770 | return results |
| 771 | |
| 772 | |
| 773 | def run_config_sweep(verbose: bool = False) -> list[BenchmarkResult]: |
no test coverage detected