File size: 5,805 Bytes
51defdc | 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 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | # 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, eight
cameras + calibration (inline or the shipped preset) + ``ego_speed``.
TT_VISIBLE_DEVICES=none python -m pytest -q code/tt_meteor/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_meteor.api import METEOR # noqa: E402
from tt_meteor.server import app as server # noqa: E402
from tt_meteor.tests.stubs import StubModel # noqa: E402
from tt_meteor.ttaw.server.app import decode_request # noqa: E402
CAMS = list(METEOR.CAMERA_ORDER)
@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 _png_b64(seed: int, hw=(432, 768), zero=False) -> str:
from PIL import Image
rng = np.random.default_rng(seed)
a = np.zeros(hw + (3,), np.uint8) if zero else rng.integers(0, 256, size=hw + (3,), dtype=np.uint8)
buf = io.BytesIO()
Image.fromarray(a).save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode()
def _images(cams=CAMS, absent=()):
return [{"camera": c, "data": _png_b64(i, zero=c in absent)} for i, c in enumerate(cams)]
def _inline_calib(cams=CAMS):
K = [[400.0, 0.0, 384.0], [0.0, 400.0, 216.0], [0.0, 0.0, 1.0]]
return {"cameras": {c: {"intrinsics": K, "T_ref_from_camera": {"translation": [1.0, 0.1 * i, 1.5],
"rotation_wxyz": [0.5, -0.5, 0.5, -0.5]}}
for i, c in enumerate(cams)}}
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"] == METEOR.MODEL_NAME and info["weights"]["revision"] == "0" * 40
assert info["device"]["dispatch"] == "eth" and info["device"]["grid"] == "12x10"
assert info["autoware"]["package"] == "none (METEOR runtime: tier4/METEOR deploy/cpp)"
assert info["output"]["labels"] == list(METEOR.LABELS)
assert info["input"]["camera_order"] == CAMS
assert info["input"]["calibration_presets"] == ["synthetic_8cam"] # PandaSet preset staged, not shipped
assert info["variant"] == METEOR.DEFAULT_VARIANT
assert client.get("/v1/models").json()["data"][0]["id"] == "AutowareFoundation/meteor"
@pytest.mark.parametrize("calibration", ["inline", "preset"])
def test_predict_ok_and_equals_api(client, calibration):
cal = _inline_calib() if calibration == "inline" else {"preset": "synthetic_8cam"}
body = {"images": _images(absent=("CAM_BACK_NARROW",)), "calibration": cal, "ego_speed": 10.9,
"params": {"vehicle_threshold": 0.5}, "stream": {"id": "rig-1", "reset": False}}
r = client.post("/predict", json=body)
assert r.status_code == 200, r.text
out = r.json()
assert out["num_detections"] == 2 and out["frame_id"] == "base_link" and out["model"] == METEOR.MODEL_NAME
assert set(out["detections"][0]) >= {"label", "label_id", "score", "center", "size", "yaw", "stationary"}
assert out["trajectory"][0] == pytest.approx([10.9 * 0.5, 0.0]) and out["columns"] == ["x", "y"]
assert out["lane"]["format"] == "png" and set(out["detections_2d"]) == set(CAMS)
assert {"decode", "model_call", "total", "device"} <= set(out["timing_ms"])
state = server.app.state.ttaw
req = server.MeteorRequest(**{k: v for k, v in body.items() if k != "params"})
api = state.model(**decode_request(req, state.spec, 1 << 30)).to_dict()["detections"]
assert state.predict(req)["detections"] == api
@pytest.mark.parametrize("payload,code", [
({"images": [{"camera": "CAM_FRONT_WIDE", "data": "not base64!"}], "ego_speed": 1.0}, 400), # undecodable
({"images": _images(CAMS[:5]), "calibration": _inline_calib(), "ego_speed": 1.0}, 400), # cameras missing
({"images": _images(), "calibration": _inline_calib(CAMS[:5]), "ego_speed": 1.0}, 400), # calibration missing
({"images": _images(), "calibration": {"preset": "../etc"}, "ego_speed": 1.0}, 400), # bad preset
({"images": _images(), "calibration": {"preset": "synthetic_8cam"}}, 400), # no ego_speed
({"images": _images(), "calibration": {"preset": "synthetic_8cam"}, "ego_speed": 1.0,
"params": {"vehicle_threshold": 2.0}}, 400),
({"images": _images(), "calibration": {"preset": "synthetic_8cam"}, "ego_speed": 1.0,
"params": {"no_such_knob": 1}}, 400),
({"images": _images(), "calibration": {"preset": "synthetic_8cam"}, "ego_speed": "fast"}, 422), # not a number
({}, 400), # no images
({"imagez": []}, 422), # schema violation
])
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")
|