Graceful degradation, circuit breakers, health checks. Тестирование устойчивости к сбоям
PYTHONHASHSEED=1 pytest tests/integration -q --random-order --reruns 2 --reruns-delay 0.5 --maxfail=1--random-order вскрывает зависимость от кеша между тестами.--reruns используем только для диагностики, не для маскировки.@pytest.mark.integration
def test_graceful_degradation_when_cache_down(cache_down, api_client):
resp = api_client.get("/tasks/u1")
assert resp.status_code == 200
assert resp.json()["id"] == "u1"
assert "cache.unavailable" in api_client.captured_metricscache_down убивает Redis-подключение, проверяем обратное давление и метрики.@pytest.mark.asyncio
async def test_idempotency_on_retry(task_service, make_task_payload):
payload = make_task_payload(idempotency_key="k1")
r1, r2 = await asyncio.gather(
task_service.create_task(payload),
task_service.create_task(payload),
)
assert {r1.status, r2.status} == {"accepted"}
assert await task_service.ledger.count("k1") == 1@pytest.mark.integration
def test_database_connection_pool_exhaustion(db_pool):
conns = [db_pool.acquire() for _ in range(db_pool.max_size)]
with pytest.raises(TimeoutError):
db_pool.acquire(timeout=0.1)
for conn in conns:
conn.release()pytest-xdist -n 4 с max_size ≥ воркеров).import os
import pytest
def pytest_runtest_makereport(item, call):
if call.when == "call" and call.excinfo is not None:
item.config.cache.set(f"flaky/{item.nodeid}", {
"seed": os.getenv("PYTHONHASHSEED"),
"markers": list(item.iter_markers()),
})-m flaky --runslow отдельно от основного.Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: Автоматизация: pytest plugin для flaky tests