#!/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())