80 lines
2.1 KiB
Python
80 lines
2.1 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Test script to verify that the osm_cal.py script can successfully submit events to the database.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
# Add the parent directory to the path so we can import from oedb
|
||
|
|
sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
|
||
|
|
|
||
|
|
from extractors.osm_cal import submit_event, load_env_from_file
|
||
|
|
from oedb.utils.logging import logger
|
||
|
|
|
||
|
|
def create_test_event():
|
||
|
|
"""
|
||
|
|
Create a test event with a simple point geometry.
|
||
|
|
"""
|
||
|
|
# Create a point geometry (Paris coordinates)
|
||
|
|
geometry = {
|
||
|
|
"type": "Point",
|
||
|
|
"coordinates": [2.3522, 48.8566]
|
||
|
|
}
|
||
|
|
|
||
|
|
# Create event properties
|
||
|
|
now = datetime.now()
|
||
|
|
tomorrow = now + timedelta(days=1)
|
||
|
|
|
||
|
|
properties = {
|
||
|
|
"type": "scheduled",
|
||
|
|
"what": "test.event",
|
||
|
|
"where": "Paris, France",
|
||
|
|
"label": "Test Event",
|
||
|
|
"description": "This is a test event created by the test_osm_cal.py script",
|
||
|
|
"start": now.isoformat(),
|
||
|
|
"stop": tomorrow.isoformat(),
|
||
|
|
"url": "https://example.com",
|
||
|
|
"external_id": f"test-event-{now.timestamp()}",
|
||
|
|
"source": "Test Script"
|
||
|
|
}
|
||
|
|
|
||
|
|
# Create the event object
|
||
|
|
event = {
|
||
|
|
"type": "Feature",
|
||
|
|
"geometry": geometry,
|
||
|
|
"properties": properties
|
||
|
|
}
|
||
|
|
|
||
|
|
return event
|
||
|
|
|
||
|
|
def test_submit_event():
|
||
|
|
"""
|
||
|
|
Test the submit_event function with a test event.
|
||
|
|
"""
|
||
|
|
logger.info("Starting test_submit_event")
|
||
|
|
|
||
|
|
# Load environment variables from .env file
|
||
|
|
if not load_env_from_file():
|
||
|
|
logger.error("Required .env file not found. Exiting.")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
# Create a test event
|
||
|
|
event = create_test_event()
|
||
|
|
logger.info(f"Created test event: {event['properties']['label']}")
|
||
|
|
|
||
|
|
# Submit the event
|
||
|
|
logger.info("Submitting test event to the database...")
|
||
|
|
result = submit_event(event)
|
||
|
|
|
||
|
|
if result:
|
||
|
|
logger.success("Successfully submitted test event to the database")
|
||
|
|
else:
|
||
|
|
logger.error("Failed to submit test event to the database")
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
test_submit_event()
|