From 3a70feaba58bc300528ef0f1c7937d0968085e26 Mon Sep 17 00:00:00 2001 From: Admin Date: Sat, 18 Jul 2026 19:13:30 +0000 Subject: [PATCH] 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 --- backend/scripts/.gitkeep | 0 backend/test_system.py | 348 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 348 insertions(+) create mode 100644 backend/scripts/.gitkeep create mode 100644 backend/test_system.py diff --git a/backend/scripts/.gitkeep b/backend/scripts/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/backend/test_system.py b/backend/test_system.py new file mode 100644 index 0000000..0776ed5 --- /dev/null +++ b/backend/test_system.py @@ -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)