code/tt_frnet/tests/test_server_host.py

# 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")