454 lines
17 KiB
Python
454 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Comprehensive API Test Suite
|
|
Tests all major game functionality including:
|
|
- Authentication (web & telegram)
|
|
- Player creation and management
|
|
- Movement and exploration
|
|
- Inventory and items
|
|
- Combat system
|
|
- Interactables
|
|
- Admin functions
|
|
"""
|
|
|
|
import asyncio
|
|
import httpx
|
|
import json
|
|
from datetime import datetime
|
|
import sys
|
|
|
|
# Configuration
|
|
BASE_URL = "http://localhost:8000"
|
|
API_INTERNAL_KEY = "bot-internal-key-9f8e7d6c5b4a3210fedcba9876543210"
|
|
|
|
# ANSI color codes for pretty output
|
|
class Colors:
|
|
GREEN = '\033[92m'
|
|
RED = '\033[91m'
|
|
YELLOW = '\033[93m'
|
|
BLUE = '\033[94m'
|
|
PURPLE = '\033[95m'
|
|
CYAN = '\033[96m'
|
|
BOLD = '\033[1m'
|
|
END = '\033[0m'
|
|
|
|
class TestRunner:
|
|
def __init__(self):
|
|
self.passed = 0
|
|
self.failed = 0
|
|
self.tests = []
|
|
self.client = None
|
|
self.test_user = None
|
|
self.test_token = None
|
|
|
|
async def setup(self):
|
|
"""Initialize HTTP client"""
|
|
self.client = httpx.AsyncClient(timeout=30.0, follow_redirects=True)
|
|
|
|
async def cleanup(self):
|
|
"""Cleanup resources"""
|
|
if self.client:
|
|
await self.client.aclose()
|
|
|
|
def log_test(self, name: str, passed: bool, details: str = ""):
|
|
"""Log test result"""
|
|
status = f"{Colors.GREEN}✅ PASS{Colors.END}" if passed else f"{Colors.RED}❌ FAIL{Colors.END}"
|
|
print(f"{status} - {name}")
|
|
if details:
|
|
print(f" {Colors.CYAN}→ {details}{Colors.END}")
|
|
|
|
self.tests.append({"name": name, "passed": passed, "details": details})
|
|
if passed:
|
|
self.passed += 1
|
|
else:
|
|
self.failed += 1
|
|
|
|
def print_summary(self):
|
|
"""Print test summary"""
|
|
total = self.passed + self.failed
|
|
rate = (self.passed / total * 100) if total > 0 else 0
|
|
|
|
print(f"\n{Colors.BOLD}{'='*70}{Colors.END}")
|
|
print(f"{Colors.BOLD}TEST SUMMARY{Colors.END}")
|
|
print(f"{Colors.BOLD}{'='*70}{Colors.END}")
|
|
print(f"Total Tests: {Colors.BOLD}{total}{Colors.END}")
|
|
print(f"Passed: {Colors.GREEN}{self.passed}{Colors.END}")
|
|
print(f"Failed: {Colors.RED}{self.failed}{Colors.END}")
|
|
print(f"Success Rate: {Colors.YELLOW}{rate:.1f}%{Colors.END}")
|
|
print(f"{Colors.BOLD}{'='*70}{Colors.END}\n")
|
|
|
|
if self.failed > 0:
|
|
print(f"{Colors.RED}Failed tests:{Colors.END}")
|
|
for test in self.tests:
|
|
if not test['passed']:
|
|
print(f" • {test['name']}: {test['details']}")
|
|
|
|
async def test_health_check(self):
|
|
"""Test health check endpoint"""
|
|
try:
|
|
response = await self.client.get(f"{BASE_URL}/health")
|
|
passed = response.status_code == 200
|
|
self.log_test("Health Check", passed, f"Status: {response.status_code}")
|
|
except Exception as e:
|
|
self.log_test("Health Check", False, f"Error: {str(e)}")
|
|
|
|
async def test_register_web_user(self):
|
|
"""Test web user registration"""
|
|
timestamp = int(datetime.now().timestamp())
|
|
username = f"test_user_{timestamp}"
|
|
|
|
try:
|
|
response = await self.client.post(
|
|
f"{BASE_URL}/api/auth/register",
|
|
json={
|
|
"username": username,
|
|
"password": "TestPass123!",
|
|
"character_name": "Test Survivor"
|
|
}
|
|
)
|
|
|
|
# Registration can return 200 or 201, both with a token
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
self.test_user = username
|
|
self.test_token = data.get('access_token')
|
|
self.log_test("Web User Registration", True,
|
|
f"Created user: {username}, Got token: {self.test_token[:20] if self.test_token else 'None'}...")
|
|
else:
|
|
self.log_test("Web User Registration", False,
|
|
f"Status: {response.status_code}, Response: {response.text[:200]}")
|
|
except Exception as e:
|
|
self.log_test("Web User Registration", False, f"Error: {str(e)}")
|
|
|
|
async def test_login(self):
|
|
"""Test user login"""
|
|
if not self.test_user:
|
|
self.log_test("Login", False, "No test user available")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.post(
|
|
f"{BASE_URL}/api/auth/login",
|
|
json={
|
|
"username": self.test_user,
|
|
"password": "TestPass123!"
|
|
}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.test_token = data.get('access_token')
|
|
self.log_test("Login", True, f"Got token: {self.test_token[:20]}...")
|
|
else:
|
|
self.log_test("Login", False, f"Status: {response.status_code}")
|
|
except Exception as e:
|
|
self.log_test("Login", False, f"Error: {str(e)}")
|
|
|
|
async def test_get_user_info(self):
|
|
"""Test getting current user info"""
|
|
if not self.test_token:
|
|
self.log_test("Get User Info", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.get(
|
|
f"{BASE_URL}/api/auth/me",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.log_test("Get User Info", True,
|
|
f"User: {data.get('username')}, Location: {data.get('location_id')}")
|
|
else:
|
|
self.log_test("Get User Info", False, f"Status: {response.status_code}")
|
|
except Exception as e:
|
|
self.log_test("Get User Info", False, f"Error: {str(e)}")
|
|
|
|
async def test_get_location(self):
|
|
"""Test getting current location details"""
|
|
if not self.test_token:
|
|
self.log_test("Get Location", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.get(
|
|
f"{BASE_URL}/api/game/location",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
directions = data.get('directions', [])
|
|
interactables = data.get('interactables', [])
|
|
self.log_test("Get Location", True,
|
|
f"{data.get('name')} - Directions: {directions}, Interactables: {len(interactables)}")
|
|
else:
|
|
self.log_test("Get Location", False,
|
|
f"Status: {response.status_code}, Response: {response.text[:200]}")
|
|
except Exception as e:
|
|
self.log_test("Get Location", False, f"Error: {str(e)}")
|
|
|
|
async def test_inspect_area(self):
|
|
"""Test inspecting the current area"""
|
|
if not self.test_token:
|
|
self.log_test("Inspect Area", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.post(
|
|
f"{BASE_URL}/api/game/inspect",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.log_test("Inspect Area", True, f"Found: {data.get('message', '')[:100]}")
|
|
else:
|
|
self.log_test("Inspect Area", False, f"Status: {response.status_code}")
|
|
except Exception as e:
|
|
self.log_test("Inspect Area", False, f"Error: {str(e)}")
|
|
|
|
async def test_get_inventory(self):
|
|
"""Test getting player inventory"""
|
|
if not self.test_token:
|
|
self.log_test("Get Inventory", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.get(
|
|
f"{BASE_URL}/api/game/inventory",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
items = response.json() # Returns array directly
|
|
self.log_test("Get Inventory", True, f"Items: {len(items)}")
|
|
else:
|
|
self.log_test("Get Inventory", False, f"Status: {response.status_code}")
|
|
except Exception as e:
|
|
self.log_test("Get Inventory", False, f"Error: {str(e)}")
|
|
|
|
async def test_get_profile(self):
|
|
"""Test getting player profile"""
|
|
if not self.test_token:
|
|
self.log_test("Get Profile", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.get(
|
|
f"{BASE_URL}/api/game/profile",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.log_test("Get Profile", True,
|
|
f"HP: {data.get('hp')}/{data.get('max_hp')}, Level: {data.get('level')}")
|
|
else:
|
|
self.log_test("Get Profile", False, f"Status: {response.status_code}")
|
|
except Exception as e:
|
|
self.log_test("Get Profile", False, f"Error: {str(e)}")
|
|
|
|
async def test_movement(self):
|
|
"""Test player movement"""
|
|
if not self.test_token:
|
|
self.log_test("Movement", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
# First get current location to see available directions
|
|
loc_response = await self.client.get(
|
|
f"{BASE_URL}/api/game/location",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if loc_response.status_code != 200:
|
|
self.log_test("Movement", False, "Could not get current location")
|
|
return
|
|
|
|
location = loc_response.json()
|
|
directions = location.get('directions', [])
|
|
|
|
if not directions:
|
|
self.log_test("Movement", False,
|
|
f"No directions available at {location.get('name')}")
|
|
return
|
|
|
|
# Try to move in the first available direction
|
|
test_direction = directions[0]
|
|
response = await self.client.post(
|
|
f"{BASE_URL}/api/game/move",
|
|
headers={"Authorization": f"Bearer {self.test_token}"},
|
|
json={"direction": test_direction}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.log_test("Movement", True,
|
|
f"Moved {test_direction} to {data.get('new_location_id')}")
|
|
|
|
# Move back
|
|
back_direction = {"north": "south", "south": "north",
|
|
"east": "west", "west": "east",
|
|
"northeast": "southwest", "southwest": "northeast",
|
|
"northwest": "southeast", "southeast": "northwest"}.get(test_direction)
|
|
|
|
if back_direction:
|
|
await self.client.post(
|
|
f"{BASE_URL}/api/game/move",
|
|
headers={"Authorization": f"Bearer {self.test_token}"},
|
|
json={"direction": back_direction}
|
|
)
|
|
else:
|
|
error_msg = response.json().get('detail', response.text[:200])
|
|
self.log_test("Movement", False,
|
|
f"Status: {response.status_code}, Error: {error_msg}")
|
|
except Exception as e:
|
|
self.log_test("Movement", False, f"Error: {str(e)}")
|
|
|
|
async def test_interactable(self):
|
|
"""Test interacting with objects"""
|
|
if not self.test_token:
|
|
self.log_test("Interactables", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
# Get current location
|
|
loc_response = await self.client.get(
|
|
f"{BASE_URL}/api/game/location",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if loc_response.status_code != 200:
|
|
self.log_test("Interactables", False, "Could not get location")
|
|
return
|
|
|
|
location = loc_response.json()
|
|
interactables = location.get('interactables', [])
|
|
|
|
if not interactables:
|
|
self.log_test("Interactables", False,
|
|
f"No interactables at {location.get('name')}")
|
|
return
|
|
|
|
# Try to interact with first interactable
|
|
interactable = interactables[0]
|
|
actions = interactable.get('actions', [])
|
|
|
|
if not actions:
|
|
self.log_test("Interactables", False, "No actions available")
|
|
return
|
|
|
|
action = actions[0]
|
|
response = await self.client.post(
|
|
f"{BASE_URL}/api/game/interact",
|
|
headers={"Authorization": f"Bearer {self.test_token}"},
|
|
json={
|
|
"interactable_id": interactable['instance_id'],
|
|
"action_id": action['id']
|
|
}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.log_test("Interactables", True,
|
|
f"Action '{action['name']}' on {interactable['name']}: {data.get('message', '')[:100]}")
|
|
else:
|
|
self.log_test("Interactables", False,
|
|
f"Status: {response.status_code}, Error: {response.text[:200]}")
|
|
except Exception as e:
|
|
self.log_test("Interactables", False, f"Error: {str(e)}")
|
|
|
|
async def test_game_state(self):
|
|
"""Test getting full game state"""
|
|
if not self.test_token:
|
|
self.log_test("Game State", False, "No auth token")
|
|
return
|
|
|
|
try:
|
|
response = await self.client.get(
|
|
f"{BASE_URL}/api/game/state",
|
|
headers={"Authorization": f"Bearer {self.test_token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
player = data.get('player', {})
|
|
location = data.get('location', {})
|
|
inventory = data.get('inventory', [])
|
|
self.log_test("Game State", True,
|
|
f"Player: {player.get('name')}, Location: {location.get('name')}, Items: {len(inventory)}")
|
|
else:
|
|
self.log_test("Game State", False,
|
|
f"Status: {response.status_code}, Error: {response.text[:200]}")
|
|
except Exception as e:
|
|
self.log_test("Game State", False, f"Error: {str(e)}")
|
|
|
|
async def test_image_serving(self):
|
|
"""Test that images are being served correctly"""
|
|
try:
|
|
# Test a known image path
|
|
response = await self.client.get(f"{BASE_URL}/images/locations/downtown.png")
|
|
|
|
if response.status_code == 200 and 'image' in response.headers.get('content-type', ''):
|
|
self.log_test("Image Serving", True,
|
|
f"Image served correctly, size: {len(response.content)} bytes")
|
|
else:
|
|
self.log_test("Image Serving", False,
|
|
f"Status: {response.status_code}, Content-Type: {response.headers.get('content-type')}")
|
|
except Exception as e:
|
|
self.log_test("Image Serving", False, f"Error: {str(e)}")
|
|
|
|
async def run_all_tests(self):
|
|
"""Run all tests in sequence"""
|
|
print(f"\n{Colors.BOLD}{Colors.PURPLE}{'='*70}{Colors.END}")
|
|
print(f"{Colors.BOLD}{Colors.PURPLE}COMPREHENSIVE API TEST SUITE{Colors.END}")
|
|
print(f"{Colors.BOLD}{Colors.PURPLE}{'='*70}{Colors.END}\n")
|
|
|
|
await self.setup()
|
|
|
|
try:
|
|
# Basic health check
|
|
print(f"\n{Colors.BOLD}Testing System Health{Colors.END}")
|
|
await self.test_health_check()
|
|
await self.test_image_serving()
|
|
|
|
# Authentication flow
|
|
print(f"\n{Colors.BOLD}Testing Authentication{Colors.END}")
|
|
await self.test_register_web_user()
|
|
await self.test_login()
|
|
await self.test_get_user_info()
|
|
|
|
# Game state
|
|
print(f"\n{Colors.BOLD}Testing Game State{Colors.END}")
|
|
await self.test_get_profile()
|
|
await self.test_get_location()
|
|
await self.test_get_inventory()
|
|
await self.test_game_state()
|
|
|
|
# Gameplay
|
|
print(f"\n{Colors.BOLD}Testing Gameplay{Colors.END}")
|
|
await self.test_inspect_area()
|
|
await self.test_movement()
|
|
await self.test_interactable()
|
|
|
|
# Summary
|
|
self.print_summary()
|
|
|
|
finally:
|
|
await self.cleanup()
|
|
|
|
|
|
async def main():
|
|
"""Main entry point"""
|
|
runner = TestRunner()
|
|
await runner.run_all_tests()
|
|
|
|
# Exit with appropriate code
|
|
sys.exit(0 if runner.failed == 0 else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|