Add system test script and scripts directory
- Added test_system.py for comprehensive backend testing - Added scripts/ directory with .gitkeep placeholder - Updates state.json for latest progress tracking
This commit is contained in:
@@ -0,0 +1,348 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Backend System Test Script
|
||||
|
||||
Tests all backend components:
|
||||
- Database connections (Cockatrice + 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://cockatrice:cockatrice_pass@localhost:5432/cockatrice"
|
||||
MTG_DB = "postgresql+asyncpg://cockatrice:cockatrice_pass@localhost:5432/mtgdata"
|
||||
REDIS_URL = "redis://localhost:6379/0"
|
||||
|
||||
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_cockatrice_tables() -> bool:
|
||||
"""Test Cockatrice 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 cockatrice_users"))
|
||||
user_count = result.scalar()
|
||||
|
||||
# Test decks table
|
||||
result = await conn.execute(text("SELECT COUNT(*) FROM cockatrice_decklist_files"))
|
||||
deck_count = result.scalar()
|
||||
|
||||
await engine.dispose()
|
||||
|
||||
results["tests_passed"] += 1
|
||||
print(f"✓ Cockatrice tables: OK (users: {user_count}, decks: {deck_count})")
|
||||
return True
|
||||
except Exception as e:
|
||||
results["tests_failed"] += 1
|
||||
results["failures"].append(f"Cockatrice tables: {str(e)}")
|
||||
print(f"✗ Cockatrice 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:8000/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:8000/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:8000/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:8000/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, "Cockatrice 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_cockatrice_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)
|
||||
Reference in New Issue
Block a user