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