"""Tests for data models and scenario manager.""" import pytest from mcp_servers.data.models import ( HUBS, CrewMember, CrewRole, DelayCause, FlightData, FlightStatus, HubInfo, MELItem, MPStatus, Passenger, RebookingCase, ) from mcp_servers.data.scenarios.manager import ScenarioManager, SCENARIO_MODULES class TestModels: def test_flight_data_serializes(self): from datetime import datetime, timezone f = FlightData( flight_id="UA100", origin="ORD", destination="SFO", scheduled_departure=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc), scheduled_arrival=datetime(2026, 1, 1, 15, 0, tzinfo=timezone.utc), status=FlightStatus.DELAYED, delay_minutes=30, delay_cause=DelayCause.WEATHER, aircraft_tail="N12345", gate="A1", ) d = f.model_dump(mode="json") assert d["flight_id"] == "UA100" assert d["status"] == "DELAYED" assert d["delay_cause"] == "WEATHER" def test_crew_member_serializes(self): c = CrewMember( crew_id="CR-1", name="Test Pilot", role=CrewRole.CAPTAIN, duty_hours_elapsed=10.0, duty_hours_limit=14.0, rest_hours_since_last=12.0, base_hub="ORD", ) d = c.model_dump(mode="json") assert d["role"] == "CAPTAIN" assert d["duty_hours_elapsed"] == 10.0 def test_mp_status_values(self): assert MPStatus.GLOBAL_SERVICES.value == "GLOBAL_SERVICES" assert MPStatus.K1.value == "1K" def test_hubs_reference_data(self): assert "ORD" in HUBS assert "EWR" in HUBS assert "IAH" in HUBS assert "SFO" in HUBS assert "DEN" in HUBS assert len(HUBS) == 5 ord = HUBS["ORD"] assert ord.code == "ORD" assert ord.city == "Chicago" assert ord.latitude == pytest.approx(41.97, abs=0.01) assert ord.gates > 0 assert ord.runways > 0 class TestScenarioManager: def test_list_scenarios(self): mgr = ScenarioManager() scenarios = mgr.list_scenarios() assert len(scenarios) == len(SCENARIO_MODULES) ids = [s["scenario_id"] for s in scenarios] assert "normal_ops" in ids assert "weather_disruption_ord" in ids assert "maintenance_delay_sfo" in ids assert "crew_swap_ewr" in ids def test_default_active_scenario(self): mgr = ScenarioManager() assert mgr.active_id == "weather_disruption_ord" def test_switch_scenario(self): mgr = ScenarioManager() result = mgr.set_active("normal_ops") assert mgr.active_id == "normal_ops" assert result["scenario_id"] == "normal_ops" def test_switch_invalid_scenario(self): mgr = ScenarioManager() with pytest.raises(ValueError, match="Unknown scenario"): mgr.set_active("nonexistent") def test_flights_per_scenario(self): mgr = ScenarioManager() for sid in SCENARIO_MODULES: mgr.set_active(sid) assert len(mgr.flights) > 0, f"No flights in {sid}" for f in mgr.flights: assert f.flight_id, f"Empty flight_id in {sid}" assert f.origin, f"Empty origin in {sid}" def test_crew_per_scenario(self): mgr = ScenarioManager() for sid in SCENARIO_MODULES: mgr.set_active(sid) assert len(mgr.crew) > 0, f"No crew in {sid}" for c in mgr.crew: assert c.crew_id assert c.duty_hours_limit > 0 def test_scenario_internal_consistency(self): """Crew IDs on flights should exist in the crew roster.""" mgr = ScenarioManager() for sid in SCENARIO_MODULES: mgr.set_active(sid) crew_ids = {c.crew_id for c in mgr.crew} for f in mgr.flights: for cid in f.crew_ids: assert cid in crew_ids, ( f"Flight {f.flight_id} in {sid} references crew {cid} " f"not in roster" ) def test_disrupted_flights_have_cause(self): mgr = ScenarioManager() for sid in SCENARIO_MODULES: mgr.set_active(sid) for f in mgr.flights: if f.status != FlightStatus.ON_TIME: assert f.delay_cause is not None, ( f"Flight {f.flight_id} in {sid} is {f.status} " f"but has no delay_cause" ) def test_metadata_counts(self): mgr = ScenarioManager() mgr.set_active("weather_disruption_ord") meta = mgr.get_metadata() assert meta["flight_count"] == len(mgr.flights) disrupted = sum(1 for f in mgr.flights if f.status != FlightStatus.ON_TIME) assert meta["disrupted_flights"] == disrupted