SPARC/test_database_mode.py

#!/usr/bin/env python3
"""Test script to verify database caching functionality.

This script tests the LLMAnalyzer with database caching without requiring
actual API keys or patent downloads.
"""

from SPARC.llm import LLMAnalyzer
from SPARC.database import DatabaseClient
from SPARC import config

def test_database_storage():
    """Test that messages are always stored in database."""
    print("Testing Database Storage & Caching")
    print("=" * 70)

    # Initialize analyzer (database is always used)
    print("\n1. Initializing LLMAnalyzer...")
    analyzer = LLMAnalyzer(use_cache=True)

    print(f"   - use_cache: {analyzer.use_cache}")
    print(f"   - db_client: {analyzer.db_client is not None}")
    print(f"   - client (API): {analyzer.client is not None}")

    # Test single patent analysis (without API key, stores placeholder)
    print("\n2. Testing single patent analysis (no API key)...")
    result = analyzer.analyze_patent_content(
        patent_content="Test patent content about semiconductor innovation",
        company_name="TestCorp"
    )
    print(f"   Result: {result[:80]}...")

    # Test portfolio analysis
    print("\n3. Testing portfolio analysis (no API key)...")
    test_patents = [
        {"patent_id": "US001", "content": "First test patent"},
        {"patent_id": "US002", "content": "Second test patent"},
    ]
    result = analyzer.analyze_patent_portfolio(
        patents_data=test_patents,
        company_name="TestCorp"
    )
    print(f"   Result: {result[:80]}...")

    # Verify messages were stored
    print("\n4. Verifying messages were stored...")
    db_client = DatabaseClient(config.database_url)
    messages = db_client.get_messages(company_name="TestCorp", limit=10)
    print(f"   Found {len(messages)} stored messages")

    for msg in messages:
        cached_status = "CACHED" if msg.get('is_cached') else "NEW"
        print(f"   - ID: {msg['id']}, Type: {msg['analysis_type']}, Status: {cached_status}")

    # Get analytics
    print("\n5. Getting analytics...")
    analytics = db_client.get_analytics(days=1)
    print(f"   Total messages: {analytics['total_messages']}")
    print(f"   By company: {analytics['by_company']}")
    print(f"   By type: {analytics['by_type']}")

    print("\n" + "=" * 70)
    print("Database storage test completed successfully!")

def test_caching():
    """Test that caching works correctly."""
    print("\nTesting Cache Functionality")
    print("=" * 70)

    db_client = DatabaseClient(config.database_url)
    db_client.initialize_schema()

    # Store a fake cached response
    print("\n1. Storing a test response in database...")
    test_prompt = "Test prompt for caching"
    test_response = "This is a cached response from previous API call"

    db_client.store_message(
        prompt=test_prompt,
        response=test_response,
        company_name="CacheTest",
        analysis_type="test",
        model="test-model"
    )

    # Try to retrieve from cache
    print("\n2. Testing cache retrieval...")
    cached = db_client.get_cached_response(
        prompt=test_prompt,
        company_name="CacheTest",
        analysis_type="test"
    )

    if cached:
        print(f"   Cache hit! Response: {cached['response']}")
    else:
        print("   Cache miss (unexpected)")

    # Test cache miss
    print("\n3. Testing cache miss...")
    cached = db_client.get_cached_response(
        prompt="Different prompt",
        company_name="CacheTest",
        analysis_type="test"
    )

    if cached:
        print("   Unexpected cache hit")
    else:
        print("   Cache miss as expected")

    print("\n" + "=" * 70)
    print("Cache test completed successfully!")

def test_test_mode():
    """Test that test mode works correctly."""
    print("\nTesting Test Mode")
    print("=" * 70)

    print("\n1. Initializing LLMAnalyzer in test mode...")
    analyzer = LLMAnalyzer(test_mode=True)

    print(f"   - test_mode: {analyzer.test_mode}")
    print(f"   - db_client: {analyzer.db_client is not None}")

    print("\n2. Testing single patent analysis (test mode)...")
    result = analyzer.analyze_patent_content(
        patent_content="Test patent content",
        company_name="TestCorp2"
    )
    print(f"   Result: {result}")

    print("\n" + "=" * 70)
    print("Test mode test completed successfully!")

if __name__ == "__main__":
    test_database_storage()
    print("\n")
    test_caching()
    print("\n")
    test_test_mode()