Add comprehensive test scripts for thumbnail generation and xAI collections API
- 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.
This commit is contained in:
bsiggel
253
tests/test_thumbnail_generation.py
Normal file
253
tests/test_thumbnail_generation.py
Normal file
@@ -0,0 +1,253 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user