src/gaia/utils/credential_validator.py

"""
API Credential Validator

This module provides utilities for validating API credentials before running tests.
It checks if required API keys are set and validates them with minimal API calls
to ensure they are valid before running comprehensive tests.
"""

import os
import logging
import requests
from typing import Dict, Any, List, Optional

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


def validate_api_credentials(tools: List[str] = None) -> Dict[str, bool]:
    """
    Validate API credentials for the specified tools.
    
    Args:
        tools: List of tool names to validate (default: all tools)
        
    Returns:
        Dict[str, bool]: Dictionary with validation status for each tool
    """
    if not tools:
        tools = ["duckduckgo", "perplexity", "serper", "arxiv", "youtube", "reasoning"]
    
    validation_results = {tool: False for tool in tools}
    
    # Validate DuckDuckGo (no API key required, just check connectivity)
    if "duckduckgo" in tools:
        validation_results["duckduckgo"] = validate_duckduckgo()
    
    # Validate Perplexity
    if "perplexity" in tools:
        validation_results["perplexity"] = validate_perplexity()
    
    # Validate Serper
    if "serper" in tools:
        validation_results["serper"] = validate_serper()
    
    # Validate ArXiv (no API key required, just check connectivity)
    if "arxiv" in tools:
        validation_results["arxiv"] = validate_arxiv()
    
    # Validate YouTube (no API key required for basic usage)
    if "youtube" in tools:
        validation_results["youtube"] = validate_youtube()
    
    # Validate Reasoning (no API key required, check if class is importable)
    if "reasoning" in tools:
        validation_results["reasoning"] = validate_reasoning()
    
    return validation_results


def validate_duckduckgo() -> bool:
    """
    Validate DuckDuckGo connectivity.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    try:
        response = requests.get("https://api.duckduckgo.com/?q=test&format=json")
        return response.status_code == 200
    except Exception as e:
        logger.error(f"DuckDuckGo validation error: {str(e)}")
        return False


def validate_perplexity() -> bool:
    """
    Validate Perplexity API key.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    api_key = os.environ.get("PERPLEXITY_API_KEY")
    
    if not api_key:
        logger.error("PERPLEXITY_API_KEY environment variable not set")
        return False
    
    try:
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        # Make a minimal API call to validate the key
        response = requests.post(
            "https://api.perplexity.ai/chat/completions",
            headers=headers,
            json={
                "model": "sonar-small-chat",
                "messages": [{"role": "user", "content": "Hello"}],
                "max_tokens": 1
            }
        )
        
        if response.status_code == 200:
            return True
        else:
            logger.error(f"Perplexity API validation failed: Status {response.status_code}, {response.text}")
            return False
    
    except Exception as e:
        logger.error(f"Perplexity validation error: {str(e)}")
        return False


def validate_serper() -> bool:
    """
    Validate Serper API key.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    api_key = os.environ.get("SERPER_API_KEY")
    
    if not api_key:
        logger.error("SERPER_API_KEY environment variable not set")
        return False
    
    try:
        headers = {
            "X-API-KEY": api_key,
            "Content-Type": "application/json"
        }
        
        # Make a minimal API call to validate the key
        response = requests.post(
            "https://google.serper.dev/search",
            headers=headers,
            json={
                "q": "test",
                "num": 1
            }
        )
        
        if response.status_code == 200:
            return True
        else:
            logger.error(f"Serper API validation failed: Status {response.status_code}, {response.text}")
            return False
    
    except Exception as e:
        logger.error(f"Serper validation error: {str(e)}")
        return False


def validate_arxiv() -> bool:
    """
    Validate ArXiv connectivity.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    try:
        response = requests.get("http://export.arxiv.org/api/query?search_query=test&start=0&max_results=1")
        return response.status_code == 200
    except Exception as e:
        logger.error(f"ArXiv validation error: {str(e)}")
        return False


def validate_youtube() -> bool:
    """
    Validate YouTube connectivity.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    try:
        # Try to access a known YouTube video page
        response = requests.get("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
        return response.status_code == 200
    except Exception as e:
        logger.error(f"YouTube validation error: {str(e)}")
        return False


def validate_reasoning() -> bool:
    """
    Validate Reasoning tool importability.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    try:
        # Try to import the ReasoningTool class
        from src.gaia.tools.reasoning_tools import ReasoningTool
        return True
    except ImportError as e:
        logger.error(f"Reasoning tool import error: {str(e)}")
        return False
    except Exception as e:
        logger.error(f"Reasoning tool validation error: {str(e)}")
        return False


def validate_supabase_connection() -> bool:
    """
    Validate Supabase connection for memory integration tests.
    
    Returns:
        bool: True if validation is successful, False otherwise
    """
    try:
        # Try to import Supabase components
        from src.gaia.memory.supabase_memory import SupabaseMemory
        from src.gaia.agent.config import get_memory_config
        
        # Get memory configuration
        memory_config = get_memory_config()
        memory_config["enabled"] = True
        
        # Initialize Supabase memory
        memory = SupabaseMemory(memory_config)
        
        # Check if initialized
        return memory.initialized
    
    except ImportError as e:
        logger.error(f"Supabase component import error: {str(e)}")
        return False
    except Exception as e:
        logger.error(f"Supabase connection validation error: {str(e)}")
        return False


if __name__ == "__main__":
    # Run credential validation for all tools
    print("\n=== API Credential Validation ===\n")
    
    results = validate_api_credentials()
    
    print("\nValidation Results:\n")
    for tool, is_valid in results.items():
        status = "✅ VALID" if is_valid else "❌ INVALID"
        print(f"{tool.capitalize():15} : {status}")
    
    # Also check Supabase connection
    supabase_valid = validate_supabase_connection()
    status = "✅ VALID" if supabase_valid else "❌ INVALID"
    print(f"\nSupabase Memory:  {status}")
    
    # Summary
    total = len(results) + 1  # +1 for Supabase
    valid_count = sum(1 for is_valid in results.values() if is_valid) + (1 if supabase_valid else 0)
    
    print(f"\n{valid_count} of {total} validations succeeded")
    
    if valid_count < total:
        print("\nPlease check the error messages above and configure the missing credentials.")
        print("For details on required API keys, see the project documentation.")