453 lines
15 KiB
Python
453 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Comprehensive API Testing Suite
|
|
Tests all API endpoints with realistic test data
|
|
"""
|
|
import asyncio
|
|
import httpx
|
|
import json
|
|
from typing import Dict, Any
|
|
|
|
# Configuration
|
|
API_URL = "http://localhost:8000"
|
|
API_INTERNAL_KEY = "bot-internal-key-9f8e7d6c5b4a3210fedcba9876543210"
|
|
|
|
class Colors:
|
|
GREEN = '\033[92m'
|
|
RED = '\033[91m'
|
|
YELLOW = '\033[93m'
|
|
BLUE = '\033[94m'
|
|
RESET = '\033[0m'
|
|
|
|
class APITester:
|
|
def __init__(self):
|
|
self.client = httpx.AsyncClient(timeout=30.0)
|
|
self.test_results = []
|
|
self.auth_token = None
|
|
self.test_user_id = None
|
|
self.test_telegram_id = 999999999
|
|
self.test_username = "test_user"
|
|
self.test_password = "Test123!@#"
|
|
|
|
async def log(self, message: str, color: str = Colors.RESET):
|
|
print(f"{color}{message}{Colors.RESET}")
|
|
|
|
async def test_endpoint(self, name: str, method: str, url: str,
|
|
expected_status: int = 200, **kwargs) -> Dict[str, Any]:
|
|
"""Test a single endpoint"""
|
|
try:
|
|
if method == "GET":
|
|
response = await self.client.get(url, **kwargs)
|
|
elif method == "POST":
|
|
response = await self.client.post(url, **kwargs)
|
|
elif method == "PATCH":
|
|
response = await self.client.patch(url, **kwargs)
|
|
elif method == "DELETE":
|
|
response = await self.client.delete(url, **kwargs)
|
|
else:
|
|
raise ValueError(f"Unsupported method: {method}")
|
|
|
|
success = response.status_code == expected_status
|
|
|
|
result = {
|
|
"name": name,
|
|
"success": success,
|
|
"status": response.status_code,
|
|
"expected": expected_status
|
|
}
|
|
|
|
if success:
|
|
await self.log(f"✅ {name} - Status: {response.status_code}", Colors.GREEN)
|
|
try:
|
|
data = response.json()
|
|
if data and not isinstance(data, dict):
|
|
await self.log(f" Response: {str(data)[:100]}", Colors.BLUE)
|
|
elif data:
|
|
await self.log(f" Response: {json.dumps(data, indent=2)[:200]}...", Colors.BLUE)
|
|
except:
|
|
pass
|
|
else:
|
|
await self.log(f"❌ {name} - Expected: {expected_status}, Got: {response.status_code}", Colors.RED)
|
|
await self.log(f" Response: {response.text[:200]}", Colors.RED)
|
|
|
|
self.test_results.append(result)
|
|
return response
|
|
|
|
except Exception as e:
|
|
await self.log(f"❌ {name} - Exception: {str(e)}", Colors.RED)
|
|
self.test_results.append({
|
|
"name": name,
|
|
"success": False,
|
|
"error": str(e)
|
|
})
|
|
return None
|
|
|
|
async def setup_test_data(self):
|
|
"""Create test data in the database"""
|
|
await self.log("\n🔧 Setting up test data...", Colors.YELLOW)
|
|
|
|
# Create internal API headers
|
|
internal_headers = {"X-Internal-Key": API_INTERNAL_KEY}
|
|
|
|
# Create a Telegram user
|
|
await self.test_endpoint(
|
|
"Create Telegram Player",
|
|
"POST",
|
|
f"{API_URL}/api/internal/player",
|
|
params={"telegram_id": self.test_telegram_id, "name": "Test Telegram User"},
|
|
headers=internal_headers,
|
|
expected_status=200
|
|
)
|
|
|
|
# Get the player to get their ID
|
|
response = await self.test_endpoint(
|
|
"Get Telegram Player by telegram_id",
|
|
"GET",
|
|
f"{API_URL}/api/internal/player/{self.test_telegram_id}",
|
|
headers=internal_headers,
|
|
expected_status=200
|
|
)
|
|
|
|
if response and response.status_code == 200:
|
|
player_data = response.json()
|
|
self.test_user_id = player_data.get('id')
|
|
await self.log(f" Test user ID: {self.test_user_id}", Colors.BLUE)
|
|
|
|
# Add some items to inventory
|
|
await self.test_endpoint(
|
|
"Add item to inventory (knife)",
|
|
"POST",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}/inventory",
|
|
headers=internal_headers,
|
|
json={"item_id": "knife", "quantity": 1},
|
|
expected_status=200
|
|
)
|
|
|
|
await self.test_endpoint(
|
|
"Add item to inventory (water)",
|
|
"POST",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}/inventory",
|
|
headers=internal_headers,
|
|
json={"item_id": "water", "quantity": 3},
|
|
expected_status=200
|
|
)
|
|
|
|
# Create a dropped item
|
|
await self.test_endpoint(
|
|
"Create dropped item",
|
|
"POST",
|
|
f"{API_URL}/api/internal/dropped-items",
|
|
headers=internal_headers,
|
|
params={"item_id": "bandage", "quantity": 2, "location_id": "start_point"},
|
|
expected_status=200
|
|
)
|
|
|
|
# Create a wandering enemy
|
|
await self.test_endpoint(
|
|
"Spawn wandering enemy",
|
|
"POST",
|
|
f"{API_URL}/api/internal/wandering-enemies",
|
|
headers=internal_headers,
|
|
params={"npc_id": "mutant_rat", "location_id": "start_point", "current_hp": 30, "max_hp": 30},
|
|
expected_status=200
|
|
)
|
|
|
|
async def test_health_check(self):
|
|
await self.log("\n📋 Testing Health Check", Colors.YELLOW)
|
|
await self.test_endpoint(
|
|
"Health Check",
|
|
"GET",
|
|
f"{API_URL}/health"
|
|
)
|
|
|
|
async def test_auth_endpoints(self):
|
|
await self.log("\n🔐 Testing Authentication Endpoints", Colors.YELLOW)
|
|
|
|
# Register
|
|
response = await self.test_endpoint(
|
|
"Register Web User",
|
|
"POST",
|
|
f"{API_URL}/api/auth/register",
|
|
json={
|
|
"username": self.test_username,
|
|
"password": self.test_password,
|
|
"name": "Test User"
|
|
},
|
|
expected_status=200
|
|
)
|
|
|
|
# Login
|
|
response = await self.test_endpoint(
|
|
"Login Web User",
|
|
"POST",
|
|
f"{API_URL}/api/auth/login",
|
|
json={
|
|
"username": self.test_username,
|
|
"password": self.test_password
|
|
},
|
|
expected_status=200
|
|
)
|
|
|
|
if response and response.status_code == 200:
|
|
data = response.json()
|
|
self.auth_token = data.get('access_token')
|
|
await self.log(f" Auth token obtained", Colors.BLUE)
|
|
|
|
# Get current user
|
|
if self.auth_token:
|
|
await self.test_endpoint(
|
|
"Get Current User (Me)",
|
|
"GET",
|
|
f"{API_URL}/api/auth/me",
|
|
headers={"Authorization": f"Bearer {self.auth_token}"}
|
|
)
|
|
|
|
async def test_game_endpoints(self):
|
|
if not self.auth_token:
|
|
await self.log("\n⚠️ Skipping game endpoints (no auth token)", Colors.YELLOW)
|
|
return
|
|
|
|
await self.log("\n🎮 Testing Game Endpoints", Colors.YELLOW)
|
|
|
|
headers = {"Authorization": f"Bearer {self.auth_token}"}
|
|
|
|
# Game state
|
|
await self.test_endpoint(
|
|
"Get Game State",
|
|
"GET",
|
|
f"{API_URL}/api/game/state",
|
|
headers=headers
|
|
)
|
|
|
|
# Profile
|
|
await self.test_endpoint(
|
|
"Get Player Profile",
|
|
"GET",
|
|
f"{API_URL}/api/game/profile",
|
|
headers=headers
|
|
)
|
|
|
|
# Location
|
|
await self.test_endpoint(
|
|
"Get Current Location",
|
|
"GET",
|
|
f"{API_URL}/api/game/location",
|
|
headers=headers
|
|
)
|
|
|
|
# Inventory
|
|
await self.test_endpoint(
|
|
"Get Inventory",
|
|
"GET",
|
|
f"{API_URL}/api/game/inventory",
|
|
headers=headers
|
|
)
|
|
|
|
# Move (should fail - need stamina/valid direction)
|
|
await self.test_endpoint(
|
|
"Move (expect failure)",
|
|
"POST",
|
|
f"{API_URL}/api/game/move",
|
|
headers=headers,
|
|
json={"direction": "north"},
|
|
expected_status=400 # Expect failure
|
|
)
|
|
|
|
# Inspect
|
|
await self.test_endpoint(
|
|
"Inspect Area",
|
|
"POST",
|
|
f"{API_URL}/api/game/inspect",
|
|
headers=headers
|
|
)
|
|
|
|
async def test_internal_endpoints(self):
|
|
await self.log("\n🔧 Testing Internal Bot API Endpoints", Colors.YELLOW)
|
|
|
|
internal_headers = {"X-Internal-Key": API_INTERNAL_KEY}
|
|
|
|
if not self.test_user_id:
|
|
await self.log(" No test user ID available", Colors.RED)
|
|
return
|
|
|
|
# Player operations
|
|
await self.test_endpoint(
|
|
"Get Player by ID",
|
|
"GET",
|
|
f"{API_URL}/api/internal/player/by_id/{self.test_user_id}",
|
|
headers=internal_headers
|
|
)
|
|
|
|
await self.test_endpoint(
|
|
"Update Player",
|
|
"PATCH",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}",
|
|
headers=internal_headers,
|
|
json={"hp": 95}
|
|
)
|
|
|
|
# Inventory operations
|
|
await self.test_endpoint(
|
|
"Get Player Inventory",
|
|
"GET",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}/inventory",
|
|
headers=internal_headers
|
|
)
|
|
|
|
# Movement
|
|
await self.test_endpoint(
|
|
"Move Player",
|
|
"POST",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}/move",
|
|
headers=internal_headers,
|
|
json={"location_id": "abandoned_house"}
|
|
)
|
|
|
|
# Location queries
|
|
await self.test_endpoint(
|
|
"Get Dropped Items in Location",
|
|
"GET",
|
|
f"{API_URL}/api/internal/location/start_point/dropped-items",
|
|
headers=internal_headers
|
|
)
|
|
|
|
await self.test_endpoint(
|
|
"Get Wandering Enemies in Location",
|
|
"GET",
|
|
f"{API_URL}/api/internal/location/start_point/wandering-enemies",
|
|
headers=internal_headers
|
|
)
|
|
|
|
# Combat operations
|
|
await self.test_endpoint(
|
|
"Get Combat State",
|
|
"GET",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}/combat",
|
|
headers=internal_headers
|
|
)
|
|
|
|
# Create combat
|
|
combat_response = await self.test_endpoint(
|
|
"Create Combat",
|
|
"POST",
|
|
f"{API_URL}/api/internal/combat/create",
|
|
headers=internal_headers,
|
|
json={
|
|
"player_id": self.test_user_id,
|
|
"npc_id": "zombie",
|
|
"npc_hp": 50,
|
|
"npc_max_hp": 50,
|
|
"location_id": "abandoned_house",
|
|
"from_wandering": False
|
|
}
|
|
)
|
|
|
|
if combat_response and combat_response.status_code == 200:
|
|
# Update combat
|
|
await self.test_endpoint(
|
|
"Update Combat",
|
|
"PATCH",
|
|
f"{API_URL}/api/internal/combat/{self.test_user_id}",
|
|
headers=internal_headers,
|
|
json={"npc_hp": 40, "turn": "npc"}
|
|
)
|
|
|
|
# End combat
|
|
await self.test_endpoint(
|
|
"End Combat",
|
|
"DELETE",
|
|
f"{API_URL}/api/internal/combat/{self.test_user_id}",
|
|
headers=internal_headers
|
|
)
|
|
|
|
# Cooldown operations
|
|
await self.test_endpoint(
|
|
"Set Cooldown",
|
|
"POST",
|
|
f"{API_URL}/api/internal/cooldown/test_cooldown_key",
|
|
headers=internal_headers,
|
|
params={"duration_seconds": 300}
|
|
)
|
|
|
|
await self.test_endpoint(
|
|
"Get Cooldown",
|
|
"GET",
|
|
f"{API_URL}/api/internal/cooldown/test_cooldown_key",
|
|
headers=internal_headers
|
|
)
|
|
|
|
# Corpse operations
|
|
await self.test_endpoint(
|
|
"Create NPC Corpse",
|
|
"POST",
|
|
f"{API_URL}/api/internal/corpses/npc",
|
|
headers=internal_headers,
|
|
params={
|
|
"npc_id": "zombie",
|
|
"location_id": "abandoned_house",
|
|
"loot_remaining": json.dumps([{"item_id": "cloth", "quantity": 2}])
|
|
}
|
|
)
|
|
|
|
await self.test_endpoint(
|
|
"Get NPC Corpses in Location",
|
|
"GET",
|
|
f"{API_URL}/api/internal/location/abandoned_house/corpses/npc",
|
|
headers=internal_headers
|
|
)
|
|
|
|
# Status effects
|
|
await self.test_endpoint(
|
|
"Get Player Status Effects",
|
|
"GET",
|
|
f"{API_URL}/api/internal/player/{self.test_user_id}/status-effects",
|
|
headers=internal_headers
|
|
)
|
|
|
|
async def print_summary(self):
|
|
await self.log("\n" + "="*60, Colors.BLUE)
|
|
await self.log("📊 TEST SUMMARY", Colors.BLUE)
|
|
await self.log("="*60, Colors.BLUE)
|
|
|
|
total = len(self.test_results)
|
|
passed = sum(1 for r in self.test_results if r.get('success', False))
|
|
failed = total - passed
|
|
|
|
await self.log(f"\nTotal Tests: {total}", Colors.BLUE)
|
|
await self.log(f"Passed: {passed}", Colors.GREEN)
|
|
await self.log(f"Failed: {failed}", Colors.RED if failed > 0 else Colors.GREEN)
|
|
await self.log(f"Success Rate: {(passed/total*100):.1f}%", Colors.GREEN if failed == 0 else Colors.YELLOW)
|
|
|
|
if failed > 0:
|
|
await self.log("\n❌ Failed Tests:", Colors.RED)
|
|
for result in self.test_results:
|
|
if not result.get('success', False):
|
|
await self.log(f" - {result['name']}", Colors.RED)
|
|
if 'error' in result:
|
|
await self.log(f" Error: {result['error']}", Colors.RED)
|
|
elif 'status' in result:
|
|
await self.log(f" Expected: {result['expected']}, Got: {result['status']}", Colors.RED)
|
|
|
|
async def run_all_tests(self):
|
|
await self.log("🚀 Starting API Test Suite", Colors.BLUE)
|
|
await self.log("="*60, Colors.BLUE)
|
|
|
|
try:
|
|
await self.test_health_check()
|
|
await self.setup_test_data()
|
|
await self.test_auth_endpoints()
|
|
await self.test_game_endpoints()
|
|
await self.test_internal_endpoints()
|
|
|
|
await self.print_summary()
|
|
|
|
finally:
|
|
await self.client.aclose()
|
|
|
|
async def main():
|
|
tester = APITester()
|
|
await tester.run_all_tests()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|