359 lines
12 KiB
Python
359 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Backend System Test Script
|
|
|
|
Tests all backend components:
|
|
- Database connections (MTG Online + MTG)
|
|
- Redis connection and caching
|
|
- Card database queries
|
|
- API endpoints
|
|
- Health checks
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
# Add backend to path
|
|
sys.path.insert(0, '/home/wall-o/projects/mtgonline/backend')
|
|
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy import text
|
|
|
|
# Database URLs
|
|
COCKATRICE_DB = "postgresql+asyncpg://mtgonline:mtgonline_pass@localhost:5432/mtgonline"
|
|
MTG_DB = "postgresql+asyncpg://mtgonline:mtgonline_pass@localhost:5432/mtgdata"
|
|
REDIS_URL = "redis://localhost:6379/0"
|
|
|
|
# Test URLs - detect environment
|
|
import socket
|
|
try:
|
|
# Try Docker network first
|
|
socket.getaddrinfo('mtg_backend', 8000)
|
|
API_BASE = 'http://mtg_backend:8000'
|
|
except:
|
|
# Fall back to host port
|
|
API_BASE = 'http://localhost:8001'
|
|
print(f"Using API base: {API_BASE}")
|
|
results = {
|
|
"tests_run": 0,
|
|
"tests_passed": 0,
|
|
"tests_failed": 0,
|
|
"failures": [],
|
|
"timestamp": datetime.now().isoformat()
|
|
}
|
|
|
|
async def test_connection(db_url: str, name: str) -> bool:
|
|
"""Test database connection."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
engine = create_async_engine(db_url, echo=False)
|
|
async with engine.connect() as conn:
|
|
result = await conn.execute(text("SELECT 1"))
|
|
await engine.dispose()
|
|
|
|
results["tests_passed"] += 1
|
|
print(f"✓ {name} connection: OK")
|
|
return True
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"{name} connection: {str(e)}")
|
|
print(f"✗ {name} connection: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_redis() -> bool:
|
|
"""Test Redis connection."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
import redis.asyncio as aioredis
|
|
client = aioredis.from_url(REDIS_URL, decode_responses=True)
|
|
await client.ping()
|
|
await client.close()
|
|
|
|
results["tests_passed"] += 1
|
|
print("✓ Redis connection: OK")
|
|
return True
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Redis connection: {str(e)}")
|
|
print(f"✗ Redis connection: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_cache_operations() -> bool:
|
|
"""Test Redis cache operations."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
import redis.asyncio as aioredis
|
|
client = aioredis.from_url(REDIS_URL, decode_responses=True)
|
|
|
|
# Test set/get
|
|
await client.set("test_key", "test_value", ex=60)
|
|
value = await client.get("test_key")
|
|
assert value == "test_value", f"Expected 'test_value', got '{value}'"
|
|
|
|
# Test delete
|
|
await client.delete("test_key")
|
|
value = await client.get("test_key")
|
|
assert value is None, f"Expected None after delete, got '{value}'"
|
|
|
|
await client.close()
|
|
|
|
results["tests_passed"] += 1
|
|
print("✓ Redis cache operations: OK")
|
|
return True
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Redis cache operations: {str(e)}")
|
|
print(f"✗ Redis cache operations: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_mtgonline_tables() -> bool:
|
|
"""Test MTG Online database tables."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
engine = create_async_engine(COCKATRICE_DB, echo=False)
|
|
async with engine.connect() as conn:
|
|
# Test users table
|
|
result = await conn.execute(text("SELECT COUNT(*) FROM mtgonline_users"))
|
|
user_count = result.scalar()
|
|
|
|
# Test decks table
|
|
result = await conn.execute(text("SELECT COUNT(*) FROM mtgonline_decklist_files"))
|
|
deck_count = result.scalar()
|
|
|
|
await engine.dispose()
|
|
|
|
results["tests_passed"] += 1
|
|
print(f"✓ MTG Online tables: OK (users: {user_count}, decks: {deck_count})")
|
|
return True
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"MTG Online tables: {str(e)}")
|
|
print(f"✗ MTG Online tables: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_mtg_tables() -> bool:
|
|
"""Test MTG database tables."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
engine = create_async_engine(MTG_DB, echo=False)
|
|
async with engine.connect() as conn:
|
|
# Test mtg_sets table
|
|
result = await conn.execute(text("SELECT COUNT(*) FROM mtg_sets"))
|
|
set_count = result.scalar()
|
|
|
|
# Test mtg_cards table
|
|
result = await conn.execute(text("SELECT COUNT(*) FROM mtg_cards"))
|
|
card_count = result.scalar()
|
|
|
|
await engine.dispose()
|
|
|
|
results["tests_passed"] += 1
|
|
print(f"✓ MTG tables: OK (sets: {set_count}, cards: {card_count})")
|
|
return True
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"MTG tables: {str(e)}")
|
|
print(f"✗ MTG tables: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_mtg_card_query() -> bool:
|
|
"""Test MTG card query functionality."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
engine = create_async_engine(MTG_DB, echo=False)
|
|
async with engine.connect() as conn:
|
|
# Search for a common card
|
|
result = await conn.execute(
|
|
text("SELECT name, type_line, rarity FROM mtg_cards WHERE name ILIKE '%Black Lotus%' LIMIT 1")
|
|
)
|
|
row = result.fetchone()
|
|
|
|
await engine.dispose()
|
|
|
|
if row:
|
|
results["tests_passed"] += 1
|
|
print(f"✓ MTG card query: OK - Found '{row[0]}' ({row[1]}, {row[2]})")
|
|
return True
|
|
else:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append("MTG card query: No results found")
|
|
print("✗ MTG card query: FAILED - No results found")
|
|
return False
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"MTG card query: {str(e)}")
|
|
print(f"✗ MTG card query: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_health_endpoint() -> bool:
|
|
"""Test backend health endpoint."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
import requests
|
|
response = requests.get("http://localhost:8001/health", timeout=5)
|
|
|
|
if response.status_code == 200:
|
|
results["tests_passed"] += 1
|
|
print(f"✓ Health endpoint: OK ({response.json()})")
|
|
return True
|
|
else:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Health endpoint: Status {response.status_code}")
|
|
print(f"✗ Health endpoint: FAILED - Status {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Health endpoint: {str(e)}")
|
|
print(f"✗ Health endpoint: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_card_search_endpoint() -> bool:
|
|
"""Test card search API endpoint."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
import requests
|
|
response = requests.get(
|
|
"http://localhost:8001/api/v1/mtg/cards/search",
|
|
params={"q": "Lightning Bolt", "limit": 5},
|
|
timeout=10
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
results["tests_passed"] += 1
|
|
print(f"✓ Card search endpoint: OK - Found {len(data.get('results', {}).get('results', []))} cards")
|
|
return True
|
|
else:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Card search endpoint: Status {response.status_code}")
|
|
print(f"✗ Card search endpoint: FAILED - Status {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Card search endpoint: {str(e)}")
|
|
print(f"✗ Card search endpoint: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_statistics_endpoint() -> bool:
|
|
"""Test card statistics API endpoint."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
import requests
|
|
response = requests.get(
|
|
"http://localhost:8001/api/v1/mtg/cards/statistics",
|
|
timeout=10
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
results["tests_passed"] += 1
|
|
stats = data.get("results", {}).get("results", {})
|
|
print(f"✓ Statistics endpoint: OK - {stats.get('total_cards', 0)} cards, {stats.get('total_sets', 0)} sets")
|
|
return True
|
|
else:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Statistics endpoint: Status {response.status_code}")
|
|
print(f"✗ Statistics endpoint: FAILED - Status {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Statistics endpoint: {str(e)}")
|
|
print(f"✗ Statistics endpoint: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def test_auth_endpoint() -> bool:
|
|
"""Test authentication endpoint."""
|
|
results["tests_run"] += 1
|
|
try:
|
|
import requests
|
|
response = requests.post(
|
|
"http://localhost:8001/api/v1/auth/register",
|
|
json={
|
|
"username": "test_user",
|
|
"email": "test@example.com",
|
|
"password": "TestPass123!",
|
|
"confirm_password": "TestPass123!"
|
|
},
|
|
timeout=10
|
|
)
|
|
|
|
if response.status_code in [200, 409]: # 409 means user already exists
|
|
results["tests_passed"] += 1
|
|
print(f"✓ Auth endpoint: OK - Status {response.status_code}")
|
|
return True
|
|
else:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Auth endpoint: Status {response.status_code}")
|
|
print(f"✗ Auth endpoint: FAILED - Status {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
results["tests_failed"] += 1
|
|
results["failures"].append(f"Auth endpoint: {str(e)}")
|
|
print(f"✗ Auth endpoint: FAILED - {str(e)}")
|
|
return False
|
|
|
|
async def run_all_tests():
|
|
"""Run all system tests."""
|
|
print("=" * 60)
|
|
print("BACKEND SYSTEM TEST")
|
|
print("=" * 60)
|
|
print(f"Started at: {datetime.now().isoformat()}")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
# Database connections
|
|
print("Testing Database Connections:")
|
|
print("-" * 60)
|
|
await test_connection(COCKATRICE_DB, "MTG Online PostgreSQL")
|
|
await test_connection(MTG_DB, "MTG PostgreSQL")
|
|
await test_redis()
|
|
print()
|
|
|
|
# Cache operations
|
|
print("Testing Cache Operations:")
|
|
print("-" * 60)
|
|
await test_cache_operations()
|
|
print()
|
|
|
|
# Database tables
|
|
print("Testing Database Tables:")
|
|
print("-" * 60)
|
|
await test_mtgonline_tables()
|
|
await test_mtg_tables()
|
|
await test_mtg_card_query()
|
|
print()
|
|
|
|
# API endpoints
|
|
print("Testing API Endpoints:")
|
|
print("-" * 60)
|
|
await test_health_endpoint()
|
|
await test_card_search_endpoint()
|
|
await test_statistics_endpoint()
|
|
await test_auth_endpoint()
|
|
print()
|
|
|
|
# Summary
|
|
print("=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
print(f"Tests Run: {results['tests_run']}")
|
|
print(f"Tests Passed: {results['tests_passed']}")
|
|
print(f"Tests Failed: {results['tests_failed']}")
|
|
print(f"Success Rate: {(results['tests_passed'] / results['tests_run'] * 100):.1f}%")
|
|
print("=" * 60)
|
|
|
|
if results["failures"]:
|
|
print("\nFailures:")
|
|
for failure in results["failures"]:
|
|
print(f" • {failure}")
|
|
|
|
print(f"\nCompleted at: {datetime.now().isoformat()}")
|
|
print("=" * 60)
|
|
|
|
return results["tests_failed"] == 0
|
|
|
|
if __name__ == "__main__":
|
|
success = asyncio.run(run_all_tests())
|
|
sys.exit(0 if success else 1)
|