Code_of_Conquest/api/scripts/verify_session_persistence.py

#!/usr/bin/env python3
"""
Verification script for Task 8.24: Session Persistence Checkpoint.

Tests the SessionService against real Appwrite database to verify:
1. Solo session creation and storage
2. Session retrieval and ownership validation
3. Conversation history persistence
4. Game state tracking (location, quests, events)
5. Session lifecycle (create, update, end)

Usage:
    python scripts/verify_session_persistence.py

Requirements:
    - Appwrite configured in .env
    - game_sessions collection created in Appwrite
    - At least one character created for testing
"""

import sys
import os

# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from dotenv import load_dotenv
load_dotenv()

from app.services.session_service import (
    SessionService,
    SessionNotFound,
    SessionLimitExceeded,
    SessionValidationError,
    get_session_service,
)
from app.services.character_service import get_character_service, CharacterNotFound
from app.models.enums import SessionStatus, SessionType, LocationType
from app.utils.logging import get_logger

logger = get_logger(__file__)


def print_header(text: str):
    """Print a section header."""
    print(f"\n{'='*60}")
    print(f" {text}")
    print(f"{'='*60}\n")


def print_result(test_name: str, passed: bool, details: str = ""):
    """Print test result."""
    status = "✅ PASS" if passed else "❌ FAIL"
    print(f"{status}: {test_name}")
    if details:
        print(f"       {details}")


def verify_session_creation(service: SessionService, user_id: str, character_id: str) -> str:
    """
    Test 1: Create a solo session and verify it's stored correctly.

    Returns session_id if successful.
    """
    print_header("Test 1: Solo Session Creation")

    try:
        # Create session
        session = service.create_solo_session(
            user_id=user_id,
            character_id=character_id,
            starting_location="Test Town",
            starting_location_type=LocationType.TOWN
        )

        # Verify fields
        checks = [
            (session.session_type == SessionType.SOLO, "session_type is SOLO"),
            (session.solo_character_id == character_id, "solo_character_id matches"),
            (session.user_id == user_id, "user_id matches"),
            (session.turn_number == 0, "turn_number is 0"),
            (session.status == SessionStatus.ACTIVE, "status is ACTIVE"),
            (session.game_state.current_location == "Test Town", "current_location set"),
            (session.game_state.location_type == LocationType.TOWN, "location_type set"),
            ("Test Town" in session.game_state.discovered_locations, "location in discovered"),
        ]

        all_passed = True
        for passed, desc in checks:
            print_result(desc, passed)
            if not passed:
                all_passed = False

        if all_passed:
            print(f"\n   Session ID: {session.session_id}")
            return session.session_id
        else:
            return None

    except Exception as e:
        print_result("Session creation", False, str(e))
        return None


def verify_session_retrieval(service: SessionService, session_id: str, user_id: str) -> bool:
    """
    Test 2: Load session from database and verify data integrity.
    """
    print_header("Test 2: Session Retrieval")

    try:
        # Load session
        session = service.get_session(session_id, user_id)

        checks = [
            (session.session_id == session_id, "session_id matches"),
            (session.user_id == user_id, "user_id matches"),
            (session.session_type == SessionType.SOLO, "session_type preserved"),
            (session.status == SessionStatus.ACTIVE, "status preserved"),
        ]

        all_passed = True
        for passed, desc in checks:
            print_result(desc, passed)
            if not passed:
                all_passed = False

        # Test ownership validation
        try:
            service.get_session(session_id, "wrong_user_id")
            print_result("Ownership validation", False, "Should have raised SessionNotFound")
            all_passed = False
        except SessionNotFound:
            print_result("Ownership validation (wrong user rejected)", True)

        return all_passed

    except Exception as e:
        print_result("Session retrieval", False, str(e))
        return False


def verify_conversation_history(service: SessionService, session_id: str) -> bool:
    """
    Test 3: Add conversation entries and verify persistence.
    """
    print_header("Test 3: Conversation History")

    try:
        # Add first entry
        service.add_conversation_entry(
            session_id=session_id,
            character_id="char_test",
            character_name="Test Hero",
            action="I explore the town",
            dm_response="You find a bustling marketplace..."
        )

        # Add second entry with quest
        service.add_conversation_entry(
            session_id=session_id,
            character_id="char_test",
            character_name="Test Hero",
            action="Talk to the merchant",
            dm_response="The merchant offers you a quest...",
            quest_offered={"quest_id": "test_quest", "name": "Test Quest"}
        )

        # Retrieve and verify
        session = service.get_session(session_id)

        checks = [
            (session.turn_number == 2, f"turn_number is 2 (got {session.turn_number})"),
            (len(session.conversation_history) == 2, f"2 entries in history (got {len(session.conversation_history)})"),
            (session.conversation_history[0].action == "I explore the town", "first action preserved"),
            (session.conversation_history[1].quest_offered is not None, "quest_offered preserved"),
            (session.conversation_history[0].timestamp != "", "timestamp auto-generated"),
        ]

        all_passed = True
        for passed, desc in checks:
            print_result(desc, passed)
            if not passed:
                all_passed = False

        # Test get_recent_history
        recent = service.get_recent_history(session_id, num_turns=1)
        check = len(recent) == 1 and recent[0].turn == 2
        print_result("get_recent_history returns last entry", check)
        if not check:
            all_passed = False

        return all_passed

    except Exception as e:
        print_result("Conversation history", False, str(e))
        return False


def verify_game_state_tracking(service: SessionService, session_id: str) -> bool:
    """
    Test 4: Test location, quest, and event tracking.
    """
    print_header("Test 4: Game State Tracking")

    try:
        all_passed = True

        # Update location
        service.update_location(
            session_id=session_id,
            new_location="Dark Forest",
            location_type=LocationType.WILDERNESS
        )

        session = service.get_session(session_id)
        check = session.game_state.current_location == "Dark Forest"
        print_result("Location updated", check, f"Got: {session.game_state.current_location}")
        if not check:
            all_passed = False

        check = session.game_state.location_type == LocationType.WILDERNESS
        print_result("Location type updated", check)
        if not check:
            all_passed = False

        check = "Dark Forest" in session.game_state.discovered_locations
        print_result("New location added to discovered", check)
        if not check:
            all_passed = False

        # Add quest
        service.add_active_quest(session_id, "quest_1")
        session = service.get_session(session_id)
        check = "quest_1" in session.game_state.active_quests
        print_result("Quest added to active_quests", check)
        if not check:
            all_passed = False

        # Add second quest
        service.add_active_quest(session_id, "quest_2")
        session = service.get_session(session_id)
        check = len(session.game_state.active_quests) == 2
        print_result("Second quest added", check)
        if not check:
            all_passed = False

        # Try to add third quest (should fail)
        try:
            service.add_active_quest(session_id, "quest_3")
            print_result("Max quest limit enforced", False, "Should have raised error")
            all_passed = False
        except SessionValidationError:
            print_result("Max quest limit enforced (2/2)", True)

        # Remove quest
        service.remove_active_quest(session_id, "quest_1")
        session = service.get_session(session_id)
        check = "quest_1" not in session.game_state.active_quests
        print_result("Quest removed", check)
        if not check:
            all_passed = False

        # Add world event
        service.add_world_event(session_id, {"type": "storm", "description": "A storm approaches"})
        session = service.get_session(session_id)
        check = len(session.game_state.world_events) == 1
        print_result("World event added", check)
        if not check:
            all_passed = False

        check = "timestamp" in session.game_state.world_events[0]
        print_result("Event timestamp auto-added", check)
        if not check:
            all_passed = False

        return all_passed

    except Exception as e:
        print_result("Game state tracking", False, str(e))
        return False


def verify_session_lifecycle(service: SessionService, session_id: str, user_id: str) -> bool:
    """
    Test 5: Test session ending and status changes.
    """
    print_header("Test 5: Session Lifecycle")

    try:
        all_passed = True

        # End session
        session = service.end_session(session_id, user_id)
        check = session.status == SessionStatus.COMPLETED
        print_result("Session status set to COMPLETED", check, f"Got: {session.status}")
        if not check:
            all_passed = False

        # Verify persisted
        session = service.get_session(session_id)
        check = session.status == SessionStatus.COMPLETED
        print_result("Completed status persisted", check)
        if not check:
            all_passed = False

        # Verify it's not counted as active
        count = service.count_user_sessions(user_id, active_only=True)
        # Note: This might include other sessions, so just check it works
        print_result(f"count_user_sessions works (active: {count})", True)

        return all_passed

    except Exception as e:
        print_result("Session lifecycle", False, str(e))
        return False


def verify_error_handling(service: SessionService) -> bool:
    """
    Test 6: Test error handling for invalid operations.
    """
    print_header("Test 6: Error Handling")

    all_passed = True

    # Invalid session ID
    try:
        service.get_session("invalid_session_id_12345")
        print_result("Invalid session ID raises error", False)
        all_passed = False
    except SessionNotFound:
        print_result("Invalid session ID raises SessionNotFound", True)
    except Exception as e:
        print_result("Invalid session ID raises error", False, str(e))
        all_passed = False

    return all_passed


def main():
    """Run all verification tests."""
    print("\n" + "="*60)
    print(" Task 8.24: Session Persistence Verification")
    print("="*60)

    # Check environment
    if not os.getenv('APPWRITE_ENDPOINT'):
        print("\n❌ ERROR: Appwrite not configured. Set APPWRITE_* in .env")
        return False

    # Get test user and character
    print("\nSetup: Finding test character...")

    char_service = get_character_service()

    # Try to find an existing character for testing
    # You may need to adjust this based on your test data
    test_user_id = os.getenv('TEST_USER_ID', '')
    test_character_id = os.getenv('TEST_CHARACTER_ID', '')

    if not test_user_id or not test_character_id:
        print("\n⚠️  No TEST_USER_ID or TEST_CHARACTER_ID in .env")
        print("   Will attempt to use mock IDs for basic testing")
        print("   For full integration test, set these environment variables")

        # Use mock approach - only tests that don't need real DB
        print("\n" + "="*60)
        print(" Running Unit Test Verification Only")
        print("="*60)

        # Run the pytest tests instead
        import subprocess
        result = subprocess.run(
            ["python", "-m", "pytest", "tests/test_session_service.py", "-v", "--tb=short"],
            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        )
        return result.returncode == 0

    # Initialize service
    service = get_session_service()

    # Run tests
    results = []

    # Test 1: Create session
    session_id = verify_session_creation(service, test_user_id, test_character_id)
    results.append(("Session Creation", session_id is not None))

    if session_id:
        # Test 2: Retrieve session
        results.append(("Session Retrieval", verify_session_retrieval(service, session_id, test_user_id)))

        # Test 3: Conversation history
        results.append(("Conversation History", verify_conversation_history(service, session_id)))

        # Test 4: Game state tracking
        results.append(("Game State Tracking", verify_game_state_tracking(service, session_id)))

        # Test 5: Session lifecycle
        results.append(("Session Lifecycle", verify_session_lifecycle(service, session_id, test_user_id)))

    # Test 6: Error handling
    results.append(("Error Handling", verify_error_handling(service)))

    # Summary
    print_header("Verification Summary")

    passed = sum(1 for _, p in results if p)
    total = len(results)

    for name, result in results:
        status = "✅" if result else "❌"
        print(f"  {status} {name}")

    print(f"\n  Total: {passed}/{total} tests passed")

    if passed == total:
        print("\n✅ All session persistence tests PASSED!")
        return True
    else:
        print("\n❌ Some tests FAILED. Check output above for details.")
        return False


if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)