228 lines
7.7 KiB
Python
228 lines
7.7 KiB
Python
from datetime import datetime
|
|
|
|
from app.models.arrival import Arrival
|
|
from app.models.departure import Departure
|
|
from app.models.local_flight import LocalFlight, LocalFlightStatus, LocalFlightType
|
|
from app.models.movement import Movement, MovementType
|
|
from app.models.overflight import Overflight, OverflightStatus
|
|
from app.models.ppr import PPRRecord, PPRStatus
|
|
|
|
|
|
def movement_payload(**overrides):
|
|
payload = {
|
|
"flight_kind": "ARRIVAL",
|
|
"movement_date": "2026-06-20",
|
|
"movement_time": "10:00",
|
|
"aircraft_registration": "G-MOV1",
|
|
"aircraft_type": "PA28",
|
|
"callsign": "GMOV1",
|
|
"from_location": "EGLL",
|
|
"to_location": "EGKK",
|
|
"pob": 2,
|
|
"runway": "27",
|
|
"wind": "270/10",
|
|
"pressure_setting": "QNH1013",
|
|
"notes": "Bulk movement",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def test_movement_list_get_and_context_for_ppr(auth_client, db):
|
|
ppr = PPRRecord(
|
|
status=PPRStatus.NEW,
|
|
ac_reg="G-MOV1",
|
|
ac_type="PA28",
|
|
ac_call="GMOV1",
|
|
captain="Movement Pilot",
|
|
in_from="EGLL",
|
|
eta=datetime(2026, 6, 20, 10, 0),
|
|
pob_in=2,
|
|
out_to="EGKK",
|
|
etd=datetime(2026, 6, 20, 11, 0),
|
|
created_by="test",
|
|
public_token="movement-ppr",
|
|
)
|
|
db.add(ppr)
|
|
db.commit()
|
|
db.refresh(ppr)
|
|
|
|
bulk_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(ppr_id=ppr.id, landing_time="10:05"),
|
|
)
|
|
|
|
assert bulk_response.status_code == 200
|
|
result = bulk_response.json()
|
|
assert result["action"] == "created"
|
|
assert result["entity_type"] == "PPR"
|
|
assert result["entity_id"] == ppr.id
|
|
|
|
list_response = auth_client.get(
|
|
"/api/v1/movements/",
|
|
params={
|
|
"movement_type": "LANDING",
|
|
"aircraft_registration": "MOV1",
|
|
"date_from": "2026-06-20",
|
|
"date_to": "2026-06-20",
|
|
"entity_type": "PPR",
|
|
},
|
|
)
|
|
get_response = auth_client.get(f"/api/v1/movements/{result['movement']['id']}")
|
|
context_response = auth_client.get(
|
|
"/api/v1/movements/bulk-context",
|
|
params={
|
|
"target_date": "2026-06-20",
|
|
"aircraft_registration": "G-MOV1",
|
|
"flight_kind": "ARRIVAL",
|
|
},
|
|
)
|
|
|
|
assert list_response.status_code == 200
|
|
assert [movement["id"] for movement in list_response.json()] == [result["movement"]["id"]]
|
|
assert get_response.status_code == 200
|
|
assert get_response.json()["aircraft_registration"] == "G-MOV1"
|
|
assert context_response.status_code == 200
|
|
context = context_response.json()
|
|
assert context["pprs"][0]["id"] == ppr.id
|
|
assert context["movements"][0]["id"] == result["movement"]["id"]
|
|
assert context["suggested"]["source"] == "movement"
|
|
|
|
|
|
def test_bulk_log_updates_existing_movement_and_creates_unmatched_arrival_departure(auth_client, db):
|
|
arrival_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(aircraft_registration="G-NEW1", landing_time="10:00", from_location="EGBB"),
|
|
)
|
|
update_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(
|
|
aircraft_registration="G-NEW1",
|
|
landing_time="10:15",
|
|
from_location="EGBB",
|
|
notes="Updated movement",
|
|
),
|
|
)
|
|
departure_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(
|
|
flight_kind="DEPARTURE",
|
|
aircraft_registration="G-NEW2",
|
|
takeoff_time="11:00",
|
|
to_location="EGCC",
|
|
),
|
|
)
|
|
|
|
assert arrival_response.status_code == 200
|
|
assert arrival_response.json()["entity_type"] == "ARRIVAL"
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["action"] == "updated"
|
|
assert update_response.json()["movement"]["timestamp"] == "2026-06-20T10:15:00"
|
|
assert departure_response.status_code == 200
|
|
assert departure_response.json()["entity_type"] == "DEPARTURE"
|
|
|
|
arrival = db.query(Arrival).filter(Arrival.registration == "G-NEW1").one()
|
|
departure = db.query(Departure).filter(Departure.registration == "G-NEW2").one()
|
|
assert arrival.status.value == "LANDED"
|
|
assert departure.status.value == "DEPARTED"
|
|
|
|
|
|
def test_bulk_log_local_and_overflight_branches(auth_client, db):
|
|
local_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(
|
|
flight_kind="LOCAL",
|
|
aircraft_registration="G-LOCX",
|
|
takeoff_time="09:00",
|
|
landing_time="09:45",
|
|
local_nature="CIRCUITS",
|
|
circuits=3,
|
|
),
|
|
)
|
|
overflight_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(
|
|
flight_kind="OVERFLIGHT",
|
|
aircraft_registration="G-OVRX",
|
|
contact_time="12:00",
|
|
qsy_time="12:15",
|
|
from_location="EGLL",
|
|
to_location="EGKK",
|
|
),
|
|
)
|
|
overflight_update_response = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(
|
|
flight_kind="OVERFLIGHT",
|
|
aircraft_registration="G-OVRX",
|
|
contact_time="12:05",
|
|
qsy_time="12:20",
|
|
from_location="EGLL",
|
|
to_location="EGCC",
|
|
),
|
|
)
|
|
|
|
assert local_response.status_code == 200
|
|
assert local_response.json()["entity_type"] == "LOCAL_FLIGHT"
|
|
local = db.query(LocalFlight).filter(LocalFlight.registration == "G-LOCX").one()
|
|
assert local.status == LocalFlightStatus.LANDED
|
|
assert local.flight_type == LocalFlightType.CIRCUITS
|
|
assert local.circuits == 3
|
|
|
|
local_movements = db.query(Movement).filter(Movement.entity_type == "LOCAL_FLIGHT").all()
|
|
assert {movement.movement_type for movement in local_movements} == {
|
|
MovementType.TAKEOFF,
|
|
MovementType.LANDING,
|
|
}
|
|
|
|
assert overflight_response.status_code == 200
|
|
assert overflight_response.json()["entity_type"] == "OVERFLIGHT"
|
|
assert overflight_update_response.status_code == 200
|
|
assert overflight_update_response.json()["action"] == "updated"
|
|
overflight = db.query(Overflight).filter(Overflight.registration == "G-OVRX").one()
|
|
assert overflight.status == OverflightStatus.INACTIVE
|
|
assert overflight.destination_airfield == "EGCC"
|
|
|
|
|
|
def test_movement_error_paths(auth_client):
|
|
missing_registration = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(aircraft_registration=""),
|
|
)
|
|
invalid_kind = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(flight_kind="BALLOON"),
|
|
)
|
|
missing_time = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(movement_time=None, landing_time=None),
|
|
)
|
|
invalid_time = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(landing_time="not-time"),
|
|
)
|
|
bad_local_times = auth_client.post(
|
|
"/api/v1/movements/bulk-log",
|
|
json=movement_payload(
|
|
flight_kind="LOCAL",
|
|
takeoff_time="11:00",
|
|
landing_time="10:00",
|
|
),
|
|
)
|
|
bad_context = auth_client.get(
|
|
"/api/v1/movements/bulk-context",
|
|
params={
|
|
"target_date": "2026-06-20",
|
|
"aircraft_registration": "G-BAD",
|
|
"flight_kind": "BALLOON",
|
|
},
|
|
)
|
|
|
|
assert missing_registration.status_code == 400
|
|
assert invalid_kind.status_code == 400
|
|
assert missing_time.status_code == 400
|
|
assert invalid_time.status_code == 400
|
|
assert bad_local_times.status_code == 400
|
|
assert bad_context.status_code == 400
|
|
assert auth_client.get("/api/v1/movements/404").status_code == 404
|