mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 11:56:31 +00:00
Oleksandr Bezdieniezhnykh
8baa96978b
- Updated the detection image endpoint to require a channel ID for event streaming. - Introduced a new endpoint for streaming detection events, allowing clients to receive real-time updates. - Enhanced the internal buffering mechanism for detection events to manage multiple channels. - Refactored the inference module to support the new event handling structure. Made-with: Cursor
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
_DETECT_TIMEOUT = 60
|
|
|
|
|
|
def _assert_health_200(http_client):
|
|
r = http_client.get("/health")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["status"] == "healthy"
|
|
assert data.get("errorMessage") is None
|
|
|
|
|
|
@pytest.mark.cpu
|
|
def test_ft_n_01_empty_image_returns_400(http_client, empty_image, auth_headers):
|
|
files = {"file": ("empty.jpg", empty_image, "image/jpeg")}
|
|
r = http_client.post("/detect/image", files=files, headers=auth_headers, timeout=30)
|
|
assert r.status_code == 400
|
|
body = r.json()
|
|
assert "detail" in body
|
|
assert body["detail"] == "Image is empty"
|
|
_assert_health_200(http_client)
|
|
|
|
|
|
@pytest.mark.cpu
|
|
def test_ft_n_02_corrupt_image_returns_400_or_422(http_client, corrupt_image, auth_headers):
|
|
files = {"file": ("corrupt.jpg", corrupt_image, "image/jpeg")}
|
|
r = http_client.post("/detect/image", files=files, headers=auth_headers, timeout=30)
|
|
assert r.status_code in (400, 422)
|
|
body = r.json()
|
|
assert "detail" in body
|
|
_assert_health_200(http_client)
|
|
|
|
|
|
@pytest.mark.cpu
|
|
def test_ft_n_03_loader_error_mode_detect_does_not_500(
|
|
http_client, mock_loader_url, image_small, auth_headers
|
|
):
|
|
cfg = requests.post(
|
|
f"{mock_loader_url}/mock/config", json={"mode": "error"}, timeout=10
|
|
)
|
|
cfg.raise_for_status()
|
|
channel_id = str(uuid.uuid4())
|
|
headers = {**auth_headers, "X-Channel-Id": channel_id}
|
|
files = {"file": ("small.jpg", image_small, "image/jpeg")}
|
|
r = http_client.post("/detect/image", files=files, headers=headers, timeout=_DETECT_TIMEOUT)
|
|
assert r.status_code != 500
|