Spaces:
Sleeping
Sleeping
| import os | |
| import requests | |
| import json | |
| import sys | |
| from pathlib import Path | |
| from dotenv import load_dotenv | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| # API endpoint for incident classification | |
| API_URL = "http://localhost:8000/api/incidents/classify" | |
| # JWT token (replace with your actual token) | |
| JWT_TOKEN = "" # Set this to test with your token | |
| def get_token(): | |
| """Get a JWT token by authenticating with the API""" | |
| try: | |
| login_url = "http://localhost:8000/api/auth/login" | |
| login_data = { | |
| "email": "testuser@example.com", # Using our newly created test user | |
| "password": "Password123!" # Using the password for the test user | |
| } | |
| response = requests.post(login_url, data=json.dumps(login_data), | |
| headers={"Content-Type": "application/json"}) | |
| if response.status_code == 200: | |
| token = response.json().get("access_token") | |
| print(f"Authentication successful. Token: {token[:15]}..." if token else "No token in response") | |
| 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 getting token: {e}") | |
| return None | |
| def create_test_image(): | |
| """Create a test image file""" | |
| # Check if we're in the right directory | |
| current_dir = Path.cwd() | |
| print(f"Current directory: {current_dir}") | |
| # Create a directory for test files if it doesn't exist | |
| test_dir = current_dir / "test_files" | |
| test_dir.mkdir(exist_ok=True) | |
| # Path to the test image file | |
| test_image_path = test_dir / "test_image.txt" | |
| # Write some content to the file | |
| with open(test_image_path, "w") as f: | |
| f.write("This is a test image file for the API") | |
| print(f"Created test file at {test_image_path}") | |
| return test_image_path | |
| def test_report_incident_with_image(): | |
| """Test reporting an incident with an image""" | |
| # Create a test image | |
| image_path = create_test_image() | |
| # Get a token if not provided | |
| token = JWT_TOKEN or get_token() | |
| if not token: | |
| print("No token available, cannot proceed with test") | |
| return | |
| try: | |
| # Prepare the form data | |
| form_data = { | |
| "description": "Test incident with image upload to Cloudinary", | |
| "name": "Cloudinary Test", | |
| "latitude": "37.7749", | |
| "longitude": "-122.4194" | |
| } | |
| # Prepare the file | |
| with open(image_path, "rb") as image_file: | |
| files = {"image": ("test_image.txt", image_file, "text/plain")} | |
| # Make the API request | |
| print(f"Sending request to {API_URL}") | |
| response = requests.post( | |
| API_URL, | |
| data=form_data, | |
| files=files, | |
| headers={"Authorization": f"Bearer {token}"} | |
| ) | |
| # Process the response | |
| if response.status_code == 200: | |
| print("Incident report successful!") | |
| print(f"Response: {response.json()}") | |
| return response.json() | |
| else: | |
| print(f"API request failed with status code {response.status_code}") | |
| print(f"Response: {response.text}") | |
| return None | |
| except Exception as e: | |
| print(f"Error testing incident report API: {e}") | |
| return None | |
| finally: | |
| # Clean up test image | |
| try: | |
| if os.path.exists(image_path): | |
| os.remove(image_path) | |
| print(f"Removed test file: {image_path}") | |
| except Exception as e: | |
| print(f"Error cleaning up test file: {e}") | |
| if __name__ == "__main__": | |
| print("Testing Marine Guard Incident API with Cloudinary Integration") | |
| test_report_incident_with_image() |