bcb6454b2a
- Implemented `test_thumbnail_generation.py` to validate the complete flow of document thumbnail generation in EspoCRM, including document creation, file upload, webhook triggering, and preview verification. - Created `test_xai_collections_api.py` to test critical operations of the xAI Collections API, covering file uploads, collection CRUD operations, document management, and response validation. - Both scripts include detailed logging for success and error states, ensuring robust testing and easier debugging.
254 lines
8.3 KiB
Python
254 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test script for Document Thumbnail Generation
|
||
Tests the complete flow:
|
||
1. Create a test document in EspoCRM
|
||
2. Upload a file attachment
|
||
3. Trigger the webhook (or wait for automatic trigger)
|
||
4. Verify preview generation
|
||
"""
|
||
|
||
import asyncio
|
||
import aiohttp
|
||
import os
|
||
import sys
|
||
import json
|
||
from pathlib import Path
|
||
from io import BytesIO
|
||
from PIL import Image
|
||
|
||
# Add bitbylaw to path
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
||
from services.espocrm import EspoCRMAPI
|
||
|
||
|
||
async def create_test_image(width: int = 800, height: int = 600) -> bytes:
|
||
"""Create a simple test PNG image"""
|
||
img = Image.new('RGB', (width, height), color='lightblue')
|
||
|
||
# Add some text/pattern so it's not just a solid color
|
||
from PIL import ImageDraw, ImageFont
|
||
draw = ImageDraw.Draw(img)
|
||
|
||
# Draw some shapes
|
||
draw.rectangle([50, 50, width-50, height-50], outline='darkblue', width=5)
|
||
draw.ellipse([width//4, height//4, 3*width//4, 3*height//4], outline='red', width=3)
|
||
|
||
# Add text
|
||
try:
|
||
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 48)
|
||
except:
|
||
font = None
|
||
|
||
text = "TEST IMAGE\nFor Thumbnail\nGeneration"
|
||
draw.text((width//2, height//2), text, fill='black', anchor='mm', font=font, align='center')
|
||
|
||
# Save to bytes
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
return buffer.getvalue()
|
||
|
||
|
||
async def create_test_document(espocrm: EspoCRMAPI) -> str:
|
||
"""Create a test document in EspoCRM"""
|
||
print("\n📄 Creating test document in EspoCRM...")
|
||
|
||
document_data = {
|
||
"name": f"Test Thumbnail Generation {asyncio.get_event_loop().time()}",
|
||
"status": "Active",
|
||
"dateiStatus": "Neu", # This should trigger preview generation
|
||
"type": "Image",
|
||
"description": "Automated test document for thumbnail generation"
|
||
}
|
||
|
||
result = await espocrm.create_entity("Document", document_data)
|
||
doc_id = result.get("id")
|
||
|
||
print(f"✅ Document created: {doc_id}")
|
||
print(f" Name: {result.get('name')}")
|
||
print(f" Datei-Status: {result.get('dateiStatus')}")
|
||
|
||
return doc_id
|
||
|
||
|
||
async def upload_test_file(espocrm: EspoCRMAPI, doc_id: str) -> str:
|
||
"""Upload a test image file to the document"""
|
||
print(f"\n📤 Uploading test image to document {doc_id}...")
|
||
|
||
# Create test image
|
||
image_data = await create_test_image(1200, 900)
|
||
print(f" Generated test image: {len(image_data)} bytes")
|
||
|
||
# Upload to EspoCRM
|
||
attachment = await espocrm.upload_attachment(
|
||
file_content=image_data,
|
||
filename="test_image.png",
|
||
parent_type="Document",
|
||
parent_id=doc_id,
|
||
field="file",
|
||
mime_type="image/png",
|
||
role="Attachment"
|
||
)
|
||
|
||
attachment_id = attachment.get("id")
|
||
print(f"✅ File uploaded: {attachment_id}")
|
||
print(f" Filename: {attachment.get('name')}")
|
||
print(f" Size: {attachment.get('size')} bytes")
|
||
|
||
return attachment_id
|
||
|
||
|
||
async def trigger_webhook(doc_id: str, action: str = "update"):
|
||
"""Manually trigger the document webhook"""
|
||
print(f"\n🔔 Triggering webhook for document {doc_id}...")
|
||
|
||
webhook_url = f"http://localhost:7777/vmh/webhook/document/{action}"
|
||
payload = {
|
||
"entityType": "Document",
|
||
"entity": {
|
||
"id": doc_id,
|
||
"entityType": "Document"
|
||
},
|
||
"data": {
|
||
"entity": {
|
||
"id": doc_id
|
||
}
|
||
}
|
||
}
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(webhook_url, json=payload) as response:
|
||
status = response.status
|
||
text = await response.text()
|
||
|
||
if status == 200:
|
||
print(f"✅ Webhook triggered successfully")
|
||
print(f" Response: {text}")
|
||
else:
|
||
print(f"❌ Webhook failed: {status}")
|
||
print(f" Response: {text}")
|
||
|
||
return status == 200
|
||
|
||
|
||
async def check_preview_generated(espocrm: EspoCRMAPI, doc_id: str, max_wait: int = 30):
|
||
"""Check if preview was generated (poll for a few seconds)"""
|
||
print(f"\n🔍 Checking for preview generation (max {max_wait}s)...")
|
||
|
||
for i in range(max_wait):
|
||
await asyncio.sleep(1)
|
||
|
||
# Get document
|
||
doc = await espocrm.get_entity("Document", doc_id)
|
||
|
||
# Check if preview field is populated
|
||
preview_id = doc.get("previewId")
|
||
if preview_id:
|
||
print(f"\n✅ Preview generated!")
|
||
print(f" Preview Attachment ID: {preview_id}")
|
||
print(f" Preview Name: {doc.get('previewName')}")
|
||
print(f" Preview Type: {doc.get('previewType')}")
|
||
|
||
# Try to download and check the preview
|
||
try:
|
||
preview_data = await espocrm.download_attachment(preview_id)
|
||
print(f" Preview Size: {len(preview_data)} bytes")
|
||
|
||
# Verify it's a WebP image
|
||
from PIL import Image
|
||
img = Image.open(BytesIO(preview_data))
|
||
print(f" Preview Format: {img.format}")
|
||
print(f" Preview Dimensions: {img.width}x{img.height}")
|
||
|
||
if img.format == "WEBP":
|
||
print(" ✅ Format is WebP as expected")
|
||
if img.width <= 600 and img.height <= 800:
|
||
print(" ✅ Dimensions within expected range")
|
||
|
||
except Exception as e:
|
||
print(f" ⚠️ Could not verify preview: {e}")
|
||
|
||
return True
|
||
|
||
if (i + 1) % 5 == 0:
|
||
print(f" Still waiting... ({i + 1}s)")
|
||
|
||
print(f"\n❌ Preview not generated after {max_wait}s")
|
||
return False
|
||
|
||
|
||
async def cleanup_test_document(espocrm: EspoCRMAPI, doc_id: str):
|
||
"""Delete the test document"""
|
||
print(f"\n🗑️ Cleaning up test document {doc_id}...")
|
||
try:
|
||
await espocrm.delete_entity("Document", doc_id)
|
||
print("✅ Test document deleted")
|
||
except Exception as e:
|
||
print(f"⚠️ Could not delete test document: {e}")
|
||
|
||
|
||
async def main():
|
||
print("=" * 80)
|
||
print("THUMBNAIL GENERATION TEST")
|
||
print("=" * 80)
|
||
|
||
# Initialize EspoCRM API
|
||
espocrm = EspoCRMAPI()
|
||
|
||
doc_id = None
|
||
try:
|
||
# Step 1: Create test document
|
||
doc_id = await create_test_document(espocrm)
|
||
|
||
# Step 2: Upload test file
|
||
attachment_id = await upload_test_file(espocrm, doc_id)
|
||
|
||
# Step 3: Update document to trigger webhook (set dateiStatus to trigger sync)
|
||
print(f"\n🔄 Updating document to trigger webhook...")
|
||
await espocrm.update_entity("Document", doc_id, {
|
||
"dateiStatus": "Neu" # This should trigger the webhook
|
||
})
|
||
print("✅ Document updated")
|
||
|
||
# Step 4: Wait a bit for webhook to be processed
|
||
print("\n⏳ Waiting 3 seconds for webhook processing...")
|
||
await asyncio.sleep(3)
|
||
|
||
# Step 5: Check if preview was generated
|
||
success = await check_preview_generated(espocrm, doc_id, max_wait=20)
|
||
|
||
# Summary
|
||
print("\n" + "=" * 80)
|
||
if success:
|
||
print("✅ TEST PASSED - Preview generation successful!")
|
||
else:
|
||
print("❌ TEST FAILED - Preview was not generated")
|
||
print("\nCheck logs with:")
|
||
print(" sudo journalctl -u motia.service --since '2 minutes ago' | grep -E '(PREVIEW|Document)'")
|
||
print("=" * 80)
|
||
|
||
# Ask if we should clean up
|
||
print(f"\nTest document ID: {doc_id}")
|
||
cleanup = input("\nDelete test document? (y/N): ").strip().lower()
|
||
if cleanup == 'y':
|
||
await cleanup_test_document(espocrm, doc_id)
|
||
else:
|
||
print(f"ℹ️ Test document kept: {doc_id}")
|
||
print(f" View in EspoCRM: https://crm.bitbylaw.com/#Document/view/{doc_id}")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ Test failed with error: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if doc_id:
|
||
print(f"\nTest document ID: {doc_id}")
|
||
cleanup = input("\nDelete test document? (y/N): ").strip().lower()
|
||
if cleanup == 'y':
|
||
await cleanup_test_document(espocrm, doc_id)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|