mirror of
https://github.com/azaion/detections.git
synced 2026-06-21 21:31:08 +00:00
Roman Meshko
55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import time
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
|
|
def _ai_config_video() -> dict:
|
|
return {}
|
|
|
|
|
|
def _ai_config_image() -> dict:
|
|
return {}
|
|
|
|
|
|
def test_ft_p08_immediate_async_response(
|
|
warm_engine, http_client, jwt_token
|
|
):
|
|
media_id = f"async-{uuid.uuid4().hex}"
|
|
body = _ai_config_image()
|
|
channel_id = str(uuid.uuid4())
|
|
headers = {"Authorization": f"Bearer {jwt_token}", "X-Channel-Id": channel_id}
|
|
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 == 202
|
|
|
|
|
|
@pytest.mark.timeout(30)
|
|
def test_ft_p09_sse_event_delivery(
|
|
warm_engine, http_client, jwt_token, sse_events_until_terminal
|
|
):
|
|
media_id = f"event-{uuid.uuid4().hex}"
|
|
channel_id = str(uuid.uuid4())
|
|
body = _ai_config_video()
|
|
auth_header = {"Authorization": f"Bearer {jwt_token}"}
|
|
post_headers = {**auth_header, "X-Channel-Id": channel_id}
|
|
r = http_client.post(f"/detect/{media_id}", json=body, headers=post_headers)
|
|
assert r.status_code == 202
|
|
collected = sse_events_until_terminal(channel_id, headers=auth_header, timeout=20)
|
|
assert collected, "no SSE events received"
|
|
|
|
|
|
def test_ft_n04_duplicate_media_id_409(
|
|
warm_engine, http_client, jwt_token
|
|
):
|
|
media_id = "dup-test"
|
|
body = _ai_config_image()
|
|
headers1 = {"Authorization": f"Bearer {jwt_token}", "X-Channel-Id": str(uuid.uuid4())}
|
|
headers2 = {"Authorization": f"Bearer {jwt_token}", "X-Channel-Id": str(uuid.uuid4())}
|
|
r1 = http_client.post(f"/detect/{media_id}", json=body, headers=headers1)
|
|
assert r1.status_code == 202
|
|
r2 = http_client.post(f"/detect/{media_id}", json=body, headers=headers2)
|
|
assert r2.status_code == 409
|