Использование hypothesis для performance тестов, поиск узких мест
«Нахождение узких мест и регрессий производительности до production.»
import time
from hypothesis import given, strategies as st, settings
def slow_function(data):
"""Функция для тестирования"""
time.sleep(0.001 * len(data))
return sum(data)
@given(st.lists(st.integers(), max_size=1000))
def test_performance_baseline(data):
start = time.perf_counter()
result = slow_function(data)
elapsed = time.perf_counter() - start
# Проверка что время в допустимых пределах
assert elapsed < 0.1, f"Too slow: {elapsed}s for {len(data)} items"
assert result == sum(data)from hypothesis import target
@given(st.lists(st.integers(), max_size=10000))
def test_scaling(data):
start = time.perf_counter()
result = slow_function(data)
elapsed = time.perf_counter() - start
# Сообщаем Hypothesis что нас интересуют большие данные
target(len(data), label='input_size')
target(elapsed, label='execution_time')
# Проверка линейной сложности
if len(data) > 0:
time_per_item = elapsed / len(data)
assert time_per_item < 0.001, f"Possible quadratic complexity: {time_per_item}s/item"from locust import HttpUser, task, between
from hypothesis import given, strategies as st
import requests
class APIUser(HttpUser):
wait_time = between(0.1, 0.5)
@task(3)
def get_item(self):
item_id = self.random.randint(1, 10000)
self.client.get(f"/api/items/{item_id}")
@task(1)
def create_item(self):
item_data = {
"name": f"Item {self.random.randint(1, 1000)}",
"price": self.random.uniform(0.01, 1000),
"quantity": self.random.randint(1, 100)
}
self.client.post("/api/items/", json=item_data)
# Property-based нагрузочный тест
@given(st.integers(min_value=1, max_value=100))
def test_concurrent_requests(concurrent_users):
"""Тестирование под нагрузкой"""
import concurrent.futures
def make_request():
response = requests.get("http://localhost:8000/api/items/1")
return response.status_code == 200
with concurrent.futures.ThreadPoolExecutor(max_workers=concurrent_users) as executor:
futures = [executor.submit(make_request) for _ in range(concurrent_users)]
results = [f.result() for f in futures]
assert all(results), f"Some requests failed with {concurrent_users} concurrent users"import tracemalloc
@given(st.lists(st.integers(), max_size=100000))
def test_memory_usage(data):
tracemalloc.start()
result = slow_function(data)
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
# Проверка что память не растёт линейно с размером данных
# (для функций которые должны быть O(1) по памяти)
assert peak < 10 * 1024 * 1024, f"Memory leak: {peak / 1024 / 1024:.2f}MB"Hypothesis автоматически находит проблемы с производительностью на больших данных и под нагрузкой.
Следующая тема: Интеграция с CI/CD — настройка в CI, database seeding, hypothesis в production.
Далее: Интеграция с CI/CD и Production