Ontwikkel een Mock API voor Geautomatiseerde Validatie van API Rate Limiting en Throttling
description
Met deze prompt kun je een mock API ontwikkelen die specifiek test op rate limiting en throttling, waardoor je vroegtijdig problemen met overbelasting en misbruik kunt detecteren. Dit verbetert de betrouwbaarheid en stabiliteit van je API zonder dat je live systemen belast. Het biedt geautomatiseerde scenario's voor foutafhandeling en herstel, wat handmatige tests overbodig maakt en diepere inzichten geeft dan standaard mocks.
prompt
Help me een mock API te ontwerpen die geautomatiseerde tests uitvoert voor API rate limiting en throttling mechanismen. API doel: <beschrijf het doel van de API>. Belangrijke endpoints: <noem de belangrijkste API endpoints>. Rate limiting paramet ...
try_prompt
generate_helper
disclaimerOnPageApi
tags
ai_answers
provider: openai
model: gpt-4.1-nano
Hier is een voorstel voor een mock API die geautomatiseerde tests voor rate limiting en throttling mechanisme simuleert, inclusief de belangrijkste endpoints en gedrag bij overschrijding van limieten. Daarnaast geef ik suggesties voor het simuleren van escalaties, vertragingen en herstel.
### API-specificaties
- **Endpoints:**
- POST `/register` : Gebruiker registreren
- POST `/login` : Inloggen, ontvangt een token
- POST `/logout` : Uitloggen
- GET `/product/{id}` : Productinformatie ophalen
- GET `/stock/{id}` : Voorraadniveau ophalen
- **Rate limiting:**
- 60 verzoeken per minuut per gebruiker (geauthenticeerd via token)
- Bij overschrijding: `429 Too Many Requests` met `Retry-After` header (in seconden)
### Mock API ontwerp
```json
{
"users": {},
"sessions": {},
"rateLimits": {},
"throttling": {}
}
```
#### 1. Gebruikersregistratie (`/register`)
- Simuleert het toevoegen van een nieuwe gebruiker.
- Geen rate limiting op dit endpoint.
#### 2. Login (`/login`)
- Validatie van gebruikersgegevens.
- Retourneert een token.
- Start een rate limiting teller voor de gebruiker.
#### 3. Logout (`/logout`)
- Beëindigt sessie.
- Reset of behoud rate limiting counters afhankelijk van ontwerp.
#### 4. Rate limiting en throttling logica
- Bij elk verzoek:
- Check of gebruiker al de limiet heeft overschreden binnen de laatste minuut.
- Als ja, retourneer `429 Too Many Requests` met `Retry-After` header.
- Bij herhaald overschrijden, simuleer escalaties (bijv. verhoogde vertragingen).
---
### Implementatie van de mock API in pseudocode (voorbeeld)
```python
import time
# In-memory datastores
users = {}
sessions = {}
rate_limits = {} # {user_id: [timestamps van verzoeken]}
throttled_users = {} # {user_id: eindtijd van throttling}
def check_rate_limit(user_id):
now = time.time()
window_start = now - 60
# Verwijder oude verzoeken
rate_limits[user_id] = [ts for ts in rate_limits.get(user_id, []) if ts > window_start]
if len(rate_limits.get(user_id, [])) >= 60:
# Check of gebruiker al in throttling
if user_id in throttled_users and throttled_users[user_id] > now:
retry_after = int(throttled_users[user_id] - now)
return False, retry_after
else:
# Start throttling
throttling_time = now + 60 # 1 minuut throttling
throttled_users[user_id] = throttling_time
return False, 60
else:
# Registreren verzoek
rate_limits.setdefault(user_id, []).append(now)
return True, None
def handle_request(endpoint, user_token=None):
# Authenticate gebruiker
user_id = sessions.get(user_token)
if not user_id:
return {"status": 401, "message": "Niet ingelogd"}
# Check throttling
allowed, retry_after = check_rate_limit(user_id)
if not allowed:
return {
"status": 429,
"message": "Te veel verzoeken",
"Retry-After": retry_after
}
# Verwerk verzoek volgens endpoint
if endpoint == "/product/{id}":
# Simuleer product info
return {"status": 200, "data": "Productinformatie"}
elif endpoint == "/stock/{id}":
# Simuleer voorraadniveaus
return {"status": 200, "data": "Voorraadniveau"}
# Andere endpoints
return {"status": 200}
# Voor escalaties en vertragingen
def simulate_escalatie(user_id):
# Verhoog limieten of voeg vertragingen toe
pass
def simulate_traging(user_id, duur):
# Voeg een vertraging toe in de verwerking
time.sleep(duur)
def herstel_na_throttling(user_id):
# Reset throttling status
if user_id in throttled_users:
del throttled_users[user_id]
```
---
### Suggesties voor het simuleren van escalaties, vertragingen en herstel
- **Escalaties:**
- Verhoog tijdelijk de limiet (bijv. van 60 naar 100 verzoeken per minuut) na een bepaald aantal overschrijdingen.
- Introduceer een "hard cap" dat voorkomt dat gebruikers meer requests kunnen doen, zelfs na herstel.
- **Vertragingen:**
- Voeg kunstmatige vertragingen toe (bijvoorbeeld `time.sleep()`) wanneer bepaalde drempels worden bereikt om het effect van throttling te simuleren.
- Bijvoorbeeld, na 50 verzoeken binnen een minuut, verhoog de responsetijd met 2 seconden.
- **Herstel na throttling:**
- Na een ingestelde periode (bijv. 5 minuten), reset de counters en status.
- Automatische reset of via een API-call om herstel te simuleren.
---
### Samenvatting
Deze mock API simuleert rate limiting en throttling door verzoeken bij te houden en te beperken tot 60 per minuut per gebruiker. Bij overschrijding wordt een `429` geretourneerd met een `Retry-After` header. Voor verdere testscenario's kunnen escalaties, vertragingen en herstelmechanismen worden geïmplementeerd door de logica uit te breiden met vertragingen en dynamische limieten.
Heeft u nog specifieke wensen of voorbeelden voor de implementatie?
provider: deepseek
model: deepseek-chat
Hier is een ontwerp voor een mock API met geautomatiseerde tests voor rate limiting en throttling:
## Mock API Structuur
### Basis Configuratie
```python
from flask import Flask, jsonify, request
import time
from datetime import datetime, timedelta
app = Flask(__name__)
# Rate limiting configuratie
RATE_LIMIT = 60 # verzoeken per minuut
RATE_WINDOW = 60 # seconden
user_requests = {}
```
### Rate Limiting Implementatie
```python
def check_rate_limit(user_id):
now = time.time()
if user_id not in user_requests:
user_requests[user_id] = []
# Verwijder oude verzoeken buiten het tijdvenster
user_requests[user_id] = [req_time for req_time in user_requests[user_id]
if now - req_time < RATE_WINDOW]
# Controleer of limiet is bereikt
if len(user_requests[user_id]) >= RATE_LIMIT:
return False, RATE_WINDOW - (now - user_requests[user_id][0])
user_requests[user_id].append(now)
return True, 0
```
### API Endpoints
```python
@app.route('/login', methods=['POST'])
def login():
user_id = request.json.get('user_id')
allowed, retry_after = check_rate_limit(user_id)
if not allowed:
response = jsonify({
'error': 'Rate limit exceeded',
'message': 'Te veel verzoeken'
})
response.status_code = 429
response.headers['Retry-After'] = str(int(retry_after))
return response
# Simuleer login logica
return jsonify({'status': 'success', 'message': 'Ingelogd'})
@app.route('/register', methods=['POST'])
def register():
user_id = request.json.get('user_id')
allowed, retry_after = check_rate_limit(user_id)
if not allowed:
response = jsonify({
'error': 'Rate limit exceeded',
'message': 'Te veel verzoeken'
})
response.status_code = 429
response.headers['Retry-After'] = str(int(retry_after))
return response
return jsonify({'status': 'success', 'message': 'Geregistreerd'})
@app.route('/logout', methods=['POST'])
def logout():
user_id = request.json.get('user_id')
allowed, retry_after = check_rate_limit(user_id)
if not allowed:
response = jsonify({
'error': 'Rate limit exceeded',
'message': 'Te veel verzoeken'
})
response.status_code = 429
response.headers['Retry-After'] = str(int(retry_after))
return response
return jsonify({'status': 'success', 'message': 'Uitgelogd'})
```
## Geautomatiseerde Tests
### Basis Rate Limiting Tests
```python
import unittest
import requests
import time
class TestRateLimiting(unittest.TestCase):
BASE_URL = "http://localhost:5000"
def test_normal_usage(self):
"""Test normale gebruik binnen limieten"""
for i in range(30):
response = requests.post(f"{self.BASE_URL}/login",
json={'user_id': 'test_user'})
self.assertIn(response.status_code, [200, 429])
def test_rate_limit_exceeded(self):
"""Test overschrijding rate limit"""
user_id = "limit_test_user"
responses = []
# Verzend 70 verzoeken (meer dan limiet van 60)
for i in range(70):
response = requests.post(f"{self.BASE_URL}/login",
json={'user_id': user_id})
responses.append(response.status_code)
# Controleer dat sommige verzoeken 429 retourneren
self.assertIn(429, responses)
def test_retry_after_header(self):
"""Test aanwezigheid Retry-After header"""
user_id = "header_test_user"
# Overschrijd limiet
for i in range(70):
response = requests.post(f"{self.BASE_URL}/login",
json={'user_id': user_id})
# Controleer Retry-After header bij 429
if response.status_code == 429:
self.assertIn('Retry-After', response.headers)
```
### Geavanceerde Test Scenario's
#### 1. Escalatie Simulatie
```python
def test_escalation_scenarios():
"""Simuleer verschillende gebruikersniveaus"""
users = {
'free_user': 60, # Basis limiet
'premium_user': 120, # Hogere limiet
'admin_user': 300 # Zeer hoge limiet
}
for user_type, limit in users.items():
print(f"Testing {user_type} with limit {limit}")
# Pas API aan voor verschillende limieten per gebruikerstype
```
#### 2. Vertraging Simulatie
```python
def simulate_delayed_recovery():
"""Test herstel na throttling periode"""
user_id = "recovery_test"
# Overschrijd limiet
for i in range(70):
requests.post(f"{BASE_URL}/login", json={'user_id': user_id})
# Wacht exact de retry-after periode
time.sleep(61) # 1 seconde meer dan window
# Test of gebruiker weer kan verzoeken doen
response = requests.post(f"{BASE_URL}/login", json={'user_id': user_id})
assert response.status_code == 200
```
#### 3. Burst Traffic Simulatie
```python
def test_burst_traffic():
"""Simuleer plotselinge piek in verkeer"""
import threading
def make_requests(user_id, count):
for i in range(count):
requests.post(f"{BASE_URL}/login", json={'user_id': user_id})
# Start meerdere threads gelijktijdig
threads = []
for i in range(5):
thread = threading.Thread(target=make_requests,
args=(f"burst_user_{i}", 15))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
```
## Test Suggesties voor Realistische Scenario's
### 1. Gradual Escalatie
```python
def test_gradual_escalation():
"""Test geleidelijke toename van verkeer"""
user_id = "escalation_user"
request_counts = [10, 30, 50, 70, 90] # Stapsgewijs meer verzoeken
for count in request_counts:
print(f"Testing {count} requests")
success_count = 0
for i in range(count):
response = requests.post(f"{BASE_URL}/login",
json={'user_id': user_id})
if response.status_code == 200:
success_count += 1
print(f"Success rate: {success_count}/{count}")
time.sleep(60) # Wacht tussen tests
```
### 2. Herstelstrategieën
```python
def test_recovery_strategies():
"""Test verschillende herstelstrategieën"""
# Exponential backoff
def exponential_backoff(user_id, max_retries=5):
for attempt in range(max_retries):
response = requests.post(f"{BASE_URL}/login",
json={'user_id': user_id})
if response.status_code != 429:
return response
wait_time = (2 ** attempt) # Exponential increase
time.sleep(wait_time)
return None
# Linear backoff
def linear_backoff(user_id, max_retries=5):
for attempt in range(max_retries):
response = requests.post(f"{BASE_URL}/login",
json={'user_id': user_id})
if response.status_code != 429:
return response
time.sleep(attempt + 1) # Linear increase
return None
```
### 3. Monitoring en Metrics
```python
class RateLimitMonitor:
def __init__(self):
self.metrics = {
'total_requests': 0,
'blocked_requests': 0,
'successful_requests': 0
}
def track_request(self, response):
self.metrics['total_requests'] += 1
if response.status_code == 429:
self.metrics['blocked_requests'] += 1
else:
self.metrics['successful_requests'] += 1
def get_success_rate(self):
if self.metrics['total_requests'] == 0:
return 0
return (self.metrics['successful_requests'] /
self.metrics['total_requests'] * 100)
```
## Implementatie Tips
1. **Test Data Variatie**: Gebruik verschillende user_id's voor realistische tests
2. **Tijd Synchronisatie**: Zorg voor nauwkeurige tijdmeting tussen client en server
3. **Parallelle Tests**: Simuleer gelijktijdige gebruikers met threading
4. **Metrics Tracking**: Houd succespercentages en response tijden bij
5. **Cleanup**: Reset rate limiting status tussen test suites
Dit ontwerp biedt een complete basis voor het testen van rate limiting en throttling mechanismen met realistische escalatie- en herstelscenario's.