File size: 4,404 Bytes
fe51a63 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | # SPDX-License-Identifier: Apache-2.0
"""Host tests of the HTTP contract (no device, no weights): the real app of this bundle with a stub model.
TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_frnet/tests/test_server_host.py
"""
from __future__ import annotations
import base64
import io
import numpy as np
import pytest
fastapi = pytest.importorskip("fastapi")
from fastapi.testclient import TestClient # noqa: E402
from tt_frnet.api import FRNet # noqa: E402
from tt_frnet.server import app as server # noqa: E402
from tt_frnet.tests.stubs import StubModel # noqa: E402
@pytest.fixture
def client(monkeypatch):
monkeypatch.setenv("TT_MESH_SHAPE", "1x1")
monkeypatch.setenv("TT_MODEL_WEIGHTS_REVISION", "0" * 40)
monkeypatch.setattr(server.app.state.ttaw, "model_factory", StubModel)
with TestClient(server.app) as c:
yield c
def _npy_b64(a):
buf = io.BytesIO()
np.save(buf, a)
return base64.b64encode(buf.getvalue()).decode()
def test_health_info_models(client):
assert client.get("/health").json()["status"] == "ok"
assert client.get("/v1/health").json()["status"] == "ok"
info = client.get("/info").json()
assert info["model"] == FRNet.MODEL_NAME and info["weights"]["revision"] == "0" * 40
assert info["device"]["dispatch"] == "eth" and info["device"]["grid"] == "12x10"
assert info["autoware"]["package"] == "autoware_lidar_frnet"
assert info["output"]["labels"] == list(FRNet.LABELS)
assert info["input"]["point_fields"] == list(FRNet.POINT_FIELDS)
assert info["variant"] == FRNet.DEFAULT_VARIANT
assert client.get("/v1/models").json()["data"][0]["id"] == "AutowareFoundation/lidar_frnet"
def test_predict_ok_and_equals_api(client):
pts = np.random.default_rng(0).normal(size=(500, 4)).astype(np.float32)
r = client.post("/predict", json={"points": {"format": "npy", "data": _npy_b64(pts)},
"params": {"class_probability_threshold": 0.4},
"calibration": {"T_base_link_from_sensor": np.eye(4).tolist()}})
assert r.status_code == 200, r.text
body = r.json()
assert body["num_points"] == 500 and body["frame_id"] == "sensor" and body["model"] == FRNet.MODEL_NAME
assert body["labels"]["dtype"] == "uint8" and body["labels"]["shape"] == [500]
assert body["points"]["shape"] == [500, 4] and body["filter_keep"]["dtype"] == "bool"
assert body["class_names"] == list(FRNet.LABELS) and sum(body["class_counts"].values()) == 500
assert body["meta"]["params"]["class_probability_threshold"] == 0.4
assert body["meta"]["calibration"] == ["T_base_link_from_sensor"]
assert {"decode", "model_call", "total", "device"} <= set(body["timing_ms"])
state = server.app.state.ttaw
req = server.PredictRequest(points={"format": "npy", "data": _npy_b64(pts)})
assert state.predict(req)["labels"] == state.model(pts).to_dict()["labels"]
@pytest.mark.parametrize("payload,code", [
({"points": {"format": "npy", "data": "not base64!"}}, 400), # undecodable
({"points": {"format": "bin", "data": base64.b64encode(b"\0" * 10).decode()}}, 400), # bad row size
({"params": {"class_probability_threshold": 2.0}, "points": {"format": "list", "values": [[0, 0, 0, 0]]}}, 400),
({"params": {"score_threshold": 0.5}, "points": {"format": "list", "values": [[0, 0, 0, 0]]}}, 400),
({"params": {"no_such_knob": 1}, "points": {"format": "list", "values": [[0, 0, 0, 0]]}}, 400),
({"calibration": {"preset": "../etc"}, "points": {"format": "list", "values": [[0, 0, 0, 0]]}}, 400),
({}, 400), # no points for a LiDAR model
({"pointz": {}}, 422), # schema violation (typo)
])
def test_predict_errors(client, payload, code):
r = client.post("/predict", json=payload)
assert r.status_code == code, r.text
def test_503_while_starting(client, monkeypatch):
monkeypatch.setattr(server.app.state.ttaw, "ready", False)
assert client.post("/predict", json={}).status_code == 503
assert client.get("/health").json()["status"] == "starting"
def test_mesh_shape_parser():
p = server.parse_mesh_shape
assert p("1x1") == p("(1, 1)") == p("1,1") == p(None) == (1, 1)
with pytest.raises(RuntimeError):
p("two by two")
|