File size: 3,678 Bytes
fea1bd1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
# -*- coding: utf-8 -*-
"""
Testes para o RateLimiter (rate_limiter.py).
"""
import pytest
import time
from unittest.mock import patch
from superezio_enterprise.rate_limiter import RateLimiter
from superezio_enterprise.config import EnterpriseConfig
@pytest.fixture
def mock_config():
"""Fornece uma configuração mockada para isolar os testes do limitador."""
# Um limite baixo para facilitar o teste de esgotamento de tokens
return EnterpriseConfig(rate_limit_per_minute=5)
@pytest.fixture
def limiter_instance(mock_config):
"""Fornece uma instância limpa do RateLimiter para cada teste."""
# Como RateLimiter é um singleton, precisamos garantir um estado limpo
with patch('superezio_enterprise.rate_limiter.CONFIG', mock_config):
instance = RateLimiter()
# Reinicializa o estado interno para o teste
instance.rate = float(mock_config.rate_limit_per_minute)
instance.per_seconds = 60.0
instance._tokens = instance.rate
instance._last_update = time.monotonic()
instance.tokens_per_second = instance.rate / instance.per_seconds
yield instance
def test_rate_limiter_singleton(limiter_instance):
"""Testa se o RateLimiter segue o padrão singleton."""
instance1 = limiter_instance
instance2 = RateLimiter()
assert instance1 is instance2
def test_initial_tokens(limiter_instance):
"""Testa se o número inicial de tokens está correto."""
assert limiter_instance._tokens == limiter_instance.rate
def test_acquire_success(limiter_instance):
"""Testa se é possível adquirir tokens quando há suficientes."""
assert limiter_instance.acquire(1) is True
assert limiter_instance.acquire(2) is True
# 5 (inicial) - 1 - 2 = 2 tokens restantes
assert limiter_instance._tokens == pytest.approx(2)
def test_acquire_failure_due_to_exhaustion(limiter_instance):
"""Testa se a aquisição falha quando não há tokens suficientes."""
# Consome todos os 5 tokens iniciais
assert limiter_instance.acquire(5) is True
assert limiter_instance._tokens == pytest.approx(0)
# A próxima tentativa deve falhar
assert limiter_instance.acquire(1) is False
def test_token_refill(limiter_instance):
"""Testa se os tokens são reabastecidos ao longo do tempo."""
# Esgota os tokens
assert limiter_instance.acquire(5) is True
assert limiter_instance.acquire(1) is False
# O reabastecimento é de 5 tokens por 60 segundos, ou 1/12 tokens por segundo.
# Esperar 12 segundos deve gerar 1 token.
time.sleep(12)
# A aquisição agora deve ser bem-sucedida
assert limiter_instance.acquire(1) is True
assert limiter_instance._tokens == pytest.approx(0, abs=0.1) # Pode haver uma pequena sobra
def test_acquire_burst(limiter_instance):
"""Testa se o balde de tokens permite rajadas (bursts)."""
# O limite é 5, então uma rajada de 5 deve ser permitida
assert limiter_instance.acquire(5) is True
assert limiter_instance._tokens == pytest.approx(0)
def test_refill_does_not_exceed_capacity(limiter_instance):
"""Testa se o reabastecimento não ultrapassa a capacidade máxima."""
# Consome 2 tokens
limiter_instance.acquire(2)
assert limiter_instance._tokens == pytest.approx(3)
# Espera um tempo longo, que teoricamente geraria mais de 2 tokens
time.sleep(36) # Deve gerar 3 tokens (5/60 * 36 = 3)
# O número de tokens não deve exceder a capacidade máxima (5)
limiter_instance.acquire(0) # Força a atualização do reabastecimento
assert limiter_instance._tokens <= limiter_instance.rate
assert limiter_instance._tokens == pytest.approx(5)
|