157 lines
6.7 KiB
Python
157 lines
6.7 KiB
Python
from app.core.frz import point_inside_swansea_frz
|
|
from app.models.drone_request import DroneRequest
|
|
|
|
|
|
def drone_payload(**overrides):
|
|
payload = {
|
|
"operator_name": "Rotor Ops",
|
|
"operator_id": "OP-123",
|
|
"flyer_name": "Remote Pilot",
|
|
"flyer_id": "FLY-456",
|
|
"email": "pilot@example.com",
|
|
"phone": "0123456789",
|
|
"flight_date": "2026-06-20",
|
|
"estimated_takeoff_time": "10:00",
|
|
"estimated_completion_time": "10:30",
|
|
"estimated_takeoff_at": "2026-06-20T10:00:00",
|
|
"estimated_completion_at": "2026-06-20T10:30:00",
|
|
"maximum_elevation_ft_amsl": 250,
|
|
"location_description": "North apron",
|
|
"location_latitude": 51.623389,
|
|
"location_longitude": -4.069231,
|
|
"location_inside_frz": "no",
|
|
"notes": "Survey flight",
|
|
"prototype_overlay": {
|
|
"airport_reference_point": {"lat": 0, "lng": 0},
|
|
"frz_radius_metres": 1,
|
|
},
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def test_public_drone_request_create_edit_cancel_and_journal(client, db, monkeypatch):
|
|
sent_emails = []
|
|
|
|
async def fake_send_email(**kwargs):
|
|
sent_emails.append(kwargs)
|
|
return True
|
|
|
|
monkeypatch.setattr("app.api.endpoints.drone_requests.email_service.send_email", fake_send_email)
|
|
|
|
create_response = client.post("/api/v1/drone-requests/public", json=drone_payload())
|
|
|
|
assert create_response.status_code == 200
|
|
created = create_response.json()
|
|
assert created["reference_number"].startswith("DRN-")
|
|
assert created["status"] == "NEW"
|
|
assert created["location_inside_frz"] is True
|
|
assert created["created_by"] == "public"
|
|
assert len(sent_emails) == 1
|
|
|
|
db_request = db.query(DroneRequest).filter(DroneRequest.id == created["id"]).one()
|
|
assert db_request.public_token
|
|
|
|
get_response = client.get(f"/api/v1/drone-requests/public/edit/{db_request.public_token}")
|
|
patch_response = client.patch(
|
|
f"/api/v1/drone-requests/public/edit/{db_request.public_token}",
|
|
json={"operator_name": "Updated Rotor Ops", "notes": "Updated notes"},
|
|
)
|
|
cancel_response = client.delete(f"/api/v1/drone-requests/public/cancel/{db_request.public_token}")
|
|
|
|
assert get_response.status_code == 200
|
|
assert patch_response.status_code == 200
|
|
assert patch_response.json()["operator_name"] == "Updated Rotor Ops"
|
|
assert cancel_response.status_code == 200
|
|
assert cancel_response.json()["status"] == "CANCELED"
|
|
assert len(sent_emails) == 2
|
|
|
|
blocked_patch = client.patch(
|
|
f"/api/v1/drone-requests/public/edit/{db_request.public_token}",
|
|
json={"operator_name": "Too Late"},
|
|
)
|
|
blocked_cancel = client.delete(f"/api/v1/drone-requests/public/cancel/{db_request.public_token}")
|
|
|
|
assert blocked_patch.status_code == 400
|
|
assert blocked_cancel.status_code == 400
|
|
assert client.get("/api/v1/drone-requests/public/edit/missing-token").status_code == 404
|
|
assert client.patch("/api/v1/drone-requests/public/edit/missing-token", json={}).status_code == 404
|
|
assert client.delete("/api/v1/drone-requests/public/cancel/missing-token").status_code == 404
|
|
|
|
|
|
def test_authenticated_drone_request_list_update_status_comment_and_journal(auth_client, db, monkeypatch):
|
|
sent_emails = []
|
|
|
|
async def fake_send_email(**kwargs):
|
|
sent_emails.append(kwargs)
|
|
return True
|
|
|
|
monkeypatch.setattr("app.api.endpoints.drone_requests.email_service.send_email", fake_send_email)
|
|
|
|
create_response = auth_client.post("/api/v1/drone-requests/public", json=drone_payload())
|
|
created = create_response.json()
|
|
|
|
list_response = auth_client.get(
|
|
"/api/v1/drone-requests/",
|
|
params={"status": "NEW", "date_from": "2026-06-20", "date_to": "2026-06-20"},
|
|
)
|
|
get_response = auth_client.get(f"/api/v1/drone-requests/{created['id']}")
|
|
update_response = auth_client.patch(
|
|
f"/api/v1/drone-requests/{created['id']}",
|
|
json={"operator_comments": "Needs tower review", "maximum_elevation_ft_amsl": 200},
|
|
)
|
|
status_response = auth_client.patch(
|
|
f"/api/v1/drone-requests/{created['id']}/status",
|
|
json={"status": "APPROVED", "comment": "Approved below 200ft"},
|
|
)
|
|
comment_response = auth_client.post(
|
|
f"/api/v1/drone-requests/{created['id']}/comments",
|
|
json={"comment": "Call tower before launch", "email_applicant": True},
|
|
)
|
|
journal_response = auth_client.get(f"/api/v1/drone-requests/{created['id']}/journal")
|
|
|
|
assert list_response.status_code == 200
|
|
assert [request["id"] for request in list_response.json()] == [created["id"]]
|
|
assert get_response.status_code == 200
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["maximum_elevation_ft_amsl"] == 200
|
|
assert status_response.status_code == 200
|
|
assert status_response.json()["status"] == "APPROVED"
|
|
assert status_response.json()["operator_comments"] == "Approved below 200ft"
|
|
assert comment_response.status_code == 200
|
|
assert comment_response.json()["operator_comments"] == "Call tower before launch"
|
|
assert journal_response.status_code == 200
|
|
entries = [entry["entry"] for entry in journal_response.json()]
|
|
assert any("Drone request" in entry and "created" in entry for entry in entries)
|
|
assert any("Status changed from NEW to APPROVED" in entry for entry in entries)
|
|
assert any("Comment added" in entry for entry in entries)
|
|
assert len(sent_emails) == 3
|
|
|
|
|
|
def test_drone_request_not_found_and_validation_paths(auth_client, client):
|
|
invalid_response = client.post(
|
|
"/api/v1/drone-requests/public",
|
|
json=drone_payload(operator_name=" ", location_latitude=100, maximum_elevation_ft_amsl=-1),
|
|
)
|
|
|
|
assert invalid_response.status_code == 422
|
|
assert auth_client.get("/api/v1/drone-requests/404").status_code == 404
|
|
assert auth_client.patch("/api/v1/drone-requests/404", json={"operator_name": "Missing"}).status_code == 404
|
|
assert auth_client.patch("/api/v1/drone-requests/404/status", json={"status": "APPROVED"}).status_code == 404
|
|
assert (
|
|
auth_client.post(
|
|
"/api/v1/drone-requests/404/comments",
|
|
json={"comment": "Missing", "email_applicant": False},
|
|
).status_code
|
|
== 404
|
|
)
|
|
assert auth_client.get("/api/v1/drone-requests/404/journal").status_code == 404
|
|
|
|
|
|
def test_swansea_frz_runway_extensions_start_at_thresholds():
|
|
assert point_inside_swansea_frz(51.626825, -4.037672) is True
|
|
assert point_inside_swansea_frz(51.583775, -4.097928) is True
|
|
assert point_inside_swansea_frz(51.603007, -4.025604) is True
|
|
assert point_inside_swansea_frz(51.607593, -4.109996) is True
|
|
assert point_inside_swansea_frz(51.68000, -4.06780) is False
|