example test for integration

DataService/api.py

@@ -2,6 +2,10 @@
 FastAPI Service for Metro Train Schedule Generation
 Provides endpoints for synthetic data generation and schedule optimization
 """
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
 from fastapi import FastAPI, HTTPException
 from fastapi.responses import JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
@@ -9,11 +13,11 @@ from pydantic import ValidationError
 from datetime import datetime
 import logging
 
-from .metro_models import (
+from DataService.metro_models import (
     DaySchedule, ScheduleRequest, Route, TrainHealthStatus
 )
-from .metro_data_generator import MetroDataGenerator
-from .schedule_optimizer import MetroScheduleOptimizer
+from DataService.metro_data_generator import MetroDataGenerator
+from DataService.schedule_optimizer import MetroScheduleOptimizer
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)

DataService/metro_data_generator.py

@@ -2,11 +2,16 @@
 Enhanced Metro Synthetic Data Generator
 Generates realistic metro train scheduling data with time-based constraints
 """
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
 import random
 import uuid
 from datetime import datetime, timedelta, time
 from typing import List, Dict, Tuple
-from .metro_models import (
+
+from DataService.metro_models import (
     TrainHealthStatus, Station, Route, FitnessCertificates,
     FitnessCertificate, CertificateStatus, JobCards, Branding
 )

DataService/schedule_optimizer.py

@@ -3,16 +3,21 @@ Metro Train Schedule Optimizer
 Generates optimal daily schedules from 5:00 AM to 11:00 PM
 Considers train health, maintenance, branding, and mileage balancing
 """
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
 import random
 from datetime import datetime, time, timedelta
 from typing import List, Dict, Tuple, Optional
-from .metro_models import (
+
+from DataService.metro_models import (
     DaySchedule, Trainset, TrainStatus, ServiceBlock, FleetSummary,
     OptimizationMetrics, Alert, Severity, DecisionRationale,
     TrainHealthStatus, Route, OperationalHours, FitnessCertificates,
     JobCards, Branding, CertificateStatus, MaintenanceType
 )
-from .metro_data_generator import MetroDataGenerator
+from DataService.metro_data_generator import MetroDataGenerator
 
 
 class MetroScheduleOptimizer:

requirements.txt

@@ -2,4 +2,5 @@ ortools==9.14.6206
 fastapi==0.104.1
 uvicorn[standard]==0.24.0
 pydantic==2.5.0
-python-multipart==0.0.6
+python-multipart==0.0.6
+requests==2.31.0

test_api.py

@@ -0,0 +1,207 @@
+"""
+API Test Suite for DataService
+Tests all endpoints and saves results to JSON
+"""
+import requests
+import json
+from datetime import datetime
+
+
+BASE_URL = "http://localhost:8000"
+results = {
+    "test_run": datetime.now().isoformat(),
+    "base_url": BASE_URL,
+    "tests": []
+}
+
+
+def test_endpoint(name, method, endpoint, **kwargs):
+    """Test an API endpoint and record results"""
+    print(f"\nTesting: {name}")
+    print(f"  {method} {endpoint}")
+    
+    try:
+        url = f"{BASE_URL}{endpoint}"
+        
+        if method == "GET":
+            response = requests.get(url, **kwargs)
+        elif method == "POST":
+            response = requests.post(url, **kwargs)
+        else:
+            raise ValueError(f"Unsupported method: {method}")
+        
+        result = {
+            "name": name,
+            "method": method,
+            "endpoint": endpoint,
+            "status_code": response.status_code,
+            "success": response.status_code in [200, 201],
+            "response_size": len(response.text),
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        if response.status_code == 200:
+            try:
+                result["response_data"] = response.json()
+                print(f"  ✓ Success - {response.status_code}")
+            except:
+                result["response_text"] = response.text[:200]
+                print(f"  ✓ Success - {response.status_code} (non-JSON)")
+        else:
+            result["error"] = response.text[:500]
+            print(f"  ✗ Failed - {response.status_code}")
+        
+        results["tests"].append(result)
+        return result
+        
+    except Exception as e:
+        result = {
+            "name": name,
+            "method": method,
+            "endpoint": endpoint,
+            "success": False,
+            "error": str(e),
+            "timestamp": datetime.now().isoformat()
+        }
+        results["tests"].append(result)
+        print(f"  ✗ Error: {e}")
+        return result
+
+
+def run_tests():
+    """Run all API tests"""
+    print("=" * 70)
+    print("DataService API Test Suite")
+    print("=" * 70)
+    
+    # Test 1: Root endpoint
+    test_endpoint(
+        "Root Endpoint",
+        "GET",
+        "/"
+    )
+    
+    # Test 2: Health check
+    test_endpoint(
+        "Health Check",
+        "GET",
+        "/health"
+    )
+    
+    # Test 3: Quick schedule generation
+    test_endpoint(
+        "Quick Schedule Generation",
+        "POST",
+        "/api/v1/generate/quick?date=2025-10-26&num_trains=25&num_stations=25"
+    )
+    
+    # Test 4: Full schedule generation
+    test_endpoint(
+        "Full Schedule Generation",
+        "POST",
+        "/api/v1/generate",
+        json={
+            "date": "2025-10-26",
+            "num_trains": 30,
+            "num_stations": 25,
+            "min_service_trains": 22,
+            "min_standby_trains": 3
+        },
+        headers={"Content-Type": "application/json"}
+    )
+    
+    # Test 5: Example schedule
+    test_endpoint(
+        "Example Schedule",
+        "GET",
+        "/api/v1/schedule/example"
+    )
+    
+    # Test 6: Route information
+    test_endpoint(
+        "Route Information (25 stations)",
+        "GET",
+        "/api/v1/route/25"
+    )
+    
+    # Test 7: Train health data
+    test_endpoint(
+        "Train Health Data (30 trains)",
+        "GET",
+        "/api/v1/trains/health/30"
+    )
+    
+    # Test 8: Depot layout
+    test_endpoint(
+        "Depot Layout",
+        "GET",
+        "/api/v1/depot/layout"
+    )
+    
+    # Test 9: Custom schedule with all parameters
+    test_endpoint(
+        "Custom Schedule Full Parameters",
+        "POST",
+        "/api/v1/generate",
+        json={
+            "date": "2025-11-01",
+            "num_trains": 35,
+            "num_stations": 25,
+            "route_name": "Aluva-Pettah Line",
+            "depot_name": "Muttom_Depot",
+            "min_service_trains": 25,
+            "min_standby_trains": 5,
+            "max_daily_km_per_train": 280,
+            "balance_mileage": True,
+            "prioritize_branding": True
+        },
+        headers={"Content-Type": "application/json"}
+    )
+    
+    # Test 10: Quick generation with minimal params
+    test_endpoint(
+        "Quick Generation Minimal",
+        "POST",
+        "/api/v1/generate/quick?date=2025-10-27&num_trains=20"
+    )
+
+
+def save_results():
+    """Save test results to JSON file"""
+    # Summary
+    total = len(results["tests"])
+    passed = sum(1 for t in results["tests"] if t.get("success", False))
+    
+    results["summary"] = {
+        "total_tests": total,
+        "passed": passed,
+        "failed": total - passed,
+        "success_rate": f"{(passed/total)*100:.1f}%" if total > 0 else "0%"
+    }
+    
+    filename = f"api_test_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+    
+    with open(filename, 'w') as f:
+        json.dump(results, f, indent=2, default=str)
+    
+    print("\n" + "=" * 70)
+    print("Test Summary")
+    print("=" * 70)
+    print(f"Total Tests: {total}")
+    print(f"Passed: {passed}")
+    print(f"Failed: {total - passed}")
+    print(f"Success Rate: {results['summary']['success_rate']}")
+    print(f"\nResults saved to: {filename}")
+    print("=" * 70)
+
+
+if __name__ == "__main__":
+    try:
+        run_tests()
+        save_results()
+    except KeyboardInterrupt:
+        print("\n\nTests interrupted by user")
+        save_results()
+    except Exception as e:
+        print(f"\n\nError during testing: {e}")
+        save_results()