weijuesen
126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
import base64
|
|
import os
|
|
|
|
os.environ.setdefault("INTERNAL_API_KEY", "test-internal-key")
|
|
os.environ.setdefault("ALLOWED_STREAM_HOSTS", "localhost,127.0.0.1,100.83.103.1")
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import main as main_module
|
|
from app.main import app
|
|
from app.detector import MockDetector
|
|
|
|
# 1x1 透明 PNG
|
|
TINY_PNG = base64.b64decode(
|
|
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
|
|
)
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_health():
|
|
r = client.get("/health")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "ok"
|
|
assert body["model"] in ("mock", "onnx")
|
|
assert body["modelVersion"]
|
|
assert body["isMock"] is True
|
|
|
|
|
|
def test_detect_ok():
|
|
r = client.post("/detect", files={"file": ("a.png", TINY_PNG, "image/png")})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["model"] == "mock"
|
|
assert body["modelVersion"]
|
|
assert body["isMock"] is True
|
|
assert body["status"] == "healthy"
|
|
assert body["abnormalProbability"] == 0
|
|
assert len(body["detections"]) >= 1
|
|
d = body["detections"][0]
|
|
assert d["class"] in ("healthy", "sick")
|
|
assert 0 <= d["confidence"] <= 1
|
|
assert d["bbox"]["w"] > 0
|
|
|
|
|
|
def test_detect_uses_abnormal_class_confidence(monkeypatch):
|
|
monkeypatch.setattr(main_module, "detector", MockDetector(class_name="sick", confidence=0.92))
|
|
r = client.post("/detect", files={"file": ("a.png", TINY_PNG, "image/png")})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "abnormal"
|
|
assert body["abnormalProbability"] == 0.92
|
|
|
|
|
|
def test_detect_empty_result_is_unknown(monkeypatch):
|
|
class EmptyDetector:
|
|
def detect(self, image_bytes):
|
|
return []
|
|
|
|
monkeypatch.setattr(main_module, "detector", EmptyDetector())
|
|
r = client.post("/detect", files={"file": ("a.png", TINY_PNG, "image/png")})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "unknown"
|
|
assert body["abnormalProbability"] == 0
|
|
|
|
|
|
def test_detect_empty_file_rejected():
|
|
r = client.post("/detect", files={"file": ("a.png", b"", "image/png")})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_detect_invalid_image_rejected():
|
|
r = client.post("/detect", files={"file": ("a.png", b"junk", "image/png")})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_stream_detect_is_disabled():
|
|
r = client.post("/stream-detect", json={})
|
|
assert r.status_code == 410
|
|
|
|
|
|
def test_internal_stream_tasks_requires_internal_key():
|
|
r = client.post("/internal/stream-tasks", json={})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_internal_stream_tasks_accepts_internal_stream_ref():
|
|
r = client.post(
|
|
"/internal/stream-tasks",
|
|
headers={"X-Internal-Key": "test-internal-key"},
|
|
json={
|
|
"taskId": "task-1",
|
|
"streamRef": {
|
|
"url": "rtsp://127.0.0.1:8554/live/1",
|
|
"maxFrames": 1,
|
|
},
|
|
},
|
|
)
|
|
assert r.status_code == 202
|
|
assert r.json()["taskId"] == "task-1"
|
|
|
|
|
|
def test_internal_stream_tasks_rejects_metadata_url():
|
|
r = client.post(
|
|
"/internal/stream-tasks",
|
|
headers={"X-Internal-Key": "test-internal-key"},
|
|
json={
|
|
"taskId": "task-2",
|
|
"streamRef": {
|
|
"url": "http://169.254.169.254/latest/meta-data",
|
|
"maxFrames": 1,
|
|
},
|
|
},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_metrics_shape():
|
|
r = client.get("/metrics")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
for key in ("model", "uptimeSeconds", "requests", "avgLatencyMs", "gpu"):
|
|
assert key in body
|