127 lines
4 KiB
Python
127 lines
4 KiB
Python
|
#!/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()
|