284 lines
11 KiB
Python
284 lines
11 KiB
Python
from datetime import datetime
|
|
|
|
from app.models.arrival import Arrival
|
|
from app.models.departure import Departure, DepartureStatus
|
|
from app.models.local_flight import LocalFlight, LocalFlightStatus, LocalFlightType
|
|
from app.models.movement import Movement, MovementType
|
|
from app.models.overflight import Overflight, OverflightStatus
|
|
|
|
|
|
def test_arrival_lifecycle_and_not_found_paths(auth_client, db):
|
|
payload = {
|
|
"registration": "g-arr",
|
|
"type": "DA40",
|
|
"callsign": "GARR",
|
|
"pob": 2,
|
|
"in_from": "egll",
|
|
"eta": "2026-06-20T09:30:00",
|
|
"notes": "Inbound test",
|
|
}
|
|
|
|
create_response = auth_client.post("/api/v1/arrivals/", json=payload)
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["registration"] == "G-ARR"
|
|
assert created["status"] == "INBOUND"
|
|
|
|
assert auth_client.get(f"/api/v1/arrivals/{created['id']}").status_code == 200
|
|
|
|
list_response = auth_client.get(
|
|
"/api/v1/arrivals/",
|
|
params={"status": "INBOUND", "date_from": "2026-06-20", "date_to": "2026-06-20"},
|
|
)
|
|
assert list_response.status_code == 200
|
|
assert [arrival["id"] for arrival in list_response.json()] == [created["id"]]
|
|
|
|
update_response = auth_client.put(
|
|
f"/api/v1/arrivals/{created['id']}",
|
|
json={"notes": "Updated inbound", "callsign": "ARRIVE"},
|
|
)
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["notes"] == "Updated inbound"
|
|
|
|
status_response = auth_client.patch(
|
|
f"/api/v1/arrivals/{created['id']}/status",
|
|
json={"status": "LANDED", "timestamp": "2026-06-20T10:00:00"},
|
|
)
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["landed_dt"] == "2026-06-20T10:00:00"
|
|
|
|
movement = db.query(Movement).filter(Movement.entity_type == "ARRIVAL").one()
|
|
assert movement.movement_type == MovementType.LANDING
|
|
assert movement.aircraft_registration == "G-ARR"
|
|
|
|
cancel_response = auth_client.delete(f"/api/v1/arrivals/{created['id']}")
|
|
assert cancel_response.status_code == 200
|
|
assert cancel_response.json()["status"] == "CANCELLED"
|
|
|
|
assert auth_client.get("/api/v1/arrivals/404").status_code == 404
|
|
assert auth_client.put("/api/v1/arrivals/404", json={"notes": "x"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/arrivals/404/status", json={"status": "LANDED"}).status_code == 404
|
|
assert auth_client.delete("/api/v1/arrivals/404").status_code == 404
|
|
|
|
|
|
def test_landing_arrival_promotes_linked_pending_departure(auth_client, db):
|
|
arrival = Arrival(
|
|
registration="G-LINK",
|
|
type="PA28",
|
|
callsign="GLINK",
|
|
pob=2,
|
|
in_from="EGLL",
|
|
status="INBOUND",
|
|
eta=datetime(2026, 6, 20, 9, 30),
|
|
created_by="test",
|
|
)
|
|
db.add(arrival)
|
|
db.commit()
|
|
db.refresh(arrival)
|
|
|
|
departure = Departure(
|
|
registration="G-LINK",
|
|
type="PA28",
|
|
callsign="GLINK",
|
|
pob=2,
|
|
out_to="EGKK",
|
|
status=DepartureStatus.PENDING,
|
|
arrival_id=arrival.id,
|
|
created_by="test",
|
|
)
|
|
db.add(departure)
|
|
db.commit()
|
|
|
|
response = auth_client.patch(
|
|
f"/api/v1/arrivals/{arrival.id}/status",
|
|
json={"status": "LANDED", "timestamp": "2026-06-20T10:00:00"},
|
|
)
|
|
|
|
db.refresh(departure)
|
|
assert response.status_code == 200
|
|
assert departure.status == DepartureStatus.BOOKED_OUT
|
|
|
|
|
|
def test_departure_lifecycle_and_not_found_paths(auth_client, db):
|
|
payload = {
|
|
"registration": "g-dep",
|
|
"type": "SR22",
|
|
"callsign": "GDEP",
|
|
"pob": 2,
|
|
"out_to": "egkk",
|
|
"etd": "2026-06-20T11:00:00",
|
|
"notes": "Outbound test",
|
|
}
|
|
|
|
create_response = auth_client.post("/api/v1/departures/", json=payload)
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["registration"] == "G-DEP"
|
|
assert created["status"] == "GROUND"
|
|
|
|
list_response = auth_client.get(
|
|
"/api/v1/departures/",
|
|
params={"status": "GROUND", "date_from": "2026-06-20", "date_to": "2026-06-20"},
|
|
)
|
|
assert list_response.status_code == 200
|
|
assert [departure["id"] for departure in list_response.json()] == [created["id"]]
|
|
|
|
update_response = auth_client.put(
|
|
f"/api/v1/departures/{created['id']}",
|
|
json={"notes": "Updated outbound", "callsign": "DEPART"},
|
|
)
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["notes"] == "Updated outbound"
|
|
|
|
status_response = auth_client.patch(
|
|
f"/api/v1/departures/{created['id']}/status",
|
|
json={"status": "LOCAL", "timestamp": "2026-06-20T11:10:00"},
|
|
)
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["takeoff_dt"] == "2026-06-20T11:10:00"
|
|
|
|
movement = db.query(Movement).filter(Movement.entity_type == "DEPARTURE").one()
|
|
assert movement.movement_type == MovementType.TAKEOFF
|
|
assert movement.to_location == "egkk"
|
|
|
|
cancel_response = auth_client.delete(f"/api/v1/departures/{created['id']}")
|
|
assert cancel_response.status_code == 200
|
|
assert cancel_response.json()["status"] == "CANCELLED"
|
|
|
|
assert auth_client.get("/api/v1/departures/404").status_code == 404
|
|
assert auth_client.put("/api/v1/departures/404", json={"notes": "x"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/departures/404/status", json={"status": "DEPARTED"}).status_code == 404
|
|
assert auth_client.delete("/api/v1/departures/404").status_code == 404
|
|
|
|
|
|
def test_local_flight_lifecycle_special_lists_and_not_found_paths(auth_client, db):
|
|
payload = {
|
|
"registration": "g-loc",
|
|
"type": "C152",
|
|
"callsign": "GLOC",
|
|
"pob": 1,
|
|
"flight_type": "LOCAL",
|
|
"duration": 45,
|
|
"etd": "2026-06-20T10:00:00",
|
|
"notes": "Local test",
|
|
}
|
|
|
|
create_response = auth_client.post("/api/v1/local-flights/", json=payload)
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["registration"] == "G-LOC"
|
|
assert created["status"] == "GROUND"
|
|
|
|
filter_response = auth_client.get(
|
|
"/api/v1/local-flights/",
|
|
params={"status": "GROUND", "flight_type": "LOCAL", "date_from": "2026-06-20"},
|
|
)
|
|
assert filter_response.status_code == 200
|
|
assert [flight["id"] for flight in filter_response.json()] == [created["id"]]
|
|
|
|
update_response = auth_client.put(
|
|
f"/api/v1/local-flights/{created['id']}",
|
|
json={"notes": "Updated local", "duration": 60},
|
|
)
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["duration"] == 60
|
|
|
|
departed_response = auth_client.patch(
|
|
f"/api/v1/local-flights/{created['id']}/status",
|
|
json={"status": "DEPARTED", "timestamp": "2026-06-20T10:05:00"},
|
|
)
|
|
landed_response = auth_client.patch(
|
|
f"/api/v1/local-flights/{created['id']}/status",
|
|
json={"status": "LANDED", "timestamp": "2026-06-20T10:45:00"},
|
|
)
|
|
|
|
assert departed_response.status_code == 200
|
|
assert departed_response.json()["takeoff_dt"] == "2026-06-20T10:05:00"
|
|
assert landed_response.status_code == 200
|
|
assert landed_response.json()["landed_dt"] == "2026-06-20T10:45:00"
|
|
|
|
movement_types = {
|
|
movement.movement_type
|
|
for movement in db.query(Movement).filter(Movement.entity_type == "LOCAL_FLIGHT").all()
|
|
}
|
|
assert movement_types == {MovementType.TAKEOFF, MovementType.LANDING}
|
|
|
|
active_response = auth_client.get("/api/v1/local-flights/active/current")
|
|
today_departures_response = auth_client.get("/api/v1/local-flights/today/departures")
|
|
booked_out_response = auth_client.get("/api/v1/local-flights/today/booked-out")
|
|
|
|
assert active_response.status_code == 200
|
|
assert today_departures_response.status_code == 200
|
|
assert booked_out_response.status_code == 200
|
|
assert booked_out_response.json()[0]["id"] == created["id"]
|
|
|
|
cancel_response = auth_client.delete(f"/api/v1/local-flights/{created['id']}")
|
|
assert cancel_response.status_code == 200
|
|
assert cancel_response.json()["status"] == "CANCELLED"
|
|
|
|
assert auth_client.get("/api/v1/local-flights/404").status_code == 404
|
|
assert auth_client.put("/api/v1/local-flights/404", json={"notes": "x"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/local-flights/404/status", json={"status": "LANDED"}).status_code == 404
|
|
assert auth_client.delete("/api/v1/local-flights/404").status_code == 404
|
|
|
|
|
|
def test_overflight_lifecycle_special_lists_and_not_found_paths(auth_client, db):
|
|
payload = {
|
|
"registration": "g-ovr",
|
|
"pob": 1,
|
|
"type": "PA28",
|
|
"departure_airfield": "egll",
|
|
"destination_airfield": "egkk",
|
|
"call_dt": "2026-06-20T09:00:00",
|
|
"notes": "Overflight test",
|
|
}
|
|
|
|
create_response = auth_client.post("/api/v1/overflights/", json=payload)
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["registration"] == "G-OVR"
|
|
assert created["departure_airfield"] == "EGLL"
|
|
assert created["status"] == "ACTIVE"
|
|
|
|
movement = db.query(Movement).filter(Movement.entity_type == "OVERFLIGHT").one()
|
|
assert movement.movement_type == MovementType.OVERFLIGHT
|
|
assert movement.from_location == "EGLL"
|
|
assert movement.to_location == "EGKK"
|
|
|
|
active_response = auth_client.get("/api/v1/overflights/active/list")
|
|
today_response = auth_client.get("/api/v1/overflights/today/list")
|
|
assert active_response.status_code == 200
|
|
assert active_response.json()[0]["id"] == created["id"]
|
|
assert today_response.status_code == 200
|
|
assert today_response.json()[0]["id"] == created["id"]
|
|
|
|
list_response = auth_client.get(
|
|
"/api/v1/overflights/",
|
|
params={"status": "ACTIVE", "date_from": "2026-06-20", "date_to": "2026-06-20"},
|
|
)
|
|
assert list_response.status_code == 200
|
|
assert [overflight["id"] for overflight in list_response.json()] == [created["id"]]
|
|
|
|
update_response = auth_client.put(
|
|
f"/api/v1/overflights/{created['id']}",
|
|
json={"notes": "Updated overflight", "destination_airfield": "egcc"},
|
|
)
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["destination_airfield"] == "EGCC"
|
|
|
|
status_response = auth_client.patch(
|
|
f"/api/v1/overflights/{created['id']}/status",
|
|
json={"status": "INACTIVE", "qsy_dt": "2026-06-20T09:20:00"},
|
|
)
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["qsy_dt"] == "2026-06-20T09:20:00"
|
|
|
|
cancel_response = auth_client.delete(f"/api/v1/overflights/{created['id']}")
|
|
assert cancel_response.status_code == 200
|
|
assert cancel_response.json()["status"] == "CANCELLED"
|
|
|
|
assert auth_client.get("/api/v1/overflights/404").status_code == 404
|
|
assert auth_client.put("/api/v1/overflights/404", json={"notes": "x"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/overflights/404/status", json={"status": "INACTIVE"}).status_code == 404
|
|
assert auth_client.delete("/api/v1/overflights/404").status_code == 404
|