add extractors, rate limit, demo submit form

extractors/test_sncf_travaux.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""
+Test script for the SNCF travaux extractor.
+
+This script tests the functionality of the SNCF travaux extractor without actually
+submitting events to the database.
+"""
+
+import sys
+import os
+import json
+from datetime import datetime
+
+# Add the parent directory to the path so we can import from the extractor
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+# Import functions from the extractor
+from extractors.sncf_travaux import fetch_sncf_data, week_to_date, create_event
+
+def test_week_to_date():
+    """Test the week_to_date function."""
+    print("Testing week_to_date function...")
+    
+    # Test with valid inputs
+    year = 2023
+    week = 31
+    start_date, end_date = week_to_date(year, week)
+    print(f"Week {week} of {year} starts on {start_date} and ends on {end_date}")
+    
+    # Test with string inputs
+    year = "2023"
+    week = "31"
+    start_date, end_date = week_to_date(year, week)
+    print(f"Week {week} of {year} starts on {start_date} and ends on {end_date}")
+    
+    # Test with invalid week number
+    year = 2023
+    week = 60  # Invalid week number
+    start_date, end_date = week_to_date(year, week)
+    print(f"Invalid week {week} of {year} returns {start_date} to {end_date}")
+    
+    print("week_to_date function test completed.\n")
+
+def test_create_event():
+    """Test the create_event function."""
+    print("Testing create_event function...")
+    
+    # Create a sample record
+    record = {
+        "lib_structdem": "Siège INFRAPOLE PACA",
+        "cod_ligne": "830000",
+        "lib_ligne": "Ligne de Paris-Lyon à Marseille-St-Charles",
+        "pk_debm": "687000",
+        "pk_finm": "862100",
+        "familletravaux": "renouvellement de la signalisation",
+        "nb_interventions": 1,
+        "num_semaine": "31",
+        "annee": "2023"
+    }
+    
+    # Create an event from the record
+    event = create_event(record)
+    
+    if event:
+        print("Event created successfully:")
+        print(f"Label: {event['properties']['label']}")
+        print(f"Start date: {event['properties']['start']}")
+        print(f"End date: {event['properties']['stop']}")
+        print(f"Type: {event['properties']['type']}")
+        print(f"What: {event['properties']['what']}")
+        print(f"Where: {event['properties']['where']}")
+        print(f"Description: {event['properties']['description'][:100]}...")
+    else:
+        print("Failed to create event from record.")
+    
+    # Test with missing required fields
+    record_missing = {
+        "lib_structdem": "Siège INFRAPOLE PACA",
+        "cod_ligne": "830000",
+        "lib_ligne": "Ligne de Paris-Lyon à Marseille-St-Charles",
+        # Missing year and week number
+    }
+    
+    event_missing = create_event(record_missing)
+    if event_missing is None:
+        print("Correctly handled record with missing required fields.")
+    else:
+        print("Failed to handle record with missing required fields.")
+    
+    print("create_event function test completed.\n")
+
+def test_fetch_sncf_data():
+    """Test the fetch_sncf_data function."""
+    print("Testing fetch_sncf_data function...")
+    print("Note: This test will make an actual API request to the SNCF API.")
+    print("If you're not connected to the internet, this test will fail.")
+    
+    # Fetch data from the SNCF API
+    records = fetch_sncf_data()
+    
+    if records:
+        print(f"Successfully fetched {len(records)} records from the SNCF API.")
+        print("First record:")
+        print(json.dumps(records[0], indent=2))
+    else:
+        print("Failed to fetch data from the SNCF API.")
+    
+    print("fetch_sncf_data function test completed.\n")
+
+def main():
+    """Run all tests."""
+    print("Starting tests for SNCF travaux extractor...")
+    
+    # Test the week_to_date function
+    test_week_to_date()
+    
+    # Test the create_event function
+    test_create_event()
+    
+    # Test the fetch_sncf_data function
+    # Uncomment the following line to test the API request
+    # test_fetch_sncf_data()
+    
+    print("All tests completed.")
+
+if __name__ == "__main__":
+    main()