146 lines
4.3 KiB
Python
146 lines
4.3 KiB
Python
"""Integration tests for lidar_manager_web REST API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
BASE = os.environ.get("TEST_BASE_URL", "http://127.0.0.1:18080")
|
|
MISSION_ID = os.environ.get("TEST_MISSION_ID", "testmission00001")
|
|
TIMEOUT = 10
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def api():
|
|
session = requests.Session()
|
|
session.headers.update({"Content-Type": "application/json"})
|
|
deadline = time.time() + TIMEOUT
|
|
while time.time() < deadline:
|
|
try:
|
|
r = session.get(f"{BASE}/api/health", timeout=1)
|
|
if r.status_code == 200 and r.json().get("ok"):
|
|
break
|
|
except requests.RequestException:
|
|
pass
|
|
time.sleep(0.2)
|
|
else:
|
|
pytest.fail(f"Server not ready at {BASE}")
|
|
return session
|
|
|
|
|
|
def test_health(api):
|
|
r = api.get(f"{BASE}/api/health", timeout=TIMEOUT)
|
|
assert r.status_code == 200
|
|
assert r.json()["ok"] is True
|
|
|
|
|
|
def test_missions_fixture(api):
|
|
r = api.get(f"{BASE}/api/missions", timeout=TIMEOUT)
|
|
assert r.status_code == 200
|
|
ids = {m["id"] for m in r.json().get("missions", [])}
|
|
assert MISSION_ID in ids
|
|
|
|
|
|
def test_mir_v2_enqueue_and_list(api):
|
|
api.delete(f"{BASE}/api/mission_queue", timeout=TIMEOUT)
|
|
r = api.post(
|
|
f"{BASE}/api/v2.0.0/mission_queue",
|
|
json={"mission_id": MISSION_ID, "priority": 3, "robot_id": "default"},
|
|
timeout=TIMEOUT,
|
|
)
|
|
assert r.status_code == 201
|
|
body = r.json()
|
|
assert body["mission_id"] == MISSION_ID
|
|
assert body["priority"] == 3
|
|
|
|
listed = api.get(f"{BASE}/api/v2.0.0/mission_queue", timeout=TIMEOUT)
|
|
assert listed.status_code == 200
|
|
assert any(item.get("mission_id") == MISSION_ID for item in listed.json())
|
|
|
|
|
|
def test_queue_pause_continue(api):
|
|
api.delete(f"{BASE}/api/mission_queue", timeout=TIMEOUT)
|
|
api.post(
|
|
f"{BASE}/api/mission_queue",
|
|
json={"mission_id": MISSION_ID},
|
|
timeout=TIMEOUT,
|
|
)
|
|
deadline = time.time() + 5
|
|
while time.time() < deadline:
|
|
runner = api.get(f"{BASE}/api/mission_queue", timeout=TIMEOUT).json().get("runner", {})
|
|
if runner.get("state") == "running":
|
|
break
|
|
time.sleep(0.2)
|
|
else:
|
|
pytest.skip("runner did not enter running state in time")
|
|
|
|
r = api.post(f"{BASE}/api/mission_queue/pause", timeout=TIMEOUT)
|
|
assert r.status_code == 200
|
|
assert r.json().get("state") == "paused"
|
|
|
|
r = api.post(f"{BASE}/api/mission_queue/continue", timeout=TIMEOUT)
|
|
assert r.status_code == 200
|
|
assert r.json().get("state") != "paused"
|
|
|
|
|
|
def test_modbus_trigger_flow(api):
|
|
trig = api.post(
|
|
f"{BASE}/api/triggers",
|
|
json={"name": "pytest-trigger", "coil_id": 1005, "mission_id": MISSION_ID},
|
|
timeout=TIMEOUT,
|
|
)
|
|
assert trig.status_code == 201
|
|
trig_id = trig.json()["id"]
|
|
|
|
before = len(api.get(f"{BASE}/api/mission_queue", timeout=TIMEOUT).json())
|
|
fire = api.post(f"{BASE}/api/modbus/coils/1005/trigger", timeout=TIMEOUT)
|
|
assert fire.status_code == 200
|
|
|
|
after = api.get(f"{BASE}/api/mission_queue", timeout=TIMEOUT).json()
|
|
assert len(after) >= before
|
|
|
|
deleted = api.delete(f"{BASE}/api/triggers/{trig_id}", timeout=TIMEOUT)
|
|
assert deleted.status_code == 204
|
|
|
|
|
|
def test_fleet_schedule_asap(api):
|
|
r = api.post(
|
|
f"{BASE}/api/fleet/schedules",
|
|
json={
|
|
"name": "pytest-schedule",
|
|
"mission_id": MISSION_ID,
|
|
"start_mode": "asap",
|
|
"priority": 0,
|
|
"robot_id": "default",
|
|
},
|
|
timeout=TIMEOUT,
|
|
)
|
|
assert r.status_code == 201
|
|
sched_id = r.json()["id"]
|
|
|
|
deleted = api.delete(f"{BASE}/api/fleet/schedules/{sched_id}", timeout=TIMEOUT)
|
|
assert deleted.status_code == 204
|
|
|
|
|
|
def test_lidar_crud(api):
|
|
created = api.post(
|
|
f"{BASE}/api/lidars",
|
|
json={"name": "pytest-lidar", "ip": "10.99.0.1", "port": 2112},
|
|
timeout=TIMEOUT,
|
|
)
|
|
assert created.status_code == 201
|
|
lidar_id = created.json()["id"]
|
|
|
|
updated = api.put(
|
|
f"{BASE}/api/lidars/{lidar_id}",
|
|
json={"name": "pytest-lidar", "ip": "10.99.0.2", "port": 2112},
|
|
timeout=TIMEOUT,
|
|
)
|
|
assert updated.status_code == 200
|
|
|
|
deleted = api.delete(f"{BASE}/api/lidars/{lidar_id}", timeout=TIMEOUT)
|
|
assert deleted.status_code == 204
|