mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 21:46:31 +00:00
Oleksandr Bezdieniezhnykh
097811a67b
- Pin all deps; h11==0.16.0 (CVE-2025-43859), python-multipart>=1.3.1 (CVE-2026-28356), PyJWT==2.12.1
- Add HMAC JWT verification (require_auth FastAPI dependency, JWT_SECRET-gated)
- Fix TokenManager._refresh() to use ADMIN_API_URL instead of ANNOTATIONS_URL
- Rename POST /detect → POST /detect/image (image-only, rejects video files)
- Replace global SSE stream with per-job SSE: GET /detect/{media_id} with event replay buffer
- Apply require_auth to all 4 protected endpoints
- Fix on_annotation/on_status closure to use mutable current_id for correct post-upload event routing
- Add non-root appuser to Dockerfile and Dockerfile.gpu
- Add JWT_SECRET to e2e/docker-compose.test.yml and run-tests.sh
- Update all e2e tests and unit tests for new endpoints and HMAC token signing
- 64/64 tests pass
Made-with: Cursor
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
import time
|
|
|
|
import pytest
|
|
|
|
_DETECT_TIMEOUT = 60
|
|
|
|
|
|
def _get_health(http_client):
|
|
r = http_client.get("/health")
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def _assert_active_ai(data):
|
|
assert data["status"] == "healthy"
|
|
assert data["aiAvailability"] not in ("None", "Downloading")
|
|
|
|
|
|
@pytest.mark.cpu
|
|
class TestHealthEngineStep01PreInit:
|
|
def test_ft_p_01_pre_init_health(self, http_client):
|
|
t0 = time.monotonic()
|
|
data = _get_health(http_client)
|
|
assert time.monotonic() - t0 < 2.0
|
|
assert data["status"] == "healthy"
|
|
assert data["aiAvailability"] == "None", (
|
|
f"engine already initialized (aiAvailability={data['aiAvailability']}); "
|
|
"pre-init tests must run before any test that triggers warm_engine"
|
|
)
|
|
assert data.get("errorMessage") is None
|
|
|
|
|
|
@pytest.mark.cpu
|
|
@pytest.mark.slow
|
|
class TestHealthEngineStep02LazyInit:
|
|
def test_ft_p_14_lazy_initialization(self, http_client, image_small, auth_headers):
|
|
before = _get_health(http_client)
|
|
assert before["aiAvailability"] == "None", (
|
|
f"engine already initialized (aiAvailability={before['aiAvailability']}); "
|
|
"lazy-init test must run before any test that triggers warm_engine"
|
|
)
|
|
files = {"file": ("lazy.jpg", image_small, "image/jpeg")}
|
|
r = http_client.post("/detect/image", files=files, headers=auth_headers, timeout=_DETECT_TIMEOUT)
|
|
r.raise_for_status()
|
|
body = r.json()
|
|
assert isinstance(body, list)
|
|
after = _get_health(http_client)
|
|
_assert_active_ai(after)
|
|
|
|
|
|
@pytest.mark.cpu
|
|
@pytest.mark.slow
|
|
class TestHealthEngineStep03Warmed:
|
|
@pytest.fixture(autouse=True)
|
|
def _warm(self, warm_engine):
|
|
pass
|
|
|
|
def test_ft_p_02_post_init_health(self, http_client):
|
|
data = _get_health(http_client)
|
|
_assert_active_ai(data)
|
|
assert data.get("errorMessage") is None
|
|
|
|
def test_ft_p_15_onnx_cpu_detect(self, http_client, image_small, auth_headers):
|
|
files = {"file": ("onnx.jpg", image_small, "image/jpeg")}
|
|
r = http_client.post("/detect/image", files=files, headers=auth_headers, timeout=_DETECT_TIMEOUT)
|
|
r.raise_for_status()
|
|
body = r.json()
|
|
assert isinstance(body, list)
|
|
if body:
|
|
d = body[0]
|
|
for k in (
|
|
"centerX",
|
|
"centerY",
|
|
"width",
|
|
"height",
|
|
"classNum",
|
|
"label",
|
|
"confidence",
|
|
):
|
|
assert k in d
|