200 lines
6.9 KiB
Python
200 lines
6.9 KiB
Python
"""
|
|
Unit tests for rate limiting
|
|
Tests DebugScopedRateThrottle with bypass logic
|
|
"""
|
|
from django.test import TestCase, override_settings
|
|
from rest_framework.test import APIRequestFactory
|
|
from rest_framework.views import APIView
|
|
from igny8_core.api.throttles import DebugScopedRateThrottle
|
|
from igny8_core.auth.models import User, Account, Plan
|
|
|
|
|
|
class ThrottlesTestCase(TestCase):
|
|
"""Test cases for rate limiting"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
self.factory = APIRequestFactory()
|
|
self.view = APIView()
|
|
self.view.throttle_scope = 'planner'
|
|
|
|
# Create test plan and account
|
|
self.plan = Plan.objects.create(
|
|
name="Test Plan",
|
|
slug="test-plan",
|
|
price=0,
|
|
credits_per_month=1000
|
|
)
|
|
|
|
# Create owner user first
|
|
self.owner_user = User.objects.create_user(
|
|
username='owner',
|
|
email='owner@test.com',
|
|
password='testpass123',
|
|
role='owner'
|
|
)
|
|
|
|
# Create test account with owner
|
|
self.account = Account.objects.create(
|
|
name="Test Account",
|
|
slug="test-account",
|
|
plan=self.plan,
|
|
owner=self.owner_user
|
|
)
|
|
|
|
# Update owner user to have account
|
|
self.owner_user.account = self.account
|
|
self.owner_user.save()
|
|
|
|
# Create regular user
|
|
self.user = User.objects.create_user(
|
|
username='user',
|
|
email='user@test.com',
|
|
password='testpass123',
|
|
role='viewer',
|
|
account=self.account
|
|
)
|
|
|
|
# Create admin user
|
|
self.admin_user = User.objects.create_user(
|
|
username='admin',
|
|
email='admin@test.com',
|
|
password='testpass123',
|
|
role='admin',
|
|
account=self.account
|
|
)
|
|
|
|
# Create system account owner
|
|
self.system_owner = User.objects.create_user(
|
|
username='system_owner',
|
|
email='system_owner@test.com',
|
|
password='testpass123',
|
|
role='owner'
|
|
)
|
|
|
|
# Create system account user
|
|
self.system_account = Account.objects.create(
|
|
name="AWS Admin",
|
|
slug="aws-admin",
|
|
plan=self.plan,
|
|
owner=self.system_owner
|
|
)
|
|
|
|
self.system_owner.account = self.system_account
|
|
self.system_owner.save()
|
|
|
|
self.system_user = User.objects.create_user(
|
|
username='system',
|
|
email='system@test.com',
|
|
password='testpass123',
|
|
role='viewer',
|
|
account=self.system_account
|
|
)
|
|
|
|
@override_settings(DEBUG=True)
|
|
def test_debug_mode_bypass(self):
|
|
"""Test throttling is bypassed in DEBUG mode"""
|
|
throttle = DebugScopedRateThrottle()
|
|
request = self.factory.get('/test/')
|
|
request.user = self.user
|
|
|
|
result = throttle.allow_request(request, self.view)
|
|
self.assertTrue(result) # Should bypass in DEBUG mode
|
|
|
|
@override_settings(DEBUG=False, IGNY8_DEBUG_THROTTLE=True)
|
|
def test_env_bypass(self):
|
|
"""Test throttling is bypassed when IGNY8_DEBUG_THROTTLE=True"""
|
|
throttle = DebugScopedRateThrottle()
|
|
request = self.factory.get('/test/')
|
|
request.user = self.user
|
|
|
|
result = throttle.allow_request(request, self.view)
|
|
self.assertTrue(result) # Should bypass when IGNY8_DEBUG_THROTTLE=True
|
|
|
|
@override_settings(DEBUG=False, IGNY8_DEBUG_THROTTLE=False)
|
|
def test_system_account_bypass(self):
|
|
"""Test throttling is bypassed for system account users"""
|
|
throttle = DebugScopedRateThrottle()
|
|
request = self.factory.get('/test/')
|
|
request.user = self.system_user
|
|
|
|
result = throttle.allow_request(request, self.view)
|
|
self.assertTrue(result) # System account users should bypass
|
|
|
|
@override_settings(DEBUG=False, IGNY8_DEBUG_THROTTLE=False)
|
|
def test_admin_bypass(self):
|
|
"""Test throttling is bypassed for admin/developer users"""
|
|
throttle = DebugScopedRateThrottle()
|
|
request = self.factory.get('/test/')
|
|
request.user = self.admin_user
|
|
|
|
result = throttle.allow_request(request, self.view)
|
|
self.assertTrue(result) # Admin users should bypass
|
|
|
|
@override_settings(DEBUG=False, IGNY8_DEBUG_THROTTLE=False)
|
|
def test_get_rate(self):
|
|
"""Test get_rate returns correct rate for scope"""
|
|
throttle = DebugScopedRateThrottle()
|
|
throttle.scope = 'planner'
|
|
|
|
rate = throttle.get_rate()
|
|
self.assertIsNotNone(rate)
|
|
self.assertIn('/', rate) # Should be in format "60/min"
|
|
|
|
@override_settings(DEBUG=False, IGNY8_DEBUG_THROTTLE=False)
|
|
def test_get_rate_default_fallback(self):
|
|
"""Test get_rate falls back to default if scope not found"""
|
|
throttle = DebugScopedRateThrottle()
|
|
throttle.scope = 'nonexistent_scope'
|
|
|
|
rate = throttle.get_rate()
|
|
self.assertIsNotNone(rate)
|
|
self.assertEqual(rate, '100/min') # Should fallback to default
|
|
|
|
def test_parse_rate_minutes(self):
|
|
"""Test parse_rate correctly parses minutes"""
|
|
throttle = DebugScopedRateThrottle()
|
|
|
|
num, duration = throttle.parse_rate('60/min')
|
|
self.assertEqual(num, 60)
|
|
self.assertEqual(duration, 60)
|
|
|
|
def test_parse_rate_seconds(self):
|
|
"""Test parse_rate correctly parses seconds"""
|
|
throttle = DebugScopedRateThrottle()
|
|
|
|
num, duration = throttle.parse_rate('10/sec')
|
|
self.assertEqual(num, 10)
|
|
self.assertEqual(duration, 1)
|
|
|
|
def test_parse_rate_hours(self):
|
|
"""Test parse_rate correctly parses hours"""
|
|
throttle = DebugScopedRateThrottle()
|
|
|
|
num, duration = throttle.parse_rate('100/hour')
|
|
self.assertEqual(num, 100)
|
|
self.assertEqual(duration, 3600)
|
|
|
|
def test_parse_rate_invalid_format(self):
|
|
"""Test parse_rate handles invalid format gracefully"""
|
|
throttle = DebugScopedRateThrottle()
|
|
|
|
num, duration = throttle.parse_rate('invalid')
|
|
self.assertEqual(num, 100) # Should default to 100
|
|
self.assertEqual(duration, 60) # Should default to 60 seconds (1 min)
|
|
|
|
@override_settings(DEBUG=True)
|
|
def test_debug_info_set(self):
|
|
"""Test debug info is set when bypassing in DEBUG mode"""
|
|
throttle = DebugScopedRateThrottle()
|
|
request = self.factory.get('/test/')
|
|
request.user = self.user
|
|
|
|
result = throttle.allow_request(request, self.view)
|
|
self.assertTrue(result)
|
|
self.assertTrue(hasattr(request, '_throttle_debug_info'))
|
|
self.assertIn('scope', request._throttle_debug_info)
|
|
self.assertIn('rate', request._throttle_debug_info)
|
|
self.assertIn('limit', request._throttle_debug_info)
|
|
|