Spaces:
Sleeping
Sleeping
File size: 5,716 Bytes
54fe70d |
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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 |
"""
Test script for Cloudinary-only image storage in the Marine Guard API.
This script ensures that:
1. We can authenticate with the API
2. We can upload an image which gets stored in Cloudinary
3. We can retrieve the incident which contains the Cloudinary URL
"""
import os
import requests
import json
from pathlib import Path
import random
import time
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# API endpoint for incident classification
API_URL = "http://localhost:8000/api/incidents/classify"
API_LIST_URL = "http://localhost:8000/api/incidents/list"
API_LOGIN_URL = "http://localhost:8000/api/auth/login"
def authenticate():
"""Authenticate with the API and get a JWT token"""
try:
login_data = {
"email": "testuser@example.com",
"password": "Password123!"
}
print(f"Authenticating with the API at {API_LOGIN_URL}...")
response = requests.post(API_LOGIN_URL, json=login_data)
if response.status_code == 200:
token = response.json().get("access_token")
print("✅ Authentication successful!")
return token
else:
print(f"❌ Authentication failed with status code {response.status_code}")
print(response.text)
return None
except Exception as e:
print(f"❌ Error during authentication: {e}")
return None
def create_test_file():
"""Create a test file to upload"""
test_dir = Path("test_files")
test_dir.mkdir(exist_ok=True)
# Use a .txt extension since we just need any file that Cloudinary can handle
file_path = test_dir / f"cloudinary_test_{random.randint(1000, 9999)}.txt"
# Write simple text content
with open(file_path, "w") as f:
f.write("This is a test file for Cloudinary upload in Marine Guard API")
print(f"Created test file at: {file_path}")
return file_path
def report_incident():
"""Report an incident with an image via the API"""
token = authenticate()
if not token:
print("Cannot proceed with test - authentication failed")
return False
# Create a test image file
image_path = create_test_file()
try:
# Prepare the form data
form_data = {
"description": "Cloudinary-only storage test with JPG image",
"name": "Cloudinary Storage Test",
"latitude": "42.3601",
"longitude": "-71.0589"
}
# Prepare the file for upload
with open(image_path, "rb") as image_file:
files = {"image": (image_path.name, image_file, "text/plain")}
# Make the API request
print(f"Reporting incident to {API_URL}...")
response = requests.post(
API_URL,
data=form_data,
files=files,
headers={"Authorization": f"Bearer {token}"}
)
# Check the response
if response.status_code == 200:
result = response.json()
incident_id = result.get("incident_id")
print(f"✅ Incident reported successfully! ID: {incident_id}")
# Verify that the image was uploaded to Cloudinary
return verify_cloudinary_upload(token, incident_id)
else:
print(f"❌ Incident reporting failed with status code {response.status_code}")
print(f"Error details: {response.text}")
return False
except Exception as e:
print(f"❌ Error reporting incident: {e}")
return False
finally:
# Clean up the test file
if os.path.exists(image_path):
os.unlink(image_path)
print(f"Deleted test file: {image_path}")
def verify_cloudinary_upload(token, incident_id):
"""Verify that the incident has a Cloudinary URL for its image"""
try:
print(f"Fetching incident details to verify Cloudinary URL...")
# Short delay to ensure the incident is saved in the database
time.sleep(1)
response = requests.get(
API_LIST_URL,
headers={"Authorization": f"Bearer {token}"}
)
if response.status_code == 200:
incidents = response.json().get("incidents", [])
for incident in incidents:
if incident.get("id") == incident_id:
image_path = incident.get("image_path")
if image_path and "cloudinary.com" in image_path:
print(f"✅ SUCCESS: Image was uploaded to Cloudinary!")
print(f"Cloudinary URL: {image_path}")
return True
else:
print(f"❌ FAILED: Image does not have a Cloudinary URL")
print(f"Image path: {image_path}")
return False
print(f"❌ FAILED: Could not find incident with ID {incident_id}")
return False
else:
print(f"❌ FAILED: Could not fetch incidents list. Status: {response.status_code}")
print(response.text)
return False
except Exception as e:
print(f"❌ Error verifying Cloudinary upload: {e}")
return False
if __name__ == "__main__":
print("=" * 60)
print("MARINE GUARD API - CLOUDINARY-ONLY STORAGE TEST")
print("=" * 60)
success = report_incident()
print("\nTEST SUMMARY")
print("=" * 60)
print(f"Cloudinary storage test: {'✅ PASSED' if success else '❌ FAILED'}")
print("=" * 60) |