Spaces:
Sleeping
Sleeping
File size: 2,250 Bytes
65eb048 8cc8e89 65eb048 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | #!/usr/bin/env python3
"""
Test the update_notifier utility for document updates
"""
import os
import sys
import asyncio
from dotenv import load_dotenv
# Add parent directory to path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from utils.utils_fn import push_document_update
# Load environment variables
load_dotenv(dotenv_path=".env", override=False)
async def test_update_notifier():
"""Test the document update notification functionality"""
print("=" * 80)
print("🧪 TESTING DOCUMENT UPDATE NOTIFIER")
print("=" * 80)
# Check environment configuration
base_url = os.getenv("SUPABASE_BASE_URL")
api_key = os.getenv("CYBERLGL_API_KEY")
print(f"\n📋 Configuration:")
print(f" SUPABASE_BASE_URL: {base_url}")
print(f" CYBERLGL_API_KEY: {'✅ Set' if api_key else '❌ Missing'}")
if not base_url or not api_key:
print("\n⚠️ Skipping test - environment variables not configured")
print(" Required: SUPABASE_BASE_URL and CYBERLGL_API_KEY")
return
# Test parameters
test_document_id = "test-doc-123"
test_user_id = "test-user-456"
test_content = "<html><body><h1>Test Document</h1><p>This is a test update.</p></body></html>"
print(f"\n📤 Test parameters:")
print(f" Document ID: {test_document_id}")
print(f" User ID: {test_user_id}")
print(f" Content size: {len(test_content)} bytes")
# Test the push function
print(f"\n🚀 Testing push_document_update...")
try:
success = await push_document_update(
document_id=test_document_id,
content=test_content,
user_id=test_user_id
)
if success:
print("\n✅ Test PASSED - Document update pushed successfully")
else:
print("\n⚠️ Test WARNING - Push returned False (check logs)")
except Exception as e:
print(f"\n❌ Test FAILED - Exception occurred: {str(e)}")
import traceback
traceback.print_exc()
print("\n" + "=" * 80)
print("🎯 TEST COMPLETED")
print("=" * 80)
if __name__ == "__main__":
asyncio.run(test_update_notifier()) |