stream-bot/tests/test_token_cache.py

"""
Test script to verify token caching functionality
"""
import os
import sys
import logging
import time
import json
from dotenv import load_dotenv

# Add the parent directory to Python's path so 'src' module can be found
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

# Load environment variables from .env file
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger('token-cache-test')

def test_token_cache(cache_file: str = "test_token_cache.json") -> bool:
    """Test the token caching functionality"""
    try:
        import requests
    except ImportError:
        logger.error("The 'requests' library is required. Install it with: pip install requests")
        return False
    
    # Import the bot class
    from src.core.twitch import TwitchBot
    from src.core.auth import TwitchAuth
    
    # Get credentials from environment variables
    client_id = os.environ.get("TWITCH_CLIENT_ID")
    client_secret = os.environ.get("TWITCH_CLIENT_SECRET")
    username = os.environ.get("TWITCH_USERNAME")
    channel = os.environ.get("TWITCH_CHANNEL")
    
    if not all([client_id, client_secret, username, channel]):
        logger.error("Missing required environment variables")
        logger.info("Please set TWITCH_CLIENT_ID, TWITCH_CLIENT_SECRET, TWITCH_USERNAME, and TWITCH_CHANNEL")
        return False
    
    # Remove the cache file if it exists (to start fresh)
    if os.path.exists(cache_file):
        logger.info(f"Removing existing cache file: {cache_file}")
        os.remove(cache_file)
    
    # Use mock tokens for testing
    mock_access_token = "mock_access_token_for_testing"
    mock_refresh_token = "mock_refresh_token_for_testing"
    mock_expires_in = 14400  # 4 hours
    
    # Create a bot instance with the test cache file
    logger.info("Creating bot instance...")
    bot = TwitchBot(
        username=username,
        client_id=client_id,
        client_secret=client_secret,
        channel=channel,
        access_token=mock_access_token,
        refresh_token=mock_refresh_token,
        token_cache_file=cache_file
    )
    
    # First call should use the provided token and save it
    logger.info("First call to get_oauth_token (should use the provided token)...")
    oauth_token1 = bot.get_oauth_token()
    
    # Verify the cache file was created
    if not os.path.exists(cache_file):
        logger.error(f"Cache file was not created: {cache_file}")
        return False
    
    # Load and display cache file
    with open(cache_file, 'r') as f:
        cache_data = json.load(f)
    
    logger.info(f"Cache file created with {len(cache_data)} entries")
    logger.info(f"Token expires at: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(cache_data['expires_at']))}")
    
    # Save the token for comparison
    token1 = bot.access_token
    
    # Create a new bot instance to test loading from cache
    logger.info("Creating a second bot instance (should load token from cache)...")
    bot2 = TwitchBot(
        username=username,
        client_id=client_id,
        client_secret=client_secret,
        channel=channel,
        token_cache_file=cache_file
    )
    
    # Second call should use the cached token
    logger.info("Second call to get_oauth_token (should use cached token)...")
    oauth_token2 = bot2.get_oauth_token()
    token2 = bot2.access_token
    
    # Compare tokens
    if token1 == token2:
        logger.info("SUCCESS: Both instances used the same token")
    else:
        logger.error("FAILURE: Different tokens were used")
        return False
    
    # Simulate token expiration by modifying the cache file
    logger.info("Simulating token expiration...")
    with open(cache_file, 'r') as f:
        cache_data = json.load(f)
    
    # Set expiry to now minus 10 minutes
    cache_data['expires_at'] = time.time() - 600
    
    with open(cache_file, 'w') as f:
        json.dump(cache_data, f)
    
    # Create a third bot instance with the "expired" token
    # Since we don't want to trigger a real OAuth flow in tests,
    # we'll provide a new mock token to use
    logger.info("Creating a third bot instance with expired token cache...")
    mock_new_token = "mock_new_token_after_expiration"
    mock_new_refresh = "mock_new_refresh_after_expiration"
    
    bot3 = TwitchBot(
        username=username,
        client_id=client_id,
        client_secret=client_secret,
        channel=channel,
        access_token=mock_new_token,
        refresh_token=mock_new_refresh,
        token_cache_file=cache_file
    )
    
    # Third call should use the new token since the cached one is expired
    logger.info("Third call to get_oauth_token (should use the new mock token due to expiration)...")
    oauth_token3 = bot3.get_oauth_token()
    token3 = bot3.access_token
    
    # Compare tokens - they should be different
    if token1 != token3:
        logger.info("SUCCESS: Token was refreshed after expiration")
    else:
        logger.error("FAILURE: Expired token was not refreshed")
        return False
    
    # Clean up
    if os.path.exists(cache_file):
        os.remove(cache_file)
        logger.info(f"Removed test cache file: {cache_file}")
    
    logger.info("All token cache tests passed!")
    return True

if __name__ == "__main__":
    print("Testing token caching functionality...")
    if test_token_cache():
        print("All token cache tests PASSED!")
        sys.exit(0)
    else:
        print("Token cache tests FAILED!")
        sys.exit(1)