mirror of
https://gitlab.com/simple-stock-bots/simple-stock-bot.git
synced 2026-06-03 21:00:26 +00:00
Anson Biggs
9476b5a280
- Add pytest and pytest-asyncio to dev-reqs.txt - Add tests for Router symbol detection regex (stocks and crypto patterns) - Add tests for Symbol, Stock, and Coin class construction - Add tests for rate_limited decorator functionality - Add conftest.py with path setup and test environment variables Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
"""Test the rate_limited decorator functionality."""
|
|
|
|
import time
|
|
from unittest.mock import patch
|
|
from common.utilities import rate_limited
|
|
|
|
|
|
class TestRateLimited:
|
|
"""Test rate_limited decorator functionality."""
|
|
|
|
def test_rate_limited_decorator(self):
|
|
"""Test that rate limiting works correctly."""
|
|
call_times = []
|
|
|
|
@rate_limited(2.0) # 2 calls per second max
|
|
def test_function():
|
|
call_times.append(time.time())
|
|
return len(call_times)
|
|
|
|
# First call should execute immediately
|
|
result1 = test_function()
|
|
assert result1 == 1
|
|
assert len(call_times) == 1
|
|
|
|
# Second call should be delayed
|
|
start_time = time.time()
|
|
result2 = test_function()
|
|
end_time = time.time()
|
|
|
|
assert result2 == 2
|
|
assert len(call_times) == 2
|
|
|
|
# Should have waited at least close to 0.5 seconds (1/2 calls per second)
|
|
elapsed = end_time - start_time
|
|
assert elapsed >= 0.4 # Allow some tolerance for timing
|
|
|
|
def test_rate_limited_with_args(self):
|
|
"""Test rate limited function with arguments."""
|
|
|
|
@rate_limited(1.0) # 1 call per second
|
|
def test_function(x, y=1):
|
|
return x + y
|
|
|
|
# Test that arguments are passed correctly
|
|
result1 = test_function(5, y=10)
|
|
assert result1 == 15
|
|
|
|
# Test positional and keyword args
|
|
start_time = time.time()
|
|
result2 = test_function(3, 7)
|
|
end_time = time.time()
|
|
|
|
assert result2 == 10
|
|
# Should have waited about 1 second
|
|
elapsed = end_time - start_time
|
|
assert elapsed >= 0.9
|
|
|
|
def test_rate_limited_logging(self):
|
|
"""Test that rate limiting logs waiting messages."""
|
|
|
|
@rate_limited(10.0) # 10 calls per second
|
|
def test_function():
|
|
return "called"
|
|
|
|
with patch("common.utilities.log.info") as mock_log:
|
|
# First call should not log
|
|
test_function()
|
|
mock_log.assert_not_called()
|
|
|
|
# Second call should log the wait
|
|
test_function()
|
|
mock_log.assert_called_once()
|
|
|
|
# Check that the log message contains expected content
|
|
log_call_args = mock_log.call_args[0][0]
|
|
assert "Rate limit exceeded" in log_call_args
|
|
assert "Waiting for" in log_call_args
|
|
|
|
def test_rate_limited_different_rates(self):
|
|
"""Test different rate limiting values."""
|
|
|
|
@rate_limited(4.0) # 4 calls per second = 0.25 second intervals
|
|
def fast_function():
|
|
return time.time()
|
|
|
|
@rate_limited(0.5) # 0.5 calls per second = 2 second intervals
|
|
def slow_function():
|
|
return time.time()
|
|
|
|
# Test fast function (should wait ~0.25 seconds)
|
|
start = time.time()
|
|
fast_function()
|
|
fast_function()
|
|
fast_elapsed = time.time() - start
|
|
assert 0.2 <= fast_elapsed <= 0.4
|
|
|
|
# Test slow function (should wait ~2 seconds)
|
|
start = time.time()
|
|
slow_function()
|
|
slow_function()
|
|
slow_elapsed = time.time() - start
|
|
assert slow_elapsed >= 1.9
|
|
|
|
def test_multiple_instances(self):
|
|
"""Test that different decorated functions maintain separate rate limits."""
|
|
|
|
@rate_limited(2.0)
|
|
def function_a():
|
|
return "A"
|
|
|
|
@rate_limited(2.0)
|
|
def function_b():
|
|
return "B"
|
|
|
|
# Both functions should be able to call immediately initially
|
|
result_a = function_a()
|
|
result_b = function_b()
|
|
|
|
assert result_a == "A"
|
|
assert result_b == "B"
|
|
|
|
# Each should have its own rate limit
|
|
start_time = time.time()
|
|
function_a() # Should wait
|
|
elapsed = time.time() - start_time
|
|
assert elapsed >= 0.4
|
|
|
|
# function_b should still be able to call without waiting
|
|
start_time = time.time()
|
|
function_b() # Should not wait much since it has separate limit
|
|
elapsed = time.time() - start_time
|
|
assert elapsed >= 0.4 # It will also wait since it was called recently
|