114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
"""
|
|
Test script for Box automation endpoints.
|
|
The Box processor is now integrated into app.py on port 5000.
|
|
|
|
Usage:
|
|
# Start the main server first
|
|
python app.py
|
|
|
|
# Then run tests
|
|
python test_box_processor.py
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
|
|
BASE_URL = "http://localhost:5000"
|
|
|
|
|
|
def test_health():
|
|
"""Test Box automation health endpoint"""
|
|
print("\n" + "=" * 70)
|
|
print("TEST: Box Automation Health")
|
|
print("=" * 70)
|
|
|
|
try:
|
|
response = requests.get(f"{BASE_URL}/api/box/health")
|
|
print(f"Status: {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('box_initialised'):
|
|
print("✓ Box is initialised and ready")
|
|
else:
|
|
print("⚠ Box is not initialised — check BOX_VIDEO_OPTIMIZER_FOLDER_ID in .env")
|
|
return True
|
|
return False
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
print("✗ Connection refused — is app.py running on port 5000?")
|
|
print(" Start with: python backend/app.py")
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ Error: {e}")
|
|
return False
|
|
|
|
|
|
def test_manual_trigger(file_id: str, filename: str):
|
|
"""Test manual processing trigger with a real Box file ID"""
|
|
print("\n" + "=" * 70)
|
|
print(f"TEST: Manual Trigger — {filename}")
|
|
print("=" * 70)
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{BASE_URL}/api/box/trigger",
|
|
json={'file_id': file_id, 'filename': filename},
|
|
timeout=300 # 5 min — allow for full download + FFmpeg conversion
|
|
)
|
|
|
|
print(f"Status: {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
status = result.get('status')
|
|
if status == 'success':
|
|
print("✓ Processing successful")
|
|
return True
|
|
elif status == 'skipped':
|
|
print(f"⚠ Skipped — invalid filename: {result.get('error')}")
|
|
return True
|
|
else:
|
|
print(f"✗ Failed: {result.get('error')}")
|
|
return False
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
print("✗ Request timed out — processing may still be running on server")
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ Error: {e}")
|
|
return False
|
|
|
|
|
|
def run_tests():
|
|
print("\n" + "=" * 70)
|
|
print("BOX AUTOMATION — TEST SUITE")
|
|
print(f"Target: {BASE_URL}")
|
|
print("=" * 70)
|
|
|
|
if not test_health():
|
|
print("\n✗ Health check failed — stopping tests")
|
|
return
|
|
|
|
print("\n📝 To test with a real Box file:")
|
|
print(" 1. Upload a video to the Box IN folder")
|
|
print(" 2. Get the file ID from the Box URL or Box admin")
|
|
print(" 3. Uncomment and update the test below, then re-run")
|
|
print()
|
|
print(" Example valid filename: my_campaign_tiktok_9x16.mp4")
|
|
print(" Example invalid: my_campaign_no_platform.mp4")
|
|
|
|
# Uncomment to test with a real Box file ID:
|
|
# test_manual_trigger("YOUR_BOX_FILE_ID", "my_campaign_tiktok_9x16.mp4")
|
|
# test_manual_trigger("YOUR_BOX_FILE_ID", "invalid_no_platform.mp4") # tests error path
|
|
|
|
print("\n" + "=" * 70)
|
|
print("DONE")
|
|
print("=" * 70)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run_tests()
|