mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 22:56:31 +00:00
Oleksandr Bezdieniezhnykh
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
import json
|
|
import threading
|
|
import time
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
|
|
def _ai_config_video(mock_loader_url: str) -> dict:
|
|
base = mock_loader_url.rstrip("/")
|
|
return {
|
|
"probability_threshold": 0.25,
|
|
"tracking_intersection_threshold": 0.6,
|
|
"altitude": 400,
|
|
"focal_length": 24,
|
|
"sensor_width": 23.5,
|
|
"paths": [f"{base}/load/video_short01.mp4"],
|
|
"frame_period_recognition": 4,
|
|
"frame_recognition_seconds": 2,
|
|
}
|
|
|
|
|
|
def test_ft_p08_immediate_async_response(
|
|
warm_engine, http_client, jwt_token, mock_loader_url
|
|
):
|
|
media_id = f"async-{uuid.uuid4().hex}"
|
|
body = _ai_config_video(mock_loader_url)
|
|
headers = {"Authorization": f"Bearer {jwt_token}"}
|
|
t0 = time.monotonic()
|
|
r = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
|
|
elapsed = time.monotonic() - t0
|
|
assert elapsed < 2.0
|
|
assert r.status_code == 200
|
|
assert r.json() == {"status": "started", "mediaId": media_id}
|
|
|
|
|
|
@pytest.mark.slow
|
|
@pytest.mark.timeout(120)
|
|
def test_ft_p09_sse_event_delivery(
|
|
warm_engine, http_client, jwt_token, mock_loader_url, sse_client_factory
|
|
):
|
|
media_id = f"sse-{uuid.uuid4().hex}"
|
|
body = _ai_config_video(mock_loader_url)
|
|
headers = {"Authorization": f"Bearer {jwt_token}"}
|
|
collected: list[dict] = []
|
|
thread_exc: list[BaseException] = []
|
|
done = threading.Event()
|
|
|
|
def _listen():
|
|
try:
|
|
with sse_client_factory() as sse:
|
|
time.sleep(0.3)
|
|
for event in sse.events():
|
|
if not event.data or not str(event.data).strip():
|
|
continue
|
|
data = json.loads(event.data)
|
|
if data.get("mediaId") != media_id:
|
|
continue
|
|
collected.append(data)
|
|
if (
|
|
data.get("mediaStatus") == "AIProcessed"
|
|
and data.get("mediaPercent") == 100
|
|
):
|
|
break
|
|
except BaseException as e:
|
|
thread_exc.append(e)
|
|
finally:
|
|
done.set()
|
|
|
|
th = threading.Thread(target=_listen, daemon=True)
|
|
th.start()
|
|
time.sleep(0.5)
|
|
r = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
|
|
assert r.status_code == 200
|
|
ok = done.wait(timeout=120)
|
|
assert ok, "SSE listener did not finish within 120s"
|
|
th.join(timeout=5)
|
|
assert not thread_exc, thread_exc
|
|
assert any(e.get("mediaStatus") == "AIProcessing" for e in collected)
|
|
final = collected[-1]
|
|
assert final.get("mediaStatus") == "AIProcessed"
|
|
assert final.get("mediaPercent") == 100
|
|
|
|
|
|
def test_ft_n04_duplicate_media_id_409(
|
|
warm_engine, http_client, jwt_token, mock_loader_url
|
|
):
|
|
media_id = "dup-test"
|
|
body = _ai_config_video(mock_loader_url)
|
|
headers = {"Authorization": f"Bearer {jwt_token}"}
|
|
r1 = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
|
|
assert r1.status_code == 200
|
|
r2 = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
|
|
assert r2.status_code == 409
|