test_real_cache.py

#!/usr/bin/env python3
"""Test script that uses the actual Granola cache file."""

import asyncio
import json
import os
from pathlib import Path

from granola_mcp_server.server import GranolaMCPServer


async def test_real_cache():
    """Test the server with real Granola cache if available."""
    cache_path = os.path.expanduser("~/Library/Application Support/Granola/cache-v3.json")
    
    print(f"Looking for Granola cache at: {cache_path}")
    
    # Check if cache file exists
    if not Path(cache_path).exists():
        print("❌ No Granola cache file found")
        print("   This means either:")
        print("   - Granola.ai is not installed")
        print("   - No meetings have been recorded yet")
        print("   - Cache is stored in a different location")
        return False
    
    # Check if cache file is readable
    try:
        with open(cache_path, 'r', encoding='utf-8') as f:
            cache_size = len(f.read())
        print(f"✅ Cache file found! Size: {cache_size} bytes")
    except PermissionError:
        print("❌ Cache file exists but is not readable (permission denied)")
        return False
    except Exception as e:
        print(f"❌ Error reading cache file: {e}")
        return False
    
    try:
        print("\n" + "="*60)
        print("TESTING WITH REAL GRANOLA CACHE")
        print("="*60)
        
        # Initialize server with real cache
        server = GranolaMCPServer(cache_path=cache_path)
        
        print("Loading real cache data...")
        await server._load_cache()
        
        if not server.cache_data:
            print("❌ Failed to load cache data")
            return False
        
        # Show what we loaded
        meeting_count = len(server.cache_data.meetings)
        doc_count = len(server.cache_data.documents)
        transcript_count = len(server.cache_data.transcripts)
        
        print(f"✅ Successfully loaded real cache data:")
        print(f"   📅 {meeting_count} meetings")
        print(f"   📄 {doc_count} documents")
        print(f"   🗣️ {transcript_count} transcripts")
        
        if meeting_count == 0:
            print("\n⚠️  No meetings found in cache - Granola may not have processed any meetings yet")
            return True
        
        print(f"\n📊 CACHE ANALYSIS:")
        if server.cache_data.last_updated:
            print(f"   Last updated: {server.cache_data.last_updated}")
        
        # Show sample meeting titles
        sample_meetings = list(server.cache_data.meetings.values())[:3]
        if sample_meetings:
            print(f"\n📅 SAMPLE MEETINGS:")
            for i, meeting in enumerate(sample_meetings, 1):
                print(f"   {i}. {meeting.title}")
                print(f"      Date: {meeting.date.strftime('%Y-%m-%d %H:%M')}")
                if meeting.participants:
                    print(f"      Participants: {', '.join(meeting.participants)}")
                print()
        
        print("="*60)
        print("TESTING SERVER FUNCTIONALITY WITH REAL DATA")
        print("="*60)
        
        # Test search functionality
        print("\n🔍 Testing search with 'meeting'...")
        results = await server._search_meetings("meeting", 3)
        print("Search results:", results[0].text[:200] + "..." if results[0].text else "No results")
        
        # Test getting details of first meeting
        if sample_meetings:
            first_meeting_id = sample_meetings[0].id
            print(f"\n📋 Testing meeting details for ID: {first_meeting_id}...")
            details = await server._get_meeting_details(first_meeting_id)
            print("Meeting details:", details[0].text[:300] + "..." if details[0].text else "No details")
            
            # Test transcript if available
            if first_meeting_id in server.cache_data.transcripts:
                print(f"\n🗣️ Testing transcript for meeting: {first_meeting_id}...")
                transcript = await server._get_meeting_transcript(first_meeting_id)
                print("Transcript preview:", transcript[0].text[:200] + "..." if transcript[0].text else "No transcript")
        
        # Validate document panel content when present
        try:
            with open(cache_path, 'r', encoding='utf-8') as cache_file:
                cache_json = json.load(cache_file)
            state = json.loads(cache_json.get('cache', '{}')).get('state', {})
        except Exception as exc:
            print(f"\n⚠️ Unable to parse raw cache for panel validation: {exc}")
            state = {}

        panels = state.get('documentPanels', {})
        meetings_meta = state.get('meetingsMetadata', {})

        checked = skipped = failures = 0
        recent_meetings = sorted(
            meetings_meta.items(),
            key=lambda item: item[1].get('created_at', ''),
            reverse=True
        )[:10]

        for meeting_id, _meta in recent_meetings:
            panel_data = panels.get(meeting_id)
            if not panel_data:
                skipped += 1
                continue

            if meeting_id not in server.cache_data.documents:
                skipped += 1
                continue

            docs = await server._get_meeting_documents(meeting_id)
            document_text = docs[0].text if docs else ''

            if document_text and document_text.strip():
                checked += 1
            else:
                print(f"\n❌ Panel-backed meeting '{meeting_id}' produced empty document content")
                failures += 1

        print(f"\n📄 Panel validation summary: checked={checked}, skipped={skipped}, failures={failures}")

        if failures:
            raise AssertionError("Some meetings with document panels yielded empty content")

        # Test pattern analysis
        print(f"\n📊 Testing participant pattern analysis...")
        patterns = await server._analyze_meeting_patterns("participants")
        print("Pattern analysis:", patterns[0].text[:300] + "..." if patterns[0].text else "No patterns")
        
        print("\n" + "="*60)
        print("✅ ALL TESTS PASSED WITH REAL GRANOLA DATA!")
        print("="*60)
        return True
        
    except Exception as e:
        print(f"❌ Test failed with error: {e}")
        import traceback
        traceback.print_exc()
        return False


async def main():
    """Main test function."""
    print("🧪 GRANOLA MCP SERVER - REAL CACHE TEST")
    print("="*60)
    
    success = await test_real_cache()
    
    if success:
        print("\n🎉 SUCCESS: Server works perfectly with real Granola data!")
    else:
        print("\n⚠️  Note: Server will still work fine, just with empty data")
        print("   To get real meeting data:")
        print("   1. Install Granola.ai")
        print("   2. Record some meetings")
        print("   3. Run this test again")


if __name__ == "__main__":
    asyncio.run(main())