Spaces:
Running
Running
| #!/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()) |