| """ |
| drone_env/tests/test_api.py |
| # Integration tests for the Drone Delivery Env FastAPI server. |
| endpoints using TestClient. |
| """ |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent)) |
|
|
| import pytest |
| from fastapi.testclient import TestClient |
| from drone_env.server.app import app |
|
|
| client = TestClient(app) |
|
|
|
|
| def test_health(): |
| r = client.get("/health") |
| assert r.status_code == 200 |
| assert r.json()["status"] == "ok" |
|
|
|
|
| def test_root(): |
| """Test dashboard serving or JSON fallback.""" |
| r = client.get("/") |
| assert r.status_code == 200 |
| ctype = r.headers.get("Content-Type", "") |
| if "text/html" in ctype: |
| assert "Drone Delivery" in r.text or "<!DOCTYPE html>" in r.text |
| elif "application/json" in ctype: |
| data = r.json() |
| assert "name" in data |
| else: |
| |
| assert len(r.text) > 0 |
|
|
|
|
| def test_tasks_list(): |
| r = client.get("/tasks") |
| assert r.status_code == 200 |
| tasks = r.json()["tasks"] |
| names = [t["name"] for t in tasks] |
| assert "easy_delivery" in names |
| assert "medium_delivery" in names |
| assert "hard_delivery" in names |
|
|
|
|
| def test_reset_default(): |
| r = client.post("/reset") |
| assert r.status_code == 200 |
| obs = r.json() |
| assert "grid" in obs |
| assert obs["done"] is False |
| assert obs["step_count"] == 0 |
| assert 0 < obs["grid_width"] <= 20 |
| assert 0 < obs["grid_height"] <= 20 |
|
|
|
|
| def test_reset_with_task(): |
| for task in ["easy_delivery", "medium_delivery", "hard_delivery"]: |
| r = client.post("/reset", json={"task_name": task}) |
| assert r.status_code == 200 |
| obs = r.json() |
| assert obs["done"] is False |
|
|
|
|
| def test_step_returns_observation(): |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| for direction in ["UP", "DOWN", "LEFT", "RIGHT", "WAIT"]: |
| r = client.post("/step", json={"direction": direction}) |
| assert r.status_code == 200 |
| obs = r.json() |
| assert "reward_last" in obs |
| assert "battery" in obs |
| assert "done" in obs |
|
|
|
|
| def test_state_endpoint(): |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| r = client.get("/state") |
| assert r.status_code == 200 |
| state = r.json() |
| assert "episode_id" in state |
| assert "task_name" in state |
| assert state["task_name"] == "easy_delivery" |
|
|
|
|
| def test_grade_known_task(): |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| r = client.get("/grade/easy_delivery") |
| assert r.status_code == 200 |
| score = r.json()["score"] |
| assert 0.0 <= score <= 1.0 |
|
|
|
|
| def test_grade_unknown_task(): |
| r = client.get("/grade/unknown_task") |
| assert r.status_code == 404 |
|
|
|
|
| def test_path_history_after_steps(): |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| for _ in range(5): |
| client.post("/step", json={"direction": "RIGHT"}) |
| r = client.get("/path_history") |
| assert r.status_code == 200 |
| history = r.json()["path_history"] |
| assert len(history) == 5 |
|
|
|
|
| def test_reward_in_range(): |
| """All per-step rewards must stay reasonable (no NaN, no extreme values).""" |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| for direction in ["UP","RIGHT","DOWN","LEFT","WAIT","UP","RIGHT"]: |
| r = client.post("/step", json={"direction": direction}) |
| obs = r.json() |
| reward = obs["reward_last"] |
| assert -2.0 <= reward <= 2.0, f"Reward out of range: {reward}" |
|
|
|
|
| def test_battery_decreases(): |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| r1 = client.post("/step", json={"direction": "WAIT"}) |
| bat1 = r1.json()["battery"] |
| r2 = client.post("/step", json={"direction": "WAIT"}) |
| bat2 = r2.json()["battery"] |
| assert bat2 <= bat1 |
|
|
|
|
| def test_analyse_endpoint(): |
| r = client.get("/analyse/easy_delivery") |
| assert r.status_code == 200 |
| data = r.json() |
| |
| assert "status" in data |
|
|
|
|
| def test_analyse_unknown_task(): |
| r = client.get("/analyse/nonexistent") |
| assert r.status_code == 404 |
|
|
|
|
| def test_full_episode_doesnt_crash(): |
| """Run a full episode of random steps — environment must not raise.""" |
| import random |
| directions = ["UP","DOWN","LEFT","RIGHT","WAIT"] |
| client.post("/reset", json={"task_name": "easy_delivery"}) |
| for _ in range(160): |
| r = client.post("/step", json={"direction": random.choice(directions)}) |
| assert r.status_code == 200 |
| if r.json().get("done"): |
| break |
| |
| r = client.post("/reset", json={"task_name": "easy_delivery"}) |
| assert r.status_code == 200 |
| assert r.json()["done"] is False |
|
|
