- Add complete metrics microservice structure - Implement StarRocks database integration - Add user registration data query APIs: - Daily registered users by date range - Recent N days registration data - Registration data by start date and days - Registration summary statistics - Add comprehensive error handling and logging - Include test scripts and documentation
107 lines
3.0 KiB
Python
Executable File
107 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Test script for registration API endpoints
|
|
"""
|
|
import requests
|
|
import json
|
|
from datetime import date, timedelta
|
|
|
|
# API base URL
|
|
BASE_URL = "http://localhost:8009"
|
|
|
|
def test_daily_registered_users():
|
|
"""Test the daily registered users endpoint"""
|
|
print("Testing daily registered users endpoint...")
|
|
|
|
# Test with last 7 days
|
|
end_date = date.today()
|
|
start_date = end_date - timedelta(days=6)
|
|
|
|
url = f"{BASE_URL}/api/metrics/daily-registered-users"
|
|
params = {
|
|
"start_date": str(start_date),
|
|
"end_date": str(end_date),
|
|
"product_id": "freeleaps"
|
|
}
|
|
|
|
try:
|
|
response = requests.get(url, params=params)
|
|
print(f"Status Code: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"Response: {json.dumps(data, indent=2)}")
|
|
print(f"Number of days: {len(data['dates'])}")
|
|
print(f"Total registrations: {data['total_registrations']}")
|
|
else:
|
|
print(f"Error: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f"Request failed: {e}")
|
|
|
|
def test_registration_summary():
|
|
"""Test the registration summary endpoint"""
|
|
print("\nTesting registration summary endpoint...")
|
|
|
|
end_date = date.today()
|
|
start_date = end_date - timedelta(days=6)
|
|
|
|
url = f"{BASE_URL}/api/metrics/registration-summary"
|
|
params = {
|
|
"start_date": str(start_date),
|
|
"end_date": str(end_date),
|
|
"product_id": "freeleaps"
|
|
}
|
|
|
|
try:
|
|
response = requests.get(url, params=params)
|
|
print(f"Status Code: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"Summary: {json.dumps(data, indent=2)}")
|
|
else:
|
|
print(f"Error: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f"Request failed: {e}")
|
|
|
|
def test_post_method():
|
|
"""Test the POST method for daily registered users"""
|
|
print("\nTesting POST method for daily registered users...")
|
|
|
|
end_date = date.today()
|
|
start_date = end_date - timedelta(days=6)
|
|
|
|
url = f"{BASE_URL}/api/metrics/daily-registered-users"
|
|
payload = {
|
|
"start_date": str(start_date),
|
|
"end_date": str(end_date),
|
|
"product_id": "freeleaps"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, json=payload)
|
|
print(f"Status Code: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
print(f"Response: {json.dumps(data, indent=2)}")
|
|
else:
|
|
print(f"Error: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f"Request failed: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting registration API tests...")
|
|
print(f"Testing against: {BASE_URL}")
|
|
print("=" * 50)
|
|
|
|
test_daily_registered_users()
|
|
test_registration_summary()
|
|
test_post_method()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("Tests completed!")
|