- Switched from psycopg2 to asyncpg for async SQLAlchemy support - Fixed router registration in main.py - removed duplicate prefixes - Added user_data export to routers/__init__.py - Refactored decks router to use DeckManager service layer - Integrated FuzzyCardMatcher into card_router search endpoints - Made WishlistCreate.card_id optional for proper schema validation - Set PostgreSQL password and configured scram-sha-256 auth - Updated alembic.ini to use local PostgreSQL instead of Docker hostname - Created generic_schemas.py for reusable schema patterns - Added test_routers.py and test_schema_validation.py test files - All 6 Alembic migrations applied successfully (37 tables created) - Application running on port 8000 with all services connected
320 lines
8.2 KiB
Python
320 lines
8.2 KiB
Python
"""Comprehensive schema validation tests."""
|
|
from datetime import datetime
|
|
from app.schemas import *
|
|
|
|
|
|
def test_user_schemas():
|
|
"""Test user-related schemas."""
|
|
print("Testing User Schemas...")
|
|
|
|
# Test UserCreate
|
|
user = UserCreate(
|
|
username="testuser",
|
|
password="securepass123",
|
|
email="test@example.com",
|
|
country="US"
|
|
)
|
|
assert user.username == "testuser"
|
|
assert user.password == "securepass123"
|
|
print(" ✅ UserCreate validates")
|
|
|
|
# Test UserResponse
|
|
user_resp = UserResponse(
|
|
id=1,
|
|
username="testuser",
|
|
email="test@example.com",
|
|
country="US",
|
|
real_name="Test User",
|
|
privlevel="User",
|
|
vip_status=0,
|
|
is_active=True,
|
|
is_banned=False,
|
|
ban_reason=None,
|
|
creation_date=datetime.now(),
|
|
last_login=None
|
|
)
|
|
assert user_resp.id == 1
|
|
print(" ✅ UserResponse validates")
|
|
|
|
# Test UserUpdate
|
|
update = UserUpdate(
|
|
email="new@example.com",
|
|
country="UK",
|
|
real_name="Updated Name",
|
|
new_password="newsecurepass"
|
|
)
|
|
assert update.email == "new@example.com"
|
|
print(" ✅ UserUpdate validates")
|
|
|
|
|
|
def test_deck_schemas():
|
|
"""Test deck-related schemas."""
|
|
print("Testing Deck Schemas...")
|
|
|
|
# Test DeckCreate
|
|
deck = DeckCreate(
|
|
name="Test Deck",
|
|
content="4 Lightning Bolt\n4 Shock",
|
|
format="native",
|
|
status="DRAUGHT"
|
|
)
|
|
assert deck.name == "Test Deck"
|
|
print(" ✅ DeckCreate validates")
|
|
|
|
# Test UserDeckResponse
|
|
deck_resp = UserDeckResponse(
|
|
id=1,
|
|
user_id=1,
|
|
name="Test Deck",
|
|
status="DRAFT",
|
|
folder_id=None,
|
|
format="standard",
|
|
notes=None,
|
|
is_precedent=False,
|
|
precedent_name=None,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
card_count=0,
|
|
is_owner=True
|
|
)
|
|
assert deck_resp.id == 1
|
|
print(" ✅ UserDeckResponse validates")
|
|
|
|
# Test DeckCardResponse
|
|
card = DeckCardResponse(
|
|
id=1,
|
|
deck_id=1,
|
|
card_id=100,
|
|
quantity=4,
|
|
zone="main",
|
|
position=None
|
|
)
|
|
assert card.quantity == 4
|
|
assert card.zone == "main"
|
|
print(" ✅ DeckCardResponse validates")
|
|
|
|
|
|
def test_card_schemas():
|
|
"""Test card-related schemas."""
|
|
print("Testing Card Schemas...")
|
|
|
|
# Test CardCollectionCreate
|
|
card = CardCollectionCreate(
|
|
card_id=1,
|
|
quantity=4,
|
|
condition="NEAR_MINT",
|
|
language="EN",
|
|
is_foil=False
|
|
)
|
|
assert card.quantity == 4
|
|
print(" ✅ CardCollectionCreate validates")
|
|
|
|
# Test MtgCardResponse
|
|
mtg_card = MtgCardResponse(
|
|
id=1,
|
|
name="Lightning Bolt",
|
|
mana_cost="{R}",
|
|
type_line="Instant",
|
|
oracle_text="Lightning Bolt deals 3 damage to any target.",
|
|
power=None,
|
|
toughness=None,
|
|
rarity="Common",
|
|
layout="normal",
|
|
artist="Dan Frazier",
|
|
created_at=datetime.now()
|
|
)
|
|
assert mtg_card.name == "Lightning Bolt"
|
|
print(" ✅ MtgCardResponse validates")
|
|
|
|
# Test CardImportBatchCreate
|
|
batch = CardImportBatchCreate(
|
|
user_id=1,
|
|
card_names=["Lightning Bolt", "Shock", "Thoughtseize"],
|
|
source="manual"
|
|
)
|
|
assert len(batch.card_names) == 3
|
|
print(" ✅ CardImportBatchCreate validates")
|
|
|
|
|
|
def test_proto_schemas():
|
|
"""Test protocol schemas."""
|
|
print("Testing Protocol Schemas...")
|
|
|
|
# Test SessionCommand
|
|
cmd = SessionCommand(
|
|
message_type="SessionCommand",
|
|
cmd_type=1001,
|
|
cmd_id=1,
|
|
data={"username": "test"}
|
|
)
|
|
assert cmd.cmd_type == 1001
|
|
print(" ✅ SessionCommand validates")
|
|
|
|
# Test GameEvent
|
|
event = GameEvent(
|
|
message_type="GameEvent",
|
|
event_type=1000,
|
|
player_id=1,
|
|
data={"game_id": 123}
|
|
)
|
|
assert event.event_type == 1000
|
|
print(" ✅ GameEvent validates")
|
|
|
|
|
|
def test_group_schemas():
|
|
"""Test group-related schemas."""
|
|
print("Testing Group Schemas...")
|
|
|
|
# Test GroupCreate
|
|
group = GroupCreate(
|
|
name="Test Group",
|
|
description="A test group",
|
|
is_public=True,
|
|
max_members=50
|
|
)
|
|
assert group.name == "Test Group"
|
|
print(" ✅ GroupCreate validates")
|
|
|
|
# Test GroupResponse
|
|
group_resp = GroupResponse(
|
|
id=1,
|
|
name="Test Group",
|
|
description="A test group",
|
|
owner_id=1,
|
|
is_public=True,
|
|
max_members=50,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
member_count=5,
|
|
is_member=True
|
|
)
|
|
assert group_resp.id == 1
|
|
print(" ✅ GroupResponse validates")
|
|
|
|
|
|
def test_network_schemas():
|
|
"""Test network-related schemas."""
|
|
print("Testing Network Schemas...")
|
|
|
|
# Test NetworkCreate
|
|
network = NetworkCreate(
|
|
name="Test Network",
|
|
description="A test network",
|
|
is_public=True
|
|
)
|
|
assert network.name == "Test Network"
|
|
print(" ✅ NetworkCreate validates")
|
|
|
|
# Test NetworkResponse
|
|
network_resp = NetworkResponse(
|
|
id=1,
|
|
name="Test Network",
|
|
description="A test network",
|
|
creator_id=1,
|
|
is_public=True,
|
|
created_at=datetime.now(),
|
|
member_count=10,
|
|
is_member=False
|
|
)
|
|
assert network_resp.id == 1
|
|
print(" ✅ NetworkResponse validates")
|
|
|
|
|
|
def test_validation_errors():
|
|
"""Test validation error handling."""
|
|
print("Testing Validation Errors...")
|
|
|
|
# Test invalid username (too short)
|
|
try:
|
|
UserCreate(username="ab", password="securepass123")
|
|
assert False, "Should have raised validation error"
|
|
except Exception as e:
|
|
assert "min_length" in str(e)
|
|
print(" ✅ Username length validation works")
|
|
|
|
# Test invalid password (too short)
|
|
try:
|
|
UserCreate(username="testuser", password="123")
|
|
assert False, "Should have raised validation error"
|
|
except Exception as e:
|
|
assert "min_length" in str(e)
|
|
print(" ✅ Password length validation works")
|
|
|
|
# Test invalid country code
|
|
try:
|
|
UserCreate(username="testuser", password="securepass123", country="USA")
|
|
assert False, "Should have raised validation error"
|
|
except Exception as e:
|
|
assert "max_length" in str(e)
|
|
print(" ✅ Country code length validation works")
|
|
|
|
|
|
def test_enum_schemas():
|
|
"""Test enum-based schemas."""
|
|
print("Testing Enum Schemas...")
|
|
|
|
# Test DeckStatus
|
|
assert DeckStatus.DRAFT.value == "DRAFT"
|
|
assert DeckStatus.FINAL.value == "FINAL"
|
|
print(" ✅ DeckStatus enum validates")
|
|
|
|
# Test DeckZone
|
|
assert DeckZone.MAIN.value == "main"
|
|
assert DeckZone.SIDEBOARD.value == "sideboard"
|
|
print(" ✅ DeckZone enum validates")
|
|
|
|
# Test CardCondition
|
|
assert CardCondition.NEAR_MINT.value == "NEAR_MINT"
|
|
assert CardCondition.HEAVILY_PLAYED.value == "HEAVILY_PLAYED"
|
|
print(" ✅ CardCondition enum validates")
|
|
|
|
|
|
def test_generic_schemas():
|
|
"""Test generic response schemas."""
|
|
print("Testing Generic Schemas...")
|
|
|
|
# Test MessageResponse
|
|
msg = MessageResponse(message="Success")
|
|
assert msg.message == "Success"
|
|
print(" ✅ MessageResponse validates")
|
|
|
|
# Test CountResponse
|
|
count = CountResponse(count=42)
|
|
assert count.count == 42
|
|
print(" ✅ CountResponse validates")
|
|
|
|
# Test ErrorResponse
|
|
error = ErrorResponse(detail="Something went wrong")
|
|
assert error.detail == "Something went wrong"
|
|
print(" ✅ ErrorResponse validates")
|
|
|
|
|
|
def main():
|
|
print("Schema Validation Tests")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
test_user_schemas()
|
|
test_deck_schemas()
|
|
test_card_schemas()
|
|
test_proto_schemas()
|
|
test_group_schemas()
|
|
test_network_schemas()
|
|
test_validation_errors()
|
|
test_enum_schemas()
|
|
test_generic_schemas()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✅ All schema validation tests passed!")
|
|
return 0
|
|
except Exception as e:
|
|
print(f"\n❌ Test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
sys.exit(main())
|