| |
| """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_transfusion/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 |
|
|
| from tt_transfusion.api import TransFusion |
| from tt_transfusion.server import app as server |
| from tt_transfusion.tests.stubs import StubModel |
|
|
|
|
| @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"] == TransFusion.MODEL_NAME and info["weights"]["revision"] == "0" * 40 |
| assert info["device"]["dispatch"] == "eth" and info["device"]["grid"] == "12x10" |
| assert info["autoware"]["package"] == "autoware_lidar_transfusion" |
| assert info["output"]["labels"] == list(TransFusion.LABELS) |
| assert info["input"]["point_fields"] == list(TransFusion.POINT_FIELDS) |
| assert info["variant"] == TransFusion.DEFAULT_VARIANT |
| assert client.get("/v1/models").json()["data"][0]["id"] == "AutowareFoundation/lidar_transfusion" |
|
|
|
|
| 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": {"score_threshold": 0.4}}) |
| assert r.status_code == 200, r.text |
| body = r.json() |
| assert body["num_detections"] == 3 and body["frame_id"] == "base_link" and body["model"] == TransFusion.MODEL_NAME |
| assert set(body["detections"][0]) >= {"label", "label_id", "score", "center", "size", "yaw"} |
| 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)["detections"] == state.model(pts).to_dict()["detections"] |
|
|
|
|
| @pytest.mark.parametrize("payload,code", [ |
| ({"points": {"format": "npy", "data": "not base64!"}}, 400), |
| ({"points": {"format": "bin", "data": base64.b64encode(b"\0" * 10).decode()}}, 400), |
| ({"params": {"score_threshold": 2.0}, "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), |
| ({"pointz": {}}, 422), |
| ]) |
| 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") |
|
|
