188 lines
7.2 KiB
Python
188 lines
7.2 KiB
Python
from datetime import datetime
|
|
|
|
from app.models.arrival import Arrival
|
|
from app.models.departure import Departure, DepartureStatus
|
|
from app.models.ppr import PPRRecord
|
|
|
|
|
|
def test_ppr_routes_require_authentication(client):
|
|
response = client.get("/api/v1/pprs/")
|
|
|
|
assert response.status_code in (401, 403)
|
|
|
|
|
|
def test_authenticated_user_can_create_read_update_and_audit_ppr(auth_client, ppr_payload):
|
|
create_response = auth_client.post("/api/v1/pprs/", json=ppr_payload)
|
|
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["id"] > 0
|
|
assert created["status"] == "NEW"
|
|
assert created["ac_reg"] == "G-TEST"
|
|
assert created["created_by"] == "test-operator"
|
|
|
|
read_response = auth_client.get(f"/api/v1/pprs/{created['id']}")
|
|
|
|
assert read_response.status_code == 200
|
|
assert read_response.json()["ac_reg"] == "G-TEST"
|
|
|
|
status_response = auth_client.patch(
|
|
f"/api/v1/pprs/{created['id']}/status",
|
|
json={"status": "LANDED", "timestamp": "2026-06-20T10:30:00"},
|
|
)
|
|
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["status"] == "LANDED"
|
|
assert status_response.json()["landed_dt"] == "2026-06-20T10:30:00"
|
|
|
|
journal_response = auth_client.get(f"/api/v1/pprs/{created['id']}/journal")
|
|
|
|
assert journal_response.status_code == 200
|
|
entries = [entry["entry"] for entry in journal_response.json()]
|
|
assert any("PPR created for G-TEST" in entry for entry in entries)
|
|
assert any("Status changed from NEW to LANDED" in entry for entry in entries)
|
|
|
|
|
|
def test_ppr_list_supports_status_date_and_pagination_filters(auth_client, ppr_factory):
|
|
ppr_factory(
|
|
ac_reg="G-NEW1",
|
|
status="NEW",
|
|
eta=datetime(2026, 6, 20, 10, 0),
|
|
etd=datetime(2026, 6, 20, 12, 0),
|
|
public_token="token-new",
|
|
)
|
|
ppr_factory(
|
|
ac_reg="G-CAN1",
|
|
status="CANCELED",
|
|
eta=datetime(2026, 6, 21, 10, 0),
|
|
etd=datetime(2026, 6, 21, 12, 0),
|
|
public_token="token-canceled",
|
|
)
|
|
|
|
status_response = auth_client.get("/api/v1/pprs/", params={"status": "NEW"})
|
|
date_response = auth_client.get(
|
|
"/api/v1/pprs/",
|
|
params={"date_from": "2026-06-21", "date_to": "2026-06-21"},
|
|
)
|
|
limited_response = auth_client.get("/api/v1/pprs/", params={"skip": 1, "limit": 1})
|
|
|
|
assert status_response.status_code == 200
|
|
assert [ppr["ac_reg"] for ppr in status_response.json()] == ["G-NEW1"]
|
|
assert date_response.status_code == 200
|
|
assert [ppr["ac_reg"] for ppr in date_response.json()] == ["G-CAN1"]
|
|
assert limited_response.status_code == 200
|
|
assert len(limited_response.json()) == 1
|
|
|
|
|
|
def test_authenticated_user_can_put_patch_acknowledge_and_delete_ppr(auth_client, ppr_payload):
|
|
created = auth_client.post("/api/v1/pprs/", json=ppr_payload).json()
|
|
|
|
put_response = auth_client.put(
|
|
f"/api/v1/pprs/{created['id']}",
|
|
json={**ppr_payload, "captain": "Updated Pilot"},
|
|
)
|
|
patch_response = auth_client.patch(
|
|
f"/api/v1/pprs/{created['id']}",
|
|
json={"notes": "Updated by patch"},
|
|
)
|
|
acknowledge_response = auth_client.post(f"/api/v1/pprs/{created['id']}/acknowledge")
|
|
delete_response = auth_client.delete(f"/api/v1/pprs/{created['id']}")
|
|
|
|
assert put_response.status_code == 200
|
|
assert put_response.json()["captain"] == "Updated Pilot"
|
|
assert patch_response.status_code == 200
|
|
assert patch_response.json()["notes"] == "Updated by patch"
|
|
assert acknowledge_response.status_code == 200
|
|
assert acknowledge_response.json()["acknowledged_by"] == "test-operator"
|
|
assert delete_response.status_code == 200
|
|
assert delete_response.json()["status"] == "DELETED"
|
|
|
|
|
|
def test_ppr_not_found_paths(auth_client):
|
|
assert auth_client.get("/api/v1/pprs/404").status_code == 404
|
|
assert auth_client.put("/api/v1/pprs/404", json={"captain": "Nobody"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/pprs/404", json={"captain": "Nobody"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/pprs/404/status", json={"status": "LANDED"}).status_code == 404
|
|
assert auth_client.post("/api/v1/pprs/404/acknowledge").status_code == 404
|
|
assert auth_client.delete("/api/v1/pprs/404").status_code == 404
|
|
assert auth_client.get("/api/v1/pprs/404/journal").status_code == 404
|
|
assert auth_client.post("/api/v1/pprs/404/activate").status_code == 404
|
|
|
|
|
|
def test_public_ppr_create_sends_email_and_generates_token(client, db, ppr_payload, monkeypatch):
|
|
sent_email = {}
|
|
|
|
async def fake_send_email(**kwargs):
|
|
sent_email.update(kwargs)
|
|
return True
|
|
|
|
monkeypatch.setattr("app.api.endpoints.pprs.email_service.send_email", fake_send_email)
|
|
|
|
create_response = client.post("/api/v1/pprs/public", json=ppr_payload)
|
|
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["created_by"] == "public"
|
|
assert sent_email["to_email"] == "pilot@example.com"
|
|
|
|
db_ppr = db.query(PPRRecord).filter(PPRRecord.id == created["id"]).one()
|
|
assert db_ppr.public_token
|
|
|
|
|
|
def test_public_ppr_token_edit_and_cancel_paths(client, ppr_factory, db):
|
|
ppr = ppr_factory(public_token="public-edit-token", email=None)
|
|
|
|
get_response = client.get("/api/v1/pprs/public/edit/public-edit-token")
|
|
patch_response = client.patch(
|
|
"/api/v1/pprs/public/edit/public-edit-token",
|
|
json={"captain": "Public Editor"},
|
|
)
|
|
cancel_response = client.delete("/api/v1/pprs/public/cancel/public-edit-token")
|
|
|
|
assert get_response.status_code == 200
|
|
assert get_response.json()["id"] == ppr.id
|
|
assert patch_response.status_code == 200
|
|
assert patch_response.json()["captain"] == "Public Editor"
|
|
assert cancel_response.status_code == 200
|
|
assert cancel_response.json()["status"] == "CANCELED"
|
|
|
|
assert client.get("/api/v1/pprs/public/edit/public-edit-token").status_code == 400
|
|
assert client.patch("/api/v1/pprs/public/edit/missing-token", json={}).status_code == 404
|
|
assert client.delete("/api/v1/pprs/public/cancel/missing-token").status_code == 404
|
|
|
|
|
|
def test_activate_ppr_creates_arrival_and_pending_departure(auth_client, ppr_factory, db):
|
|
ppr = ppr_factory(public_token="activate-token")
|
|
|
|
response = auth_client.post(f"/api/v1/pprs/{ppr.id}/activate")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["arrival_id"]
|
|
assert body["departure_id"]
|
|
|
|
arrival = db.query(Arrival).filter(Arrival.id == body["arrival_id"]).one()
|
|
departure = db.query(Departure).filter(Departure.id == body["departure_id"]).one()
|
|
|
|
assert arrival.registration == "G-FACT"
|
|
assert arrival.in_from == "EGLL"
|
|
assert departure.status == DepartureStatus.PENDING
|
|
assert departure.arrival_id == arrival.id
|
|
|
|
|
|
def test_activate_rejects_processed_ppr(auth_client, ppr_factory):
|
|
ppr = ppr_factory(status="LANDED", public_token="processed-token")
|
|
|
|
response = auth_client.post(f"/api/v1/pprs/{ppr.id}/activate")
|
|
|
|
assert response.status_code == 400
|
|
assert "cannot be activated" in response.json()["detail"]
|
|
|
|
|
|
def test_invalid_ppr_payload_returns_validation_error(auth_client, ppr_payload):
|
|
ppr_payload["pob_in"] = -1
|
|
|
|
response = auth_client.post("/api/v1/pprs/", json=ppr_payload)
|
|
|
|
assert response.status_code == 422
|