added tests
This commit is contained in:
150
tests/test_models.py
Normal file
150
tests/test_models.py
Normal file
@@ -0,0 +1,150 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user