roa2web-service-auto/reports-app/telegram-bot/test_db.py

"""
Test script for database initialization and operations.

⚠️ MANUAL INTEGRATION TEST - Not run by default in CI/CD

This script tests SQLite database operations including user management,
authentication codes, and session handling. Creates test data in the
actual database (data/telegram_bot.db).

REQUIREMENTS:
- SQLite database access (data/ directory must be writable)
- No external dependencies (standalone test)

USAGE:
  # Run as script
  python test_db.py

  # Run via pytest (requires -m integration)
  pytest test_db.py -m integration

NOTE: Test function marked with @pytest.mark.integration
"""

import pytest
import asyncio
import sys
from pathlib import Path
from datetime import datetime, timedelta

# Add app directory to path
sys.path.insert(0, str(Path(__file__).parent))

from app.db import (
    init_database,
    get_database_stats,
    create_or_update_user,
    get_user,
    is_user_linked,
    link_user_to_oracle,
    create_auth_code,
    verify_and_use_auth_code,
    create_session,
    get_user_active_session,
    DB_PATH,
)


@pytest.mark.integration
async def test_database():
    """Test all database operations."""
    print("=" * 60)
    print("🧪 ROA2WEB Telegram Bot - Database Test")
    print("=" * 60)

    try:
        # Test 1: Initialize database
        print("\n✅ Test 1: Initialize database")
        await init_database()
        print(f"   Database file: {DB_PATH}")
        print(f"   Database exists: {DB_PATH.exists()}")

        # Test 2: Create a test user
        print("\n✅ Test 2: Create test user")
        user_id = 123456789
        success = await create_or_update_user(
            telegram_user_id=user_id,
            username="testuser",
            first_name="Test",
            last_name="User"
        )
        print(f"   User created: {success}")

        # Test 3: Get user
        print("\n✅ Test 3: Get user")
        user = await get_user(user_id)
        if user:
            print(f"   User ID: {user['telegram_user_id']}")
            print(f"   Username: @{user['username']}")
            print(f"   Name: {user['first_name']} {user['last_name']}")
            print(f"   Linked: {user['oracle_username'] is not None}")
        else:
            print("   ❌ User not found")

        # Test 4: Check if user is linked
        print("\n✅ Test 4: Check if user is linked")
        linked = await is_user_linked(user_id)
        print(f"   User is linked: {linked}")

        # Test 5: Link user to Oracle
        print("\n✅ Test 5: Link user to Oracle")
        expires_at = datetime.now() + timedelta(hours=1)
        success = await link_user_to_oracle(
            telegram_user_id=user_id,
            oracle_username="TESTUSER",
            jwt_token="fake_jwt_token_for_testing",
            jwt_refresh_token="fake_refresh_token",
            token_expires_at=expires_at
        )
        print(f"   User linked: {success}")

        # Test 6: Verify link
        print("\n✅ Test 6: Verify link")
        linked = await is_user_linked(user_id)
        user = await get_user(user_id)
        print(f"   User is linked: {linked}")
        print(f"   Oracle username: {user['oracle_username']}")

        # Test 7: Create auth code
        print("\n✅ Test 7: Create authentication code")
        auth_code = "TEST1234"
        success = await create_auth_code(
            code=auth_code,
            telegram_user_id=user_id,
            oracle_username="TESTUSER"
        )
        print(f"   Auth code created: {success}")
        print(f"   Code: {auth_code}")

        # Test 8: Verify and use auth code
        print("\n✅ Test 8: Verify and use auth code")
        code_data = await verify_and_use_auth_code(auth_code)
        if code_data:
            print(f"   Code verified: True")
            print(f"   Code used: {code_data['used']}")
        else:
            print("   ❌ Code verification failed")

        # Test 9: Try to use same code again (should fail)
        print("\n✅ Test 9: Try to reuse code (should fail)")
        code_data = await verify_and_use_auth_code(auth_code)
        print(f"   Code reuse prevented: {code_data is None}")

        # Test 10: Create session
        print("\n✅ Test 10: Create conversation session")
        session_id = await create_session(
            telegram_user_id=user_id,
            conversation_state='{"context": "test"}'
        )
        print(f"   Session created: {session_id is not None}")
        if session_id:
            print(f"   Session ID: {session_id}")

        # Test 11: Get active session
        print("\n✅ Test 11: Get active session")
        session = await get_user_active_session(user_id)
        if session:
            print(f"   Active session found: True")
            print(f"   Session ID: {session['session_id']}")
            print(f"   Created: {session['created_at']}")
        else:
            print("   ❌ No active session found")

        # Test 12: Get database statistics
        print("\n✅ Test 12: Database statistics")
        stats = await get_database_stats()
        print(f"   Total users: {stats.get('total_users', 0)}")
        print(f"   Active users: {stats.get('active_users', 0)}")
        print(f"   Pending codes: {stats.get('pending_codes', 0)}")
        print(f"   Active sessions: {stats.get('active_sessions', 0)}")
        print(f"   Database size: {stats.get('db_size_mb', 0):.2f} MB")

        print("\n" + "=" * 60)
        print("✅ All tests completed successfully!")
        print("=" * 60)

    except Exception as e:
        print(f"\n❌ Test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

    return True


if __name__ == "__main__":
    success = asyncio.run(test_database())
    exit(0 if success else 1)